RxJS multicast() Multicasting Operator

RxJS multicast() operator is a multicasting operator that returns an observable that emits the results of invoking a specified selector on items emitted by a ConnectableObservable that shares a single subscription to the underlying stream.

The RxJS multicast() operator is used to share the single subscription created with other subscribers.

Syntax:

Following is the syntax of the RxJS multicast() multicasting operator:

  multicast<T, R>(subjectOrSubjectFactory: Subject<T> | (() => Subject<T>), selector?: (source: Observable<T>) => Observable<R>): OperatorFunction<T, R>  

Parameter Explanation

subjectOrSubjectFactory: It specifies the parameter passed to multicast. It is a subject or factory method that returns a subject, or it can be used to create an intermediate subject through which the source sequence’s elements will be multicast to the selector function or Subject to push source elements into.
selector: It is an optional selector function that can use the multicasted source stream as many times as needed without causing multiple subscriptions to the source stream. Subscribers to the given source will receive all notifications of the source from the time of the subscription forward. Its default value is undefined.

Return value

The RxJS multicast() operator returns an observable that emits the results of invoking the selector on the items emitted by a ConnectableObservable that shares a single subscription to the underlying stream.

Let us see some examples of the RxJS multicast() operator to understand it clearly. Before seeing and understanding the working of a multicast() operator, let us first see how the multicast() operator works.

Example 1 (A simple observable with subscription)

  import { Observable } from ‘rxjs’;  var observable = new Observable(function subscribe(subscriber) {     try {        subscriber.next(Math.random());     } catch (e) {        subscriber.error(e);     }  });  const subscribe_one = observable.subscribe(val => console.log(     “Value received from Sub1 = “+val)  );  const subscribe_two = observable.subscribe(val => console.log(     “Value received from Sub2 = “+val)     );  

Output:

After executing the above example, you will see the following result:

In the above example, you can see that the output received from Sub1 and Sub2 are different. This is because when the subscriber gets called, the observable restarts and gives the fresh value available. But we need the subscribers getting called to have the same value.

Now, let’s see how multicast works.

Example 2 (A simple multicast example)

  import { Observable,Subject } from ‘rxjs’;  import { take, multicast, mapTo } from ‘rxjs/operators’;  var observable = new Observable(function subscribe(subscriber) {     try {        subscriber.next(Math.random());     } catch (e) {        subscriber.error(e);     }  });  const multi_op = observable.pipe(multicast(() => new Subject()));  const subscribe_one = multi_op.subscribe(     x => console.log(“Value received from Sub1 = “+x)  );  const subscribe_two = multi_op.subscribe(     x => console.log(“Value received from Sub2 = “+x)  );  multi_op.connect();  

Output:

After executing the above example, you will see the following result:

In the above example, you can see that the same value is shared between the subscribers that are called.

Example 3 (Multicast example with standard Subject)

  import { Subject, interval } from ‘rxjs’;  import { take, tap, multicast, mapTo } from ‘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(‘See the result:’)),    mapTo(‘This will be same’)  );  //subscribe subject to source upon connect()  const multi = example.pipe(multicast(() => new Subject()));  const subscriberOne = multi.subscribe(val => console.log(val));  const subscriberTwo = multi.subscribe(val => console.log(val));  //subscribe subject to source  multi.connect();  

Output:

After executing the above example, you will see the following result:

Example 4 (Multicast example with ReplaySubject)

  import { interval, ReplaySubject } from ‘rxjs’;  import { take, multicast, tap, mapTo } from ‘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(‘See this result:’)),    mapTo(‘This is the result’)  );  //can use any type of subject  const multi = example.pipe(multicast(() => new ReplaySubject(5)));  //subscribe subject to source  multi.connect();  setTimeout(() => {    /*     subscriber will receive all previous values on subscription because      of ReplaySubject     */    const subscriber = multi.subscribe(val => console.group(val));  }, 5000);  

Output:

After executing the above example, you will see the following result:

Next TopicRxJS Operators

RxJS multicast() Multicasting Operator

RxJS multicast() Multicasting Operator

Syntax:

Parameter Explanation

Return value

Example 1 (A simple observable with subscription)

Example 2 (A simple multicast example)

Example 3 (Multicast example with standard Subject)

Example 4 (Multicast example with ReplaySubject)

torch.nn in PyTorch

Akka Actor Life Cycle

You may also like