[RxJS] Multicast with a selector argument, as a sandbox

Posted 2020-08-16 Answer1215

Let‘s explore a different use of the multicast() operator in RxJS, where you can provide a selector function as a sandbox where the shared Observable is available.

 

When we have code like below:

var result = Rx.Observable.interval(1000).take(6)
  .do(x => console.log(‘source ‘ + x))
  .map(x => Math.random());

var delayedResult = result.delay(500);
var merged = result.merge(delayedResult);

merged.subscribe( (x) => console.log(x))

/*
"source 0"
0.5832993222895915
"source 0"
0.031394357976560316
"source 1"
0.27602687148865
"source 1"
0.8762531748833942
"source 2"
0.49254272653868103
"source 2"
0.8024593359949526
...
*/

You can notice that, it runs ‘result‘ block twice each time, it because ‘merged‘ subscribe to ‘result‘ and ‘delayedResult‘ also subscribe to ‘result‘, therefore it log out source twice.

 

If you only want one subscribe, you can use multicast(), with a second param which is sandbox function.

Normally you will use mulitcast() with refCount():

function subjectFactory() {
  return new Rx.Subject(); 
}

var result = Rx.Observable.interval(1000).take(6)
  .do(x => console.log(‘source ‘ + x))
  .map(x => Math.random())
  .multicast(subjectFactory).refCount();

var sub = result.subscribe(x => console.log(x));

 

If you pass a second param:

var result = Rx.Observable.interval(1000).take(6)
  .do(x => console.log(‘source ‘ + x))
  .map(x => Math.random())
  .multicast(subjectFactory, function sandbox(shared) {
    var delayedShare = shared.delay(500);
    var merged = shared.merge(delayedShare);
    return merged;
  });

result.subscribe(x => console.log(x));

/*
"source 0"
0.9214861149095479
0.9214861149095479
"source 1"
0.1684919218677523
0.1684919218677523
"source 2"
0.28182876689689795
0.28182876689689795
...
*/

Notice that, is you pass a second param to multicase(), the return value is no longer an connectableObservable. It is just a normal observable. So you cannot call ‘refCount()‘ anymore. 

And inside sandbox() function, you need to retrun a observable.

From the results can see, we no longer subscribe the source twice.

 

The takeaway is you should use a selector function in multicast when you want to create, let‘s say, a diamond-shaped dependency. Here we have a bifurcation. As you see we have shared, and it‘s used in two parts, and then we converge those two parts together to return one observable. That‘s kind of like a diamond shape, where we bifurcate, and then we converge.

That is one case where you almost always want to use a selector function in multicast. If you don‘t, then usually we use just multicast with a refCount. That‘s quite common to use.