On this page

In this article I‘d like share my experience with unicast and multicast Observables (called cold and hot respectively), how these can be generated in RxJs, how to multicast and how to share side-effects.

An Observable is a representation of any set of values over any amount of time. — Ben Lesh. (RxJS In-Depth talk)

Observable 👁 #

A quick recap of what an Observable is:

  • A “Stream” or “Set” of 0 to many values of any type (Numbers, Strings, Array, Objects…).
  • Functions that ties producer to consumer.
  • “Lazy” because the logic inside is not executed until someone is subscribing to the stream. Also called a cold observable.

This is the most basic building block of RxJS for programming with asynchronous data streams. Anything can be a stream and wrapped into an Observable: network calls, user inputs, variables, properties, data structures, etc.

Cold Observable ❄️ #

We can define a Observable as cold or unicast when it starts emitting values after a subscriber is registered, under the hood it is creating a producer for each subscription. The following graphic illustrates a cold observable example.

Cold Observable and its subscriptions

Cold Observable and its subscriptions

and the following code is an example with RxJs:

// Cold Observable - Producer
const source$ = Rx.Observable.interval(1000)
  .take(2)
  .map(generateValue);

// Consumers
const consumer1 = (value) => logDataAndTime('Value from Consumer #1', value);
const consumer2 = (value) => logDataAndTime('Value from Consumer #2', value);

// Registration before emitted value
console.log('Consumer #1 Registered');
source$.subscribe(consumer1);

// Deferred Registration
setTimeout(() => {
  console.log('Consumer #2 Registered');
  source$.subscribe(consumer2);
}, 3000);

/*
* Expected Result:
* source$   : *----A----B|
* consumer1 : *----A----B|
* consumer2 : ...............*----A----B|
*/

A cold observable is created and two consumers are registered at two different times. RxJs Observables are generated by default as cold.

Hot Observable 🔥 #

We can call an observable hot when it emits values before any consumer is subscribed. Each subscription could receive different values based on their registration time.

Hot Observable and its subscriptions.

Hot Observable and its subscriptions.

Heat up an Observable 🌞 #

RxJs provides methods to generate a variety of multicasted observables to have the best control over the underlying streams, let’s explore some of them:

- publish(): ConnectableObservable #

I’ve found that the simplest way is using ConnectableObservable which is an extension of Observable and starts emitting values only when connect method is called.

// Cold
const cold$ = Observable.interval(1000)
    .map(el => el++)
    .take(2);

// Hot
const hot$ = cold$.publish();

// Consumers
const consumer1 = (value) => logDataAndTime('Consumer #1', `Value read: ${value}`);
const consumer2 = (value) => logDataAndTime('Consumer #2', `Value read: ${value}`);

// Registration before emitted value
console.log('Consumer #1 Registered');
hot$.subscribe(consumer1);

// Deferred Registration
setTimeout(() => {
  console.log('Consumer #2 Registered');
  hot$.subscribe(consumer2);
}, 3000);
hot$.connect();

/*
* Expected Result:
* hot$     : *----A----B|
* consumer1: *----A----B
* consumer2: ................*
*/

As you will notice hot$ completes its stream before consumer2 starts consuming.

- publishReplay(): ConnectableObservable #

This method allows us to specify a buffer that can be consumed by late subscribers.

// Generate Cold Obervable with 'completed' side effect
const cold$ = Observable.interval(1000)
    .map(el => ++el)
    .do(analyticCall)
    .do(printIfCompleted)
    .take(2)

// Make it Hot and setup a buffer
const hot$ = cold$.publishReplay(1);

// Consumers
const consumer1 = value => logDataAndTime('Consumer #1', `Value read: ${value}`);
const consumer2 = value => logDataAndTime('Consumer #2', `Value read: ${value}`);

// Registration before emitted value
console.log('Consumer #1 Registered');
hot$.subscribe(consumer1);

// Deferred Registration
setTimeout(() => {
  console.log('Consumer #2 Registered');
  hot$.subscribe(consumer2);
}, 3000);
hot$.connect();

/*
* Expected Result:
*
* hot$         : *----A----B|
* side effects : *----S----S
* consumer1    : *----A----B
* consumer2    : ...............*-B
*/

In this scenario we notice how the second consumer subscribes to a completed stream but still gets the final result. Here we start seeing one benefit of hot Observables: shared side effects*.*

printIfCompleted and analyticCall are processed only when the stream is active and are not recomputed when consumer2 registers.

- refCount(): Observable<T> #

refCount makes the multicasted Observable automatically start executing when the first subscriber arrives, and stop executing when the last subscriber leaves.

// Generate Cold Obervable with 'completed' side effect
const cold$ = Observable.interval(1000)
    .map(el => ++el)
    .do(analyticCall)
    .do(printIfCompleted)
    .take(2)

// Make it Hot
const hot$ = cold$.publish().refCount();

// Consumers
const consumer1 = value => logDataAndTime('Consumer #1', `Value read: ${value}`);
const consumer2 = value => logDataAndTime('Consumer #2', `Value read: ${value}`);
const consumer3 = value => logDataAndTime('Consumer #3', `Value read: ${value}`);

// Registration before emitted value
console.log('Consumer #1 Registered');
hot$.subscribe(consumer1);

