The most elegant way to switch subscriptions in RxJS

Question

As I am already quite experienced developer in .NET, I am learning new technologies and I am curious how would you resolve this problem in TypeScript + RxJS. Let's say, I have a multiple columns (fields) and I am observing their value changes. I also have a switch which has groups of columns to observe - in this example 1, 2, 3; A, B, C and X, Y, Z.

Every time I have to switch, I would like to remove subscriptions from previous group, and create for new one. The code looks like that:

const log = (cid: string) => console.info(`${cid} changed`);

this.subscriptions.add(this.observeValueChanges("column 1").subscribe(log));
this.subscriptions.add(this.observeValueChanges("column 2").subscribe(log));
this.subscriptions.add(this.observeValueChanges("column 3").subscribe(log));

//Switching to another columns collection - I don't want previous subscriptions anymore
this.subscriptions.unsubscribe();
this.subscriptions.add(this.observeValueChanges("column A").subscribe(log));
this.subscriptions.add(this.observeValueChanges("column B").subscribe(log));
this.subscriptions.add(this.observeValueChanges("column C").subscribe(log));

//Switching to another columns collection - I don't want previous subscriptions anymore
this.subscriptions.unsubscribe();
this.subscriptions.add(this.observeValueChanges("column X").subscribe(log));
this.subscriptions.add(this.observeValueChanges("column Y").subscribe(log));
this.subscriptions.add(this.observeValueChanges("column Z").subscribe(log));

for reference:

private subscriptions: Subscription = new Subscription();

and

observeValueChanges(columnId: string): Observable<any> {
     return timer(0, 1000).pipe(map(_ => columnId), publish(), refCount());
}

I can make it working by substituting subscription object by array of subscriptions, and after switching - I can unsubscribe all from array (and instead of add - I can push). What do you thing? I have no problem with making "switch signal" as observable.

Answer 1

It's interesting you used the word "switch", because the operator you are looking for is called switchMap!

Whenever a new value is received, it will "map" it to an observable and "switch" from the previous source.

switchMap is a "Higher Order Mapping Operator", which does a few nice things for you:

1. maps the incoming value to another observable,
2. subscribes to it, so its values are emitted
3. manages unsubscribing from these "inner subscriptions" automatically

I've created a working sample in this StackBlitz

import { fromEvent } from 'rxjs'; 
import { tap, map, switchMap } from 'rxjs/operators';

const group0 = document.querySelectorAll('.group0');
const group1 = document.querySelectorAll('.group1');

const selectedGroup = document.querySelectorAll('.selected-group');

// Array of possible sources to "switch" between
//
// [fromEvent] Creates an observable that emits DOM
//             events for the specified elements.
//
// [map] Just maps the event to the target value.
const sources$ = [
  fromEvent(group0, 'keyup').pipe(map(event => event.target['value'])),
  fromEvent(group1, 'keyup').pipe(map(event => event.target['value']))
];

// selectedGroup$ emits the value of the selected radio button which
// for this example is conveniently the group's index in the array.
const selectedGroup$ = fromEvent(selectedGroup, 'change').pipe(
  map(event => event.target['value'])
);

// source$ starts with the emission of selectedGroup$. Then we
// use [switchMap] to "switch" our source observable and map
// it to a new observable source. Then switchMap will
// automatically unsubscribe from the previous one.
const source$ = selectedGroup$.pipe(
  switchMap(groupIndex => sources$[groupIndex])
)

// Notice we only need a single subscription!
source$.subscribe(console.log);