rxjs multiple observable switch

There are probably more elegant solutions to this, but I normally reach for scan and create a little state machine:

import { merge, Subject } from 'rxjs';
import { scan, filter, map, take } from 'rxjs/operators';

const left = new Subject<number>();
const right = new Subject<number>();
const example = merge(
  left.pipe(map(val => ({ tag: 'left', val }))),
  right.pipe(map(val => ({ tag: 'right', val })))
).pipe(
  scan<{ tag: string; val: number }, { previous: string; emit?: number }>(
    (acc, { tag, val }) => {
      if (acc.previous !== tag) {
        return { previous: tag, emit: val };
      }
      return { ...acc, emit: null };
    },
    {
      previous: 'right',
      emit: null
    }
  ),
  filter(x => x.emit !== null),
  map(x => x.emit)
);
console.clear();
const subscribe = example
  .pipe(take(10))
  .subscribe(val => console.log('output', val));

right.next(1);
left.next(2);
right.next(3);
left.next(4);
left.next(5);
right.next(6);

Seems like there are quite a few ways to solve this. Here's another with expand.

import { interval } from 'rxjs';
import { expand, mapTo, take } from 'rxjs/operators';

const left = interval(1000).pipe(mapTo('left'));
const right = interval(5000).pipe(mapTo('right'));
const example = left.pipe(
  take(1),
  expand(x => (x === 'left' ? right : left).pipe(take(1)))
);
const subscribe = example.subscribe(val => console.log(val));

I'd be interested if someone has a way to do it with one of the buffer* operators

I created you a custom operator (toggleEmit) that takes N observables and toggles between them.

const result$ = toggleEmit(source1$, source2$);

Implemented features:

• N observables can be given to the toggleEmit operator
• Observables will be emitted one by one and start repeating. Direction is from index 0 to index N.length. When last observable emits it starts at 0 again
• Observables can only emit once at a run: 0 - N.length. Multiple emits are distincted

FYI: The code is actually pretty straight foward. It just looks big, as I added several comments to avoid confusion. If you have questions comment and I will try to answer.

const { Subject, merge } = rxjs;
const { map, scan, distinctUntilChanged, filter } = rxjs.operators;

const source1$ = new Subject();
const source2$ = new Subject();

function toggleEmit(...observables) {
  // amount of all observables
  const amount = observables.length;
  
  // create your updating state that contains the last index and value
  const createState = (value, index) => ({ index, value });

  /*
  * This function updates your state at every emit
  * Keep in mind that updateState contains 3 functions:
  * 1. is called directly: updateState(index, amount)
  * 2. is called by the map operator: map(val => updateState(index, amount)(val)) -> Its just shorthand written
  * 3. is called by the scan operator: fn(state)
  */
  const updateState = (index, amount) => update => state =>
    // Check initial object for being empty and index 0
    Object.keys(state).length == 0 && index == 0
    // Check if new index is one higher
    || index == state.index + 1
    // Check if new index is at 0 and last was at end of observables
    || state.index == amount - 1 && index == 0
      ? createState(update, index)
      : state
  
  // Function is used to avoid same index emit twice
  const noDoubleEmit = (prev, curr) => prev.index == curr.index

  return merge(
    ...observables.map((observable, index) =>
      observable.pipe(map(updateState(index, amount)))
    )
  ).pipe(
    scan((state, fn) => fn(state), {}),
    filter(state => Object.keys(state).length != 0),
    distinctUntilChanged(noDoubleEmit),
    map(state => state.value),
  );
}

const result$ = toggleEmit(source1$, source2$);

result$.subscribe(console.log);

source2$.next(0);
source1$.next(1);
source2$.next(2);
source2$.next(3);
source1$.next(4);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.3/rxjs.umd.min.js"></script>

Because it's Friday, I wrote an implementation of this.

This is javascript not typescript but you can see how I'm solving it.

