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 🙂