Friday, 22 September, 2017 UTC
Summary
Observables are known as either hot or cold, depending on the nature of the data producer. Here we’ll go over the differences, why it matters, and how to properly manage both types of observables.
Cold Observables Cold observables are observables where the data producer is created by the observable itself. For example, observables created using the of, from, range, interval and timer operators will be cold. The data is created from within the observable itself, and there truly is not data being produced until the observable is subscribed to.
Here’s an example of a very simple cold observable created using the from operator:
const obs$ = Rx.Observable .from(['🍕', '🍪', '🍔', '🌭', '🍟']) .map(val => { return `Miam ${val}!`; });When a cold observable has multiple subscribers, the whole data stream is re-emitted for each subscriber. Each subscriber becomes independent and gets its own stream of data:
const sub1 = obs$.subscribe(val => { console.log('From sub1:', val); }, null, () => { console.log('done ----------------'); }); const sub2 = obs$.subscribe(val => { console.log('From sub2:', val); }, null, () => { console.log('done ----------------'); });Here’s the outputted result:
From sub1: Miam 🍕! From sub1: Miam 🍪! From sub1: Miam 🍔! From sub1: Miam 🌭! From sub1: Miam 🍟! done ---------------- From sub2: Miam 🍕! From sub2: Miam 🍪! From sub2: Miam 🍔! From sub2: Miam 🌭! From sub2: Miam 🍟! done ----------------
Making a cold observable hot using the share operator
Sometimes it’ll make sense to make a cold observable behave as a hot observable and have the same stream shared between multiple subscribers. This is especially true if you happen to have a cold observable that fetches data from an API. You’ll want to limit the amount of network requests and share the stream will all subscribers.
Here for example, we poll the Hacker News API every 4 seconds for the latest news IDs. The data source, which is the interval, is created by the observable, so the observable itself is cold:
const endpoint = 'https://hacker-news.firebaseio.com/v0/newstories.json?print=pretty'; const obs$ = Rx.Observable .interval(4000) .switchMap(() => { return Rx.Observable.ajax({ url: endpoint, responseType: 'json', method: 'GET' }); }) .pluck('response') .map(res => res.filter((_, index) => index < 6));If we add multiple subscribers to this observables, network requests will be made by each subscriber:
const sub1 = obs$.subscribe(val => { console.log('Sub1:', val); }); setTimeout(() => { const sub2 = obs$.subscribe(val => { console.log('Sub2:', val); }); }, 10000);All we have to do to fix this is add the share operator at the end of our observable pipeline, and this will effectively turn our observable into a hot observable:
const obs$ = Rx.Observable .interval(4000) .switchMap(() => { return Rx.Observable.ajax({ url: endpoint, responseType: 'json', method: 'GET' }); }) .pluck('response') .map(res => res.filter((_, index) => index < 6)) .share();With this, multiple subscribers will share the same stream.
The share operator is known as a multicast operator. Share also manages the underlining subscriptions and disconnects when all the subscribers stop listening. It then restart the stream is a new subscriber subscribes agin. Other multicast operators include the publish, publishReplay and publishLast operators.
Note that in the above example, we create the 2nd subscriber after a delay of 10 seconds. Values that have already been emitted once won’t be repeated for subscribers that arrive late to the party. For example, if we only take the first 4 values:
const obs$ = Rx.Observable .interval(4000) .switchMap(() => { return Rx.Observable.ajax({ url: endpoint, responseType: 'json', method: 'GET' }); }) .pluck('response') .map(res => res.filter((_, index) => index < 6)) .take(4) .share();You’ll notice that the 2nd subscriber, sub2, only logs out 2 times because it misses the first emitted values.
Hot Observables Hot observables, on other hand, have their data producer outside the observable itself. These observables are closing over external data producers. For example, observables created with the fromEvent operator for either user events (clicks, mouse moves,…) or WebSocket events are hot observables. The data is being produced regardless of if there’s a subscriber or not. If there’s no subscriber when the data is being produced, the data is simply lost.
Here’s a simple hot observable of mouse clicks:
const obs$ = Rx.Observable.fromEvent(document, 'click') .map(event => ({ clientX: event.clientX, clientY: event.clientY }));Now let’s create two subscribers for it:
const sub1 = obs$.subscribe(val => { console.log('Sub1:', val); }); setTimeout(() => { console.log('Start sub2'); const sub2 = obs$.subscribe(val => { console.log('Sub2:', val); }); }, 4000);Here both subscribers share the same stream, but the second subscription misses out on the values emitted before 4 seconds.
Making a hot observable cold
To transform a hot observable into a cold one, the observable has to become the data producer. In the following example, we create an observable factory that returns a new observable for each subscription. This way, each subscriber will get its own independent stream:
const obsFactory = () => Rx.Observable.fromEvent(document, 'click') .map(event => ({ clientX: event.clientX, clientY: event.clientY })); const sub1 = obsFactory().subscribe(val => { console.log('Sub1:', val); }); setTimeout(() => { console.log('Start sub2'); const sub2 = obsFactory().subscribe(val => { console.log('Sub2:', val); }); }, 4000);As Ben Lesh mentioned in this excellent post, there is perhaps a third category of observables, warm observables, where data starts being emitted only after the 1st subscription, but where the data is still shared between multiple observables.