kosich/rxjs-proxify

Turns a Stream of Objects into an Object of Streams

{ 👓 }
_{Turn a Stream of Objects into an Object of Streams}

Access values inside RxJS Observables as if they were directly available on the stream!

stream.pipe(pluck('msg')).subscribe(…);
// turn ↑ into ↓
stream.msg.subscribe(…);

With good TypeScript support! 😲 Roughly speaking:

proxify( Observable<{ msg: string }> ) ≈ Observable<{ msg: string }> & { msg: Observable<string> }

But recursively. So stream.msg is a Proxy itself, allowing you to stream.msg.length.subscribe(…)!

Proxify lets you access Observable API as well as pluck props and call methods at any depth of an Observable, Subject, or BehaviorSubject! See the API and Examples sections to learn more.

📦 Install

npm i rxjs-proxify

or try it online!

🛠 API

There are two methods available to you: proxify and statify

Proxify

proxify(stream) will wrap your Observable, Subject or BehaviorSubject in a Proxy:

Observable Proxy
subscribe at any depth

const observable = proxify( of({ p: '🐑' }) );
observable.subscribe(console.log); // > { p: 🐑 }
observable.p.subscribe(console.log); // > 🐑

Subject Proxy
subscribe at any depth, push at the root

const subject = proxify(new Subject<{ p: string }>());
subject.subscribe(console.log);
subject.p.subscribe(console.log);
subject.next({ p: '🐥' }); // > { p: 🐥 } // > 🐥

BehaviorSubject Proxy
subscribe at any depth, push at any depth, synchronously read the current state

const behavior = proxify(new BehaviorSubject({ p: '🐖' }));
behavior.p.subscribe(console.log); // > 🐖
behavior.p.next('🐇'); // > 🐇
console.log(behavior.p.value) // > 🐇

Statify

statify(value) will put the value in a BehaviorSubject Proxy and add a distinctUntilChanged operator on each property access.

State Proxy
subscribe to distinct updates at any depth, push at any depth, synchronously read the current state

// create a state
const state = statify({ a: '🐰', z: '🏡' });

// listen to & log root state changes
state.subscribe(console.log); //> { a:🐰 z:🏡 }

// update particular substate
state.a.next('🐇'); //> { a:🐇 z:🏡 }

// read current values
console.log(state.z.value + state.a.value); //> 🏡🐇

// update root state, still logging
state.next({ a: '🐇', z: '☁️' }) //> { a:🐇 z:☁️ }

// and then…
state.z.next('🌙');   //> { a:🐇  z:🌙 }
state.a.next('🐇👀'); //> { a:🐇👀 z:🌙 }
state.z.next('🛸')    //> { a:🐇👀 z:🛸 }
state.a.next('💨');   //> { a:💨  z:🛸 }

See Examples section for more details.

📖 Examples

Basic

import { proxify } from "rxjs-proxify";
import { of } from "rxjs";

const o = of({ msg: 'Hello' }, { msg: 'World' });
const p = proxify(o);
p.msg.subscribe(console.log);

// equivalent to
// o.pipe(pluck('msg')).subscribe(console.log);

With JS destructuring

Convenient stream props splitting

import { proxify } from "rxjs-proxify";
import { of } from "rxjs";

const o = of({ msg: 'Hello', status: 'ok'  }, { msg: 'World', status: 'ok' });
const { msg, status } = proxify(o);
msg.subscribe(console.log);
status.subscribe(console.log);

// equivalent to
// const msg = o.pipe(pluck('msg'));
// const status = o.pipe(pluck('status'));
// msg.subscribe(console.log);
// status.subscribe(console.log);

⚠️ WARNING: as shown in "equivalent" comment, this operation creates several Observables from the source Observable. Which means that if your source is cold — then you might get undesired subscriptions. This is a well-known nuance of working with Observables. To avoid this, you can use a multicasting operator on source before applying proxify, e.g. with shareReplay:

const { msg, status } = proxify(o.pipe(shareReplay(1)));

With pipe

Concatenate all messages using pipe with scan operator:

import { proxify } from "rxjs-proxify";
import { of } from "rxjs";
import { scan } from "rxjs/operators";

const o = of({ msg: 'Hello' }, { msg: 'World' });
const p = proxify(o);
p.msg.pipe(scan((a,c)=> a + c)).subscribe(console.log);

// equivalent to
// o.pipe(pluck('msg'), scan((a,c)=> a + c)).subscribe(console.log);

Calling methods

Pick a method and call it:

import { proxify } from "rxjs-proxify";
import { of } from "rxjs";

const o = of({ msg: () => 'Hello' }, { msg: () => 'World' });
const p = proxify(o);
p.msg().subscribe(console.log);

// equivalent to
// o.pipe(map(x => x?.map())).subscribe(console.log);

Accessing array values

Proxify is recursive, so you can keep chaining props or indices

import { proxify } from "rxjs-proxify";
import { of } from "rxjs";

const o = of({ msg: () => ['Hello'] }, { msg: () => ['World'] });
const p = proxify(o);
p.msg()[0].subscribe(console.log);

// equivalent to
// o.pipe(map(x => x?.map()), pluck(0)).subscribe(console.log);

🤝 Want to contribute to this project?

That will be awesome!

Please create an issue before submiting a PR — we'll be able to discuss it first!

Thanks!

Enjoy 🙂

rxjs-proxify's Issues

Add TS specs

Add automatic type checker via one of these:

