RXJS: Can I interleave multiple streams based on order of entries to a single output?

user_1236657 0
Viewed 45

This is purely an academic exercise to understand operators. How can I use RXJS to interleave incoming streams

e.g. go from this:

// RxJS v6+
import { of } from 'rxjs';
//emits any number of provided values in sequence
const source = of(1, 2, 3, 4, 5);
//output: 1,2,3,4,5
const subscribe = source.subscribe(val => console.log(val));

to this:

import { of } from 'rxjs';

const source1 = of(1,3,5,7,9);
const source2 = of(2,4,6,8,10);

// a simple merge will just append source1 and source2
// how do I obtain an output = 1,2,3,4,5,6,7,8,9,10
const subscribe = merge(source1, source2).subscribe(val => console.log(val))

Constraints: This is a general question so I dont want to use predicates based on the odd/even separation of this specific example, but to create an output based purely on order of incoming elements i.e. take the first from each stream, then the second, etc.

Is that possible?

rxjs javascript
2 Answers

The standard way of doing this is indeed the merge function, which can receive any number of observables as arguments, and emits all of their items in sequece regardless of the order of the observables in the function parameters. But as you pointed out, it appended the results, the reason for this is not an issue with merge, but actually that of emmited the values synchronously, so they all ran in order in the same javascript loop. You can change that by using an observable that emits values assynchronously, or create an observable with some of the available async Schedulers

Assynchronous observable:

I created this asyncOffunction as a mockup to a Observable with values emmited over time, like a websocket client or user interations.

import { merge, Observable, EMPTY } from 'rxjs';
import { toArray } from 'rxjs/operators';

const asyncOf = <T = any>(args: T[], interval = 500): Observable<T> => {
  let i = 0;
  let intervalRef;

  if(!args?.length) {
    return EMPTY;
  }
 
  return new Observable(observer => {
    intervalRef = setInterval(() => {
      if(args[i]) {
        observer.next(args[i])
      }
      ++i;

      if(!args[i]) {
        observer.complete()
        if(intervalRef) {
           clearInterval(intervalRef)
        }
      }
    }, interval)
  })
}

const source1 = asyncOf([1, 3, 5, 7, 9]);
const source2 = asyncOf([2, 4, 6, 8, 10]);

//output: 1,2,3,4,5,6,7,8,9,10
const subscribe = merge(source1, source2)
  //.pipe(toArray()) //collect all emmited values and emmit an array when the observable completes
  .subscribe((val) => console.log(val));

In this case the merge works correctly, because both observables are running asynchronously.

Async schedulers

Another option is using one of the async schedulers (or create your own) to describe when each item should be emmited. This is closer to your example, and a somewhat deeper dive into rxjs:

import { merge, scheduled, asyncScheduler } from 'rxjs';
import { toArray } from 'rxjs/operators';

const source1 = scheduled([1, 3, 5, 7, 9], asyncScheduler);
const source2 = scheduled([2, 4, 6, 8, 10], asyncScheduler);

//output: 1,2,3,4,5,6,7,8,9,10
const subscribe = merge(source1, source2)
  //.pipe(toArray()) //collect all emmited values and emmit an array when the observable completes
  .subscribe((val) => console.log(val));

Scheduled comes on RXJS 6.5+ and deprecates the use of scheduler function in other functions like of,

user_109229480
user_71942680
Related