Hot and Cold observables: are there 'hot' and 'cold' operators?

Question

I reviewed the following SO question: What are the Hot and Cold observables?

To summarize:

• a cold observable emits its values when it has an observer to consume them, i.e. the sequence of values received by observers is independent of time of subscription. All observers will consume the same sequence of values.
• a hot observable emits value independently of its subscriptions, i.e. the values received by observers are a function of the time of subscription.

Yet, I feel like hot vs. cold is still a source of confusion. So here are my questions:

• Are all rx observables cold by default (with the exception of subjects)?

  I often read that events are the typical metaphor for hot observables, but I also read that Rx.fromEvent(input, 'click') is a cold observable(?).

• Are there/what are the Rx operators which turn a cold observables into a hot observable (apart from publish, and share)?

  For instance, how does it work with Rx operator withLatestFrom? Let cold$ be a cold observable which has somewhere been subscribed to. Will sth$.withLatestFrom(cold$,...) be a hot observable?

  Or if I do sth1$.withLatestFrom(cold$,...), sth2$.withLatestFrom(cold$,...) and subscribe to sth1 and sth2, will I always see the same value for both sth?

• I thought Rx.fromEvent creates cold observables but that is not the case, as mentioned in one of the answers. However, I am still baffled by this behaviour: https://codepen.io/anon/pen/NqQMJR?editors=101. Different subscriptions get different values from the same observable. Wasn't the click event shared?

Solution 1

I am coming back a few months later to my original question and wanted to share the gained knowledge in the meanwhile. I will use the following code as an explanation support (jsfiddle):

var ta_count = document.getElementById('ta_count');
var ta_result = document.getElementById('ta_result');
var threshold = 3;

function emits ( who, who_ ) {return function ( x ) {
  who.innerHTML = [who.innerHTML, who_ + " emits " + JSON.stringify(x)].join("\n");
};}

var messages$ = Rx.Observable.create(function (observer){
  var count= 0;
  setInterval(function(){
    observer.onNext(++count);
  }, 1000)
})
.do(emits(ta_count, 'count'))
.map(function(count){return count < threshold})
.do(emits(ta_result, 'result'))

messages$.subscribe(function(){});

As mentioned in one of the answers, defining an observable leads to a series of callback and parameter registration. The data flow has to be kicked in, and that is done via the subscribe function. A (simplified for illustration) detailed flow can be find thereafter.

Observables are cold by default. Subscribing to an observable will result in an upstream chain of subscriptions taking place. The last subscription leads to the execution of a function which will handle a source and emit its data to its observer.

That observer in its turn emits to the next observer, resulting in a downstream flow of data, down to the sink observer. The following simplified illustration shows subscription and data flows when two subscribers subscribe to the same observable.

Hot observables can be created either by using a subject, or through the multicast operator (and its derivatives, see Note 3 below).

The multicast operator under the hood makes use of a subject and returns a connectable observable. All subscriptions to the operator will be subscriptions to the inner subject. When connect is called, the inner subject subscribes to the upstream observable, and data flows downstream. Subjects manipulate internally a list of subscribed observers and multicast incoming data to all subscribed observers.

The following diagram summarizes the situation.

In the end, it matters more to understand the flow of data caused by the observer pattern and the implementation of the operators.

For instance, if obs is hot, is hotOrCold = obs.op1 cold or hot? Whatever the answer is :

• if there are no subscribers to obs.op1, no data will flow through op1. If there were subscribers to hot obs, that means obs.op1 will have possibly lost pieces of data
• supposing that op1 is not a multicast-like operator, subscribing twice to hotOrCold will subscribe twice to op1, and every value from obs will flow twice through op1.

Notes :

1. This information should be valid for Rxjs v4. While the version 5 has gone through considerable changes, most of it still applies verbatim.
2. Unsubscription, error and completion flows are not represented, as they are not in the scope of the question. Schedulers are also not taken into account. Among other things, they influence the timing of the data flow, but a priori not its direction and content.
3. According to the type of subject used for multicasting, there are different derived multicasting operators:

Subject type | `Publish` Operator | `Share` operator ------------------ | --------------------------- | ----------------- Rx.Subject | Rx.Observable.publish | share Rx.BehaviorSubject | Rx.Observable.publishValue | shareValue Rx.AsyncSubject | Rx.Observable.publishLast | N/A Rx.ReplaySubject | Rx.Observable.replay | shareReplay

Update : See also the following articles, here, and there) on that subject by Ben Lesh.

Further details on subjects can be found in this other SO question : What are the semantics of different RxJS subjects?

Solution 2

Your summary, and the linked question are both correct, I think the terminology may be confusing you. I propose you think of hot and cold observables as active and passive observables (respectively).

That is, an active (hot) observable will be emitting items whether someone has subscribed or not. The canonical example, again, button click events happen whether someone is listening to them or not. This distinction is important because, if, for example, I click a button and then subscribe to button clicks (in that order), I will not see the button click that has already happened.

A passive (cold) observable will wait until a subscriber exists before emitting items. Imagine a button where you cannot click on it until someone is listening to the events—this would ensure that you always see each and every click event.

