RxJS publish() Multicasting Operator

by Online Tutorials Library July 22, 2022

RxJS publish() Multicasting Operator

RxJS publish() operator is a multicasting operator that returns a ConnectableObservable, a variety of Observable that waits until its connect method is called before it begins emitting items to those Observers that have subscribed to it.

In other words, we can say that the RxJS publish() operator returns ConnectableObservable and requires to use connect() method to subscribe to the observables.

Syntax:

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

  publish<T, R>(selector?: OperatorFunction<T, R>): MonoTypeOperatorFunction<T> | OperatorFunction<T, R>  

Parameter Explanation

selector: It is an optional function which can use the multicasted source sequence as many times as required, without causing multiple subscriptions to the source sequence. It default value is undefined.

Return value

The RxJS publish() operator returns a ConnectableObservable that causes the source observable to emit items to its observers when the connection is established.

Let us see some examples of the RxJS publish() operator to understand it clearly.

Example 1 (Connect observable after subscribers)

  import { interval } from ‘rxjs’;  import { publish, tap } from ‘rxjs/operators’;  //emit value every 1 second  const source = interval(1000);  //do nothing until connect() is called  const example = publish()(source.pipe(    //side effects will be executed once    tap(_ => console.log(‘Now start!’)),  ));  const subscribe = example.subscribe(val =>    console.log(`Subscriber One: ${val}`)  );  const subscribeTwo = example.subscribe(val =>    console.log(`Subscriber Two: ${val}`)  );  //call connect after 5 seconds, causing source to begin emitting items  setTimeout(() => {    example.connect();  }, 5000);  

Output:

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

Example 2 (Merge each inner observable into a single one and then subscribe)

  import { of, zip, interval, merge } from “rxjs”;  import { map, publish, tap } from “rxjs/operators”;  const source$ = zip(interval(2000), of(1, 2, 3, 4, 5)).pipe(    map(values => values[1])  );  source$    .pipe(      publish(multicasted$ =>        merge(          multicasted$.pipe(tap(x => console.log(‘Stream 1:’, x))),          multicasted$.pipe(tap(x => console.log(‘Stream 2:’, x))),          multicasted$.pipe(tap(x => console.log(‘Stream 3:’, x))),        )      )    )    .subscribe();  

Output:

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

Next TopicRxJS Operators

RxJS publish() Multicasting Operator

RxJS publish() Multicasting Operator

Syntax:

Parameter Explanation

Return value

Example 1 (Connect observable after subscribers)

Example 2 (Merge each inner observable into a single one and then subscribe)

torch.nn in PyTorch

Akka Actor Life Cycle

You may also like