• https://github.com/microsoft/dtslint
• https://github.com/SamVerschueren/tsd
• …

Property should return same ref value

Accessing proxified property should return the same reference === property each time

let o = of({ a: 42 });
let p = proxify(o);
assert(p.a === p.a);

Reserved keys in observable type

I figured out how to exclude a set of key names from every level of an object in typescript using a recursive type definition. This would be useful, for example, to restrict someone from calling proxify with an Observable of a type that overrides subscribe.

// deep-omit.d.ts
export type DeepOmit<Reserved extends string> =
  | Boolean
  | Number
  | String
  | Date
  | Function
  | Array<DeepOmit<Reserved>>
  | ({ [key: string]: DeepOmit<Reserved> } & Partial<
      { [key in Reserved]: never }
    >);

Usage:

type S = DeepOmit<keyof Observable>;
// or
type S = DeepOmit<'subscribe' | 'pipe'>;

const test: S = 'a'; // OK
const test: S = false // OK
const test: S = { subscribe: false }; // Type 'boolean' is not assignable to type 'undefined'.
// Deep matching
const test: S = { a: { pipe: 'hello' } }; // Type 'string' is not assignable to type 'undefined'.
const test: S = { a: { piper: 'hi' } }; // OK

But I don't really know how to integrate with the existing typings file you have.

Handling Observable<never>

Proxified Observable<never> becomes never instead of IProxiedObservable<never>

let o: Observable<never> = of();
let p = proxify(o);
p.pipe(map(x => !x)).subscribe();

proxify(stream) with distinct values

I have a readable Observable State, not created by statify, used with proxify, and I would like to have the ability to have a distinctUntilChanged on each property like statify does.

My use case is making a redux state Observable (readonly at first) through this lib.

Proxy on undefined is interpreted as callable

Seems like Observable<undefined> type is interpreted as callable (ICallableProxy), which provides Function.prototype props:

let o = of<{ a: undefined }>({ a: undefined });
let p = proxify(o);
p.a.call(…)

Extensibility

It'd be good to have ability to extend some levels of proxify, e.g.:

Rx Simple State example

// create a state
const state = createState({ count: 0, timer: 100 });

// listen to state changes
state.count.subscribe(c => console.log('C:', c)); // > C:0

// write to the state
state.count.next(1); // > C:1

// reset the state
state.next({ timer: 2, count: 2 }) // > C:2

state.count += 1; // > C:3
state.count.next(3); // ignored: same value

// read current values:
console.log(state.timer + state.count); // > 2 + 3 = 5

OR to be able to add rxjs-autorun functions, e.g.:

// pseudocode
// state.js
import { $, _ } from 'rxjs-autorun';

export const State = v => proxify(
  new BehaviorSubject(v), // base
  { $, _ } // extender
);

// index.js
import { run } from 'rxjs-autorun';
import { State } from './state';

const state = State(0);
run(() => state.$ + ' 🐑'); // > 0 🐑
state.next(1); // > 1 🐑

pipe in boolean property causes errors

hi there,

i can't use pipe method in boolean property. i try to fix it, but for me it's too difficult. please fix this. thanks ^^

const state = proxify(new BehaviorSubject<{ a: number, b: string, c: boolean }>({ a: 0, b: '999', c: true }));
state.a.pipe(map(a => a));
state.b.pipe(map(a => a));

state.c.pipe(map(a => a));

proxifying Subjects

Subjects should be proxifiable as well, letting us do:

const o = new Subject<{ value: number }>();
const p = proxify(o);

p.subscribe(console.log);
p.value.subscribe(console.log);
p.next({ value: 0 });
p.value.next(1);

Same should apply to BehaviorSubject and ReplaySubject

`undefined` value in a CombineLatest

Hi there,
First and foremost I want to say thanks for this lib, it looks very promising and I love the idea!

Second, we've started using it in one of our projects and while the statify(...), the subscribe or the .next(...) worked well, we encountered an issue when using a proxy in a combineLatest. Here's an example:

With a proxy in a combineLatest, it produces an undefined:

const first$ = statify('nothing');
const second$ = of('foo');

const combined$ = combineLatest([first$, second$]).pipe(tap(val => console.log('Tap: ', val)));

first$.subscribe(machin => console.log('Behavior sub: ', machin));
combined$.subscribe(subed => console.log('Combined sub: ', subed));

//> Behavior sub:  nothing
//> Tap:  (2) [undefined, "foo"]
//> Combined sub:  (2) [undefined, "foo"]

first$.next('something');

//> Behavior sub:  something
//> Tap:  (2) [undefined, "foo"]
//> Combined sub:  (2) [undefined, "foo"]

Where I would have expected it to work like with a normal BehaviorSubject:

const first$ = new BehaviorSubject('nothing');
const second$ = of('foo');

const combined$ = combineLatest([first$, second$]).pipe(tap(val => console.log('Tap: ', val)));

first$.subscribe(machin => console.log('Behavior sub: ', machin));
combined$.subscribe(subed => console.log('Combined sub: ', subed));

//> Behavior sub:  nothing
//> Tap:  (2) ["nothing", "foo"]
//> Combined sub:  (2) ["nothing", "foo"]

first$.next('something');

//> Behavior sub:  something
//> Tap:  (2) ["something", "foo"]
//> Combined sub:  (2) ["something", "foo"]

Am I missing something, or is it something I misunderstood? I would love to have your insight about that!
Once again, thank you for this, and keep up the good work 👍