multicast
signature: multicast(selector: Function): Observable
Share source utilizing the provided Subject.
Examples
Example 1: multicast with standard Subject
import { interval } from 'rxjs/observable/of';
import { Subject } from 'rxjs/Subject';
import { take, tap, multicast } 'rxjs/operators';
//emit every 2 seconds, take 5
const source = interval(2000).pipe(take(5));
const example = source.pipe(
//since we are multicasting below, side effects will be executed once
tap(() => console.log('Side Effect #1')),
mapTo('Result!');
);
//subscribe subject to source upon connect()
const multi = example.pipe(multicast(() => new Subject()));
/*
subscribers will share source
output:
"Side Effect #1"
"Result!"
"Result!"
...
*/
const subscriberOne = multi.subscribe(val => console.log(val));
const subscriberTwo = multi.subscribe(val => console.log(val));
//subscribe subject to source
multi.connect();
Example 2: multicast with ReplaySubject
import { interval } from 'rxjs/observable/of';
import { take, multicast } 'rxjs/operators';
//emit every 2 seconds, take 5
const source = interval(2000).pipe(take(5));
//example with ReplaySubject
const example = source.pipe(
//since we are multicasting below, side effects will be executed once
tap(_ => console.log('Side Effect #2')),
mapTo('Result Two!')
);
//can use any type of subject
const multi = example.pipe(multicast(() => new Rx.ReplaySubject(5)));
//subscribe subject to source
multi.connect();
setTimeout(() => {
/*
subscriber will receieve all previous values on subscription because
of ReplaySubject
*/
const subscriber = multi.subscribe(val => console.group(val));
}, 5000);
Additional Resources
- multicast
- Official docs
Source Code:
https://github.com/ReactiveX/rxjs/blob/master/src/internal/operators/multicast.ts