// Deferred Registration
setTimeout(() => {
  console.log('Consumer #2 Registered');
  hot$.subscribe(consumer2);
}, 1000);

// Deferred Registration, too late
setTimeout(() => {
  console.log('Consumer #3 Registered');
  hot$.subscribe(consumer3);
}, 4000);

/*
* Expected Result:
* hot$         : *----A----B|
* side effects : *----S----S
* consumer1    : *----A----B
* consumer2    : .....*----B
* consumer2    : ..........*
*/

- share(): Observable<T> #

share behaves very similar to refCount however when the underlying observable is completed, it will restart emitting values when a new subscription is made.

// Generate Cold Obervable with 'completed' side effect
const cold$ = Observable.interval(1000)
    .map(el => ++el)
    .do(analyticCall)
    .do(printIfCompleted)
    .take(2)

// Make it Hot
const hot$ = cold$.share();

// Consumers
const consumer1 = value => logDataAndTime('Consumer #1', `Value read: ${value}`);
const consumer2 = value => logDataAndTime('Consumer #2', `Value read: ${value}`);
const consumer3 = value => logDataAndTime('Consumer #3', `Value read: ${value}`);

// Registration before emitted value
console.log('Consumer #1 Registered');
hot$.subscribe(consumer1);

// Deferred Registration
setTimeout(() => {
  console.log('Consumer #2 Registered');
  hot$.subscribe(consumer2);
}, 1000);

// Deferred Registration, too late
setTimeout(() => {
  console.log('Consumer #3 Registered');
  hot$.subscribe(consumer3);
}, 4000);

/*
* Expected Result:
* hot$         : *----A----B|
* side effects : *----S----S..----S----S
* consumer1    : *----A----B
* consumer2    : .....*----B
* consumer3    : ............*----A----B
*/

- shareReplay(): Observable<T> #

This time, three subscriptions are made. The first works as expected. The second is late, but still receives some notifications. The third is far too late, but is handed over the most recent notification emitted by the hot$ observable.

It could look really similar to publishReplay because of the buffer size as parameter. The main difference is that values start to be generated when at least 1 subscription is registered.

// Generate Cold Obervable with 'completed' side effect
const cold$ = Observable.interval(1000)
    .map(el => ++el)
    .do(analyticCall)
    .do(printIfCompleted)
    .take(2)

const hot$ = cold$.shareReplay(1);

// Consumers
const consumer1 = value => logDataAndTime('Consumer #1', `Value read: ${value}`);
const consumer2 = value => logDataAndTime('Consumer #2', `Value read: ${value}`);
const consumer3 = value => logDataAndTime('Consumer #3', `Value read: ${value}`);

setTimeout(() => {
  console.log('Consumer #1 Registered');
  hot$.subscribe(consumer1);
}, 500);
setTimeout(() => {
  console.log('Consumer #2 Registered');
  hot$.subscribe(consumer2);
}, 2000);
setTimeout(function () {
  console.log('Consumer #3 Registered');
  hot$.subscribe(consumer3);
}, 6000);

/*
* Expected Result:
* hot$         : ----A----B|
* side effects : ----S----S
* consumer1    : ..--A----B
* consumer2    : .......*AB
* consumer3    : ...........*B
*/

- multicast(): Observable<T> #

In contrast to publish or share method, multicast enables full control over the subject through which the underlying stream will be shared with the subscribers. In the following example we are multicasting a simple subject with cold$.multicast(new Subject()) which is equal to cold$.publish() .

// Generate Cold Obervable with side effects
const cold$ = Observable.interval(500)
    .map(el => ++el)
    .do(analyticCall)
    .do(printIfCompleted)
    .take(3);

const hot$ = cold$.multicast(new Subject());

hot$.connect();

// Consumers
const consumer1 = value => logDataAndTime('Consumer #1', `Value read: ${value}`);
const consumer2 = value => logDataAndTime('Consumer #2', `Value read: ${value}`);

hot$.subscribe(consumer1);
setTimeout(() => { hot$.subscribe(consumer2); }, 1100)

/*
* Expected Result:
* hot$         : *----A----B----C|
* side effects : *----S----S----S
* consumer1    : *----A----B----C
* consumer2    : ............*--C
*/

AsyncSubject , BehaviourSubject andReplaySubject can be passed into multicast method to simulate publishLast , publishBehaviour , publishReplay respectively.

Examples:

cold$.publishReplay() // cold$.multicast(new ReplaySubject()) cold$.publishBehaviour() // cold$.multicast(new BehaviourSubject()) cold$.publishLast() // cold$.multicast(new AsyncSubject())

Conclusions #

When should I use cold or hot? It depends on your goal, usually you would need to create a hot observable to have a single source of truth, share side effects and final result to the subscribers.

For example in a web socket you really don’t want to create more then one connection and you could need to share the entire sets of values: Link.

Or for an REST Api call you would like to share the result without asking the server every time: Link.

You should chose the cold observable instead, when you need to create a producer and generate side effects for every subscribers.

RxJs provides many tools to generate, manipulate and transform observables that achieve slightly different results and adapt to your needs. Being new in the Rx paradigm I wanted to gather reusable examples that would be helpful for my reactive journey.