Hot n Cold Observables

There are hot and cold observables. Let’s talk about what a cold observable is. In a cold observable two subscribers get their own copies of values like so:

  1. // cold observable example
  2. let stream$ = Rx.Observable.of(1,2,3);
  3. //subscriber 1: 1,2,3
  4. stream.subscribe(
  5.    data => console.log(data),
  6.    err => console.error(err),
  7.    () => console.log('completed')
  8. )
  10. //subscriber 2: 1,2,3
  11. stream.subscribe(
  12.    data => console.log(data),
  13.    err => console.error(err),
  14.    () => console.log('completed')
  15. )

In a hot observable a subscriber receives values when it starts to subscribe, it is however more like a live streaming in football, if you start subscribing 5 minutes in the game, you will have missed the first 5 minutes of action and you start receiving data from that moment on:

  1. let liveStreaming$ = Rx.Observable.interval(1000).take(5);
  3. liveStreaming$.subscribe( 
  4.   data => console.log('subscriber from first minute')
  5.   err => console.log(err),
  6.   () => console.log('completed')
  7. )
  9. setTimeout(() => {
  10.    liveStreaming$.subscribe( 
  11.   data => console.log('subscriber from 2nd minute')
  12.   err => console.log(err),
  13.   () => console.log('completed')
  14. ) 
  15. },2000)

From cold to hot - Katy Perry mode

In the example above it isn’t really hot, as a matter of fact both subscribers of the values will each receive 0,1,2,3,4. As this is the live streaming of a football game it doesn’t really act like we want it to, so how to fix it?

Two components are needed to make something go from cold to hot. publish() and connect().

  1. let publisher$ = Rx.Observable
  2. .interval(1000)
  3. .take(5)
  4. .publish();
  7. publisher$.subscribe( 
  8.   data => console.log('subscriber from first minute',data),
  9.   err => console.log(err),
  10.   () => console.log('completed')
  11. )
  13. setTimeout(() => {
  14.     publisher$.subscribe( 
  15.         data => console.log('subscriber from 2nd minute', data),
  16.         err => console.log(err),
  17.         () => console.log('completed')
  18.     ) 
  19. }, 3000)
  22. publisher$.connect();

In this case we see that the output for the first stream that starts to subscribe straight away is 0,1,2,3,4 whereas the second stream is emitting 3,4. It’s clear it matters when the subscription happens.

Warm observables

There is another type of observables that acts a lot like a hot observable but is in a way lazy. What I mean with this is that they are essentially not emitting any values until a subscriber arrives. Let’s compare a hot and a warm observable

hot observable

  1. let stream$ = Rx.Observable
  2. .interval(1000)
  3. .take(4)
  4. .publish();
  6. stream$.connect();
  8. setTimeout(() => {
  9.   stream$.subscribe(data => console.log(data))
  10. }, 2000);

Here we can see that the hot observable will loose the first value being emitted as the subscribe arrives late to the party.

Let’s contrast this to our warm observable

warm observable

  1. let obs = Rx.Observable.interval(1000).take(3).publish().refCount();
  3. setTimeout(() => {
  4.     obs.subscribe(data => console.log('sub1', data));
  5. },1100)
  7. setTimeout(() => {
  8.     obs.subscribe(data => console.log('sub2', data));
  9. },2100)

The refCount() operator ensures this observable becomes warm, i.e no values are emitted until sub1 subscribes. sub2 on the other hand arrives late to the party, i.e that subscription receives the value its currently on and not the values from the beginning.

So an output from this is

  1. sub1 : 0
  2. sub1 : 1
  3. sub2 : 1
  4. sub1 : 2
  5. sub2 : 2

This shows the following, first subscriber starts from 0. Had it been hot it would have started at a higher number, i/e it would have been late to the party. When the second subscriber arrives it doesn’t get 0 but rather 1 as the first number showing it has indeed become hot.

Naturally hot observables

Generally something is considered hot if the values are emitted straight away without the need for a subscriber to be present. A naturally occuring example of a hot observable is mousemove. Most other hot observables are the result of cold observables being turned hot by using publish() and connect() or by using the share() operator.

Sharing

Sharing means using a useful operator called share(). Imagine you have the following normal cold observable case :

  1. let stream$ = Rx.Observable.create((observer) => {
  2.     observer.next( 1 );
  3.     observer.next( 2 );
  4.     observer.next( 3 );
  5.     observer.complete();
  6. }).share()
  8. stream$.subscribe(
  9.     (data) => console.log('subscriber 1', data),
  10.     err => console.error(err),
  11.     () => console.log('completed')
  12. );
  13. stream$.subscribe(
  14.     (data) => console.log('subscriber 2', data),
  15.     err => console.error(err),
  16.     () => console.log('completed')
  17. );

If you set a breakpoint on observer.next(1) you will notice that it’s being hit twice, once for every subscriber. This is the behaviour we expect from a cold observable. Sharing operator is a different way of turning something into a hot observable, in fact it not only turns something hot under the right conditions but it falls back to being a cold observable under certain conditions. So what are these conditions ?

1) Created as hot Observable : An Observable has not completed when a new subscription comes and subscribers > 0

2) Created as Cold Observable Number of subscribers becomes 0 before a new subscription takes place. I.e a scenario where one or more subscriptions exist for a time but is being unsubscribed before a new one has a chance to happen

3) Created as Cold Observable when an Observable completed before a new subscription

Bottom line here is an active Observable producing values still and have at least one preexisting subscriber. We can see that the Observable in case 1) is dormant before a second subscriber happens and it suddenly becomes hot on the second subscriber and thereby starts sharing the data where it is.