Splitting observable payload into multiple observables

user_162056 0
Viewed 194

I have an Apollo query that I pipe through a map operator to emit objects. For example, {name: "doron", color: "red"}.

query$ = this.apollo.watchQuery(query)

I want to split it into 2 observables - name$ and color$ so that name$ emits the name values and color$ emits the color values.

What we do today is this:

name$ = new Subject();
color$ = new Subject();
query$.valueChanges.subscribe(p => {
  name$.next(p.name);
  color$.next(p.color);
})

This is not ideal because we need to manually unsubscribe. Also, it subscribes to the query before any of the "child" observables is being subscribed to.

How can I do this without subscribing to the query?

apollo-client rxjs angular
3 Answers

While both Fan and Dallows have good answers, I think a mix of both + a slight tweak would be nice:

query$ = query.pipe(
  // You won't subscribe to `query` twice (or more).
  shareReplay({

    // If one of the observables (`query$`, `name$` or `color$`) are subscribed to later on,
    // you'll get instantly the last value that was emitted 
    // (compared to `share` which wouldn't give you the last value if you subscribe too late).
    bufferSize: 1,

    // Whenever the number of subscribers to the query falls down to 0, it'll automatically unsubscribe from that observable as well.
    refCount: true
  })
);

name$ = query$.valueChanges.pipe(
  map(({name}) => name)
);
color$ = query$.valueChanges.pipe(
  map(({color}) => color)
);

(The explanation below are also written as a comment in the code above)

Using shareReplay, you won't subscribe to query twice (or more).

Thanks to bufferSize: 1, if one of the observables (query$, name$ or color$) are subscribed to later on, you'll get instantly the last value that was emitted (compared to share which wouldn't give you the last value if you subscribe too late).

Thanks to refCount: true, whenever the number of subscribers to the query falls down to 0, it'll automatically unsubscribe from that observable as well.

user_23985930

Instead of creating partial observables as Subjects and pushing to them, create them directly by mapping from the query.

name$ = query.valueChanges.pipe(map(({name}) => name));
color$ = query.valueChanges.pipe(map(({color}) => color));

You can then easily unsubscribe them separately or use in async pipe. I created a little showcase, to illustrate how it works: Demo

user_33596350

You can share the OQ observable and use pluck to get your value

const oq = query$.valueChanges.pipe(share());

let name$ = oq.pipe(pluck('name'));
let color$ = oq.pipe(pluck('color'))
user_86183500
Related