I/O relation of data when using RxJS observables

Viewed 59

I want to hide a part of the data that traverses through observable operators so integrety is maintained.

For example if I have a data object like { private, public } and I want to do:

  1. pluck the public property and pass it to a custom stream custom$
  2. after custom$ has completed I want to restore the relation, so that I have an object { private, data }, where data is the result of the custom$ stream.

I don't want to use concatMap, because the custom$ could potentially block incoming data and make the stream "slow".

I am currently using the following, but it seems very cumbersome and feels like I am doing something wrong:

console.clear();

const input$ = new Rx.Subject();
const custom = data$ => data$.switchMap(
  ({ value, delay }) => Rx.Observable.of(`data ${value}`).delay(delay)
);

const out$ = input$.mergeMap(
  ({ secret, data }) =>
    Rx.Observable.zip(
      Rx.Observable.of(data).let(data$ => custom(data$)),
      Rx.Observable.of(secret)
    ));


out$.subscribe(([data, secret]) => console.log(data, '=', secret));

input$.next({ secret: 1, data: { value: 1, delay: 4000 }});
input$.next({ secret: 2, data: { value: 2, delay: 1000 }});
input$.next({ secret: 3, data: { value: 3, delay: 3000 }});
<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

Is there an easier/more sane way to do this?

2 Answers
Related