multicast – RxJS Reference

Understanding the multicast Operator

The multicast operator is a tool for inserting a Subject into an observable stream. This enables a single subscription to the underlying stream to be shared across multiple subscribers. The name reflects the idea of distributing values to multiple observers simultaneously.

Consider this demonstration:

function producer(observer) {
   let counter = 0;
   setInterval(() => {
       console.log('producer\t\t: ' + counter);
       observer.next(counter);
       counter++;
   }, 1000);
}

const obs = new Observable(producer);

obs.subscribe((v) => console.log('subscription 1\t: ' + v));
obs.subscribe((v) => console.log('subscription 2\t: ' + v));

The stream obs logs values it emits at one-second intervals.

When we subscribe, emissions begin immediately. Every subscription invokes its own producer function, resulting in two separate streams running independently. Consequently, each number appears in the console four times — twice from within each producer function and twice from each subscription handler:

producer		: 0
subscription 1	: 0
producer		: 0
subscription 2	: 0
producer		: 1
subscription 1	: 1
producer		: 1
subscription 2	: 1

To share the underlying stream across multiple subscriptions while triggering the producer function only once, a Subject is required. Here’s a manual approach:

const shared = new Subject();

shared.subscribe((v) => console.log('subscription 1\t: ' + v));
shared.subscribe((v) => console.log('subscription 2\t: ' + v));

obs.subscribe(shared);

The console output looks like this:

producer		: 0
subscription 1	: 0
subscription 2	: 0
producer		: 1
subscription 1	: 1
subscription 2	: 1

Notice that we now see three lines — two from the subscription handlers and one from the producer function. This confirms the producer function executes only once.

The manual setup with a Subject and explicit subscriptions can feel verbose. This is where multicast simplifies things. The same outcome can be achieved as follows:

const shared = obs.pipe(
   multicast(new Subject())
);

shared.subscribe((v) => console.log('subscription 1\t: ' + v));
shared.subscribe((v) => console.log('subscription 2\t: ' + v));

shared.connect();

Essentially, multicast just introduces a Subject into the observable chain. The earlier example reveals that it’s the Subject itself that manages sharing — the multicast operator merely streamlines working with it.

An important detail: we invoke the connect method on shared, which is a ConnectableObservable. This connects the underlying Subject to the original stream. Subscribing to a connectable observable like shared by itself does not trigger a Subject subscription to the source — such subscriptions are essentially made to the Subject.

Calling connect on a connectable observable parallels directly subscribing a Subject to the source, as demonstrated here:

const shared = new Subject();

shared.subscribe((v) => console.log('subscription 1\t: ' + v));
shared.subscribe((v) => console.log('subscription 2\t: ' + v));

obs.subscribe(shared);

Rather than passing a Subject instance, multicast can accept a factory function that creates a Subject. These two options differ in behavior.

By default, a Subject that has stopped cannot re-subscribe to the underlying stream. Once the source completes, all subsequent subscriptions receive only the COMPLETE notification. In the example below, the stream is limited to three values before completing at 1500ms. A subscription attempted at 2000ms gets only the COMPLETE notification:

const obs = interval(500).pipe(take(3));

const shared = obs.pipe(
   multicast(new Subject())
) as ConnectableObservable<any>;

shared.subscribe({
   next(v) { console.log(v) },
   complete() { console.log('complete') }
});

shared.connect();

setTimeout(() => {
   shared.subscribe({
       next(v) { console.log(v) },
       complete() { console.log('complete') }
   });
}, 2000);

This produces the following output:

0
1
2
complete
complete

When a factory function is provided instead, multicast invokes it to create a fresh Subject when the current one is already terminated and a new subscription arrives. You’ll then need to call connect again. This behavior is illustrated here:

const obs = interval(500).pipe(take(3));

const shared = obs.pipe(
   multicast(() => new Subject())
) as ConnectableObservable<any>;

shared.subscribe({
   next(v) { console.log(v) },
   complete() { console.log('complete') }
});

shared.connect();

setTimeout(() => {

   shared.subscribe({
       next(v) { console.log(v) },
       complete() { console.log('complete') }
   });

   shared.connect();
}, 2000);

The output this time differs:

0
1
2
complete
0
1
2
complete

Observe the second connect call inside the setTimeout, after the subscription — this links a new Subject. Adding a console log within the factory function will show it executing when the subscription in the setTimeout occurs.