function roundRobinMerge(...observables) {
  const indexedObservables = observables.map(
    (obs, index) =>
        obs.pipe(
            map(
               v => ({
                    index: index,
                    value: v
                })
          )
      )
  );
  let nextIndexToListenTo = 0;
  return merge(...indexedObservables)
    .pipe(
        filter(n => n.index === nextIndexToListenTo),
        tap(n => nextIndexToListenTo = (n.index + 1)%indexedObservables.length),
        map(n => n.value)
    );
}

To see this in action, go to https://rxviz.com, and copy and paste the code below

const { interval, merge } = Rx;
const { take, map, filter, tap, share } = RxOperators;

function roundRobinMerge(...observables) {
  const indexedObservables = observables.map(
    (obs, index) =>
        obs.pipe(
            map(
            v => ({
                    index: index,
                    value: v
                })
          )
      )
  );
  let nextIndexToListenTo = 0;
  return merge(...indexedObservables)
    .pipe(
        filter(n => n.index === nextIndexToListenTo),
        tap(n => nextIndexToListenTo = (n.index + 1)%indexedObservables.length),
        map(n => n.value)
    );
}

// Now to demonstrate it
function mapper(label) {
  let index = 0;
  return () => `${label}${++index}`;
};

const a = interval(750)
    .pipe(
        map(mapper('a')),
        take(10),
      share()
   );

const b = interval(1300)
   .pipe(
     map(mapper('b')),
     take(10),
     share()
   );


merge(
  [
    a,
    b,
    roundRobinMerge(a, b)
  ]
);

You can probably achieve what you want playing with Subjects.

Here a possible solution.

You start with the concept of creating some sort of switch that is triggered any time either Observable1 or Observable2 notify. Once the switch emits, than we switchMap to one of the source observables, i.e. Observable1 and Observable2), we wait for it to notify and we switchMap to the other one, again we wait for it to notify and finally we next the switch to start all over again.

Looking at the code probably the logic is clearer.

// these are the 2 switches
const switch1 = new Subject<any>();
const switch2 = new Subject<any>();

// res1 and res2 are the Subjects that notify the results we want
const res1 = new Subject<any>();
const res2 = new Subject<any>();

// for each source Observable we create a pipe along the logic explained above
const switchToObs2 = switch1.pipe(
  switchMap(() =>
    observable1.pipe(first()).pipe(
      tap((v) => res1.next(v)),
      switchMap(() => observable2),
      tap((v) => switch1.next(v))
    )
  )
);
const switchToObs1 = switch2.pipe(
  switchMap(() =>
    observable2.pipe(first()).pipe(
      tap((v) => res2.next(v)),
      switchMap(() => observable1),
      tap((v) => switch2.next(v))
    )
  )
);

// we subscribe to the 2 pipes to activate the processing
merge(switchToObs2, switchToObs1).subscribe();

// we start the switches
switch1.next("start");
switch2.next("start");

// eventually we subscribe to the result
merge(res2, res1).subscribe(console.log);

Here a stackblitz with this solution

You could put your sources in an array and use switchMap with a BehaviorSubject:

const sources = [a$, b$];
const source$ = new BehaviorSubject<number>(0);

const emit$ = source$.pipe(
  switchMap(n => sources[n]),
  map((val, i) => {
    const nextIndex = (i+1) % sources.length;
    source$.next(nextIndex);
    return val;
  })
);

Breakdown:

• source$ emits the index of the source you want to listen to
• switchMap subscribes to the source and emits its value
• map calls source$.next() to emit the next index, then simply returns the received value

Notice this works with 1 or more sources.

Here's a working StackBlitz demo.

Here's another interesting (in my opinion anyway) example using repeat. Take a value from one observable, then take another value from the second observable, then repeat:

import { concat, interval, of } from 'rxjs';
import { take, switchMap, repeat, mapTo } from 'rxjs/operators';

const left = interval(10000).pipe(mapTo('left'));
const right = interval(1000).pipe(mapTo('right'));

const example = left.pipe(
  take(1),
  switchMap(leftVal => concat(of(leftVal), right.pipe(take(1)))),
  repeat()
);
console.clear();
const subscribe = example.subscribe(val => console.log('output', val));

stackblitz