Are all Rx observables "cold" (or passive) by default? No, Rx.fromEvent(input, 'click') for example is a hot (or active) observable.

I also read that Rx.fromEvent(input, 'click') is a cold observable(?)

That is not the case.

Are there Rx operators which turn a cold observable into a hot observable?

The concept of turning a hot (active) observable into a cold (passive) observable is this: you need to record the events that happen while nothing is subscribed and offer those items (in various ways) to subscribers that come along in the future. One way to do this is to use a Subject. For example, you could use a ReplaySubject to buffer up items emitted and replay them to future subscribers.

The two operators you named (publish and share) both use subjects internally to offer that functionality.

How does it work with Rx operator withLatestFrom? Let cold$ be a cold observable which has been subscribed to. Will something$.withLatestFrom(cold$,...) be a hot observable?

If something is a hot observable, then yes. If something is a cold observable, then no. Going back to the events example, if something is a stream of button click events:

var clickWith3 = Rx.fromEvent(input, 'click')
    .withLatest(Rx.Observable.from([1, 2, 3]);

Or if I do foo$.withLatestFrom(cold$,...), bar$.withLatestFrom(cold$,...) and subscribe to foo and bar, will I always see the same values for both?

Not always. Again, if foo and bar are clicks on different buttons for example, then you would see different values. As well, even if they were the same button, if your combination function (the 2nd argument to withLatest) does not return the same result for the same inputs, then you would not see the same values (because it would be called twice, as explained below).

I thought Rx.fromEvent creates cold observables but that is not the case, as mentioned in one of the answers. However, I am still baffled by this behaviour: codepen.io/anon/pen/NqQMJR?editors=101. Different subscriptions get different values from the same observable. Wasn't the click event shared?

I'll point you to this great answer by Enigmativity to a question I had about the same behaviour. That answer will explain it a lot better than I can, but the gist of it is that the source (the click event) is "shared", yes, but your operations on it are not. If you want to share not just the click event but also the operation on it, you will need to do so explicitly.

Solution 3

values in your codepen is lazy - nothing happens until something subscribes, at which point it runs through and wires it up. So in your example, although you are subscribing to the same variable, it is creating two different streams; one for each subscribe call.

You can think of values as being a generator of streams for click with that map attached.

.share() on the end of that map would create the behaviour we expect, because it is implicitly subscribing.

Solution 4

It's not an answer for all your questions (I would like to know all of them!) but for sure, all fromEvent Observables are hot. Click seems to be not because it's not "continous" event like mousemove, but anyway subscription to the source (addEventListener or on call) is done only once, when Observable is created. So it's hot. You can see it in source code of the operator here and there - created observable is shared no matter what is the event name or source.

View more solutions

Comments

[user3743222]

And yet, if you map that click to a function, and subscribe twice to that, it goes through the observable chain twice for the same click. Codepen is here : codepen.io/anon/pen/NqQMJR?editors=101

[user3743222]

Or if I do foo$.withLatestFrom(cold$,...), bar$.withLatestFrom(cold$,...) and subscribe to foo and bar, will I always see the same values for both? My question was more if the value taken from cold$ will be the same in both cases. My problem is the semantics of withLatestFrom. I understand that we have two different subscriptions to cold$, so I would expect two different values extracted, depending on time of subscription, which kind of destroy the semantics of the operator, so I was expecting the operator to turn cold$ into a hot stream. But best is to test and I'll post answers soon.

[user3743222]

Erik, added an answer to all those questions. Let me know what you think if you have some time.

[Ivan Kleshnin]

great study. Thank you for sharing this! Could you add an explaination what 'publish operator' and 'share operator' mean?

[user3743222]

The publish operators are the multicast operator with a different kind of subjects passed as parameter. For publish for instance, see here : github.com/Reactive-Extensions/RxJS/blob/master/src/core/lin‌​q/…. share operators are obtained by appending .refCount() to the publish operators. For share for instance, see github.com/Reactive-Extensions/RxJS/blob/master/src/core/lin‌​q/….

[user3743222]

It would still be good to properly document subjects (current documentation can be hard to figure out). The best way I think is if you log a specific question on that subject (no pun intended). It would be out of scope to answer this there and it would allow to add more details. The present answer is already long, and I want to keep it focused on the question. Question could be what are the semantics of the miscellaneous kinds of Rxjs subjects or whatever way you find to formulate it concisely.

[Ivan Kleshnin]

You are right. I've posted a question: stackoverflow.com/questions/34849873/….

[Ivan Kleshnin]

And one more question about "onComplete" here stackoverflow.com/questions/34850043/…. Maybe you've elaborated this for yourself already as well. Operator titles are clear now, thank you.

[int3h]

Slight digression, but what did you use to generate the diagrams in this answer, @user3743222 ? They are quite nice looking and really help enhance the textual explanation. I'd love to be able to create similar ones when documenting my code or explaining things to others.

[user3743222]

Actually this is probably a more complete resource : websequencediagrams.com