Sharing cold and hot observables

Question

I'm confused by the behavior of a shared stream that is created using Rx.Observable.just.

For example:

var log = function(x) { console.log(x); };

var cold = Rx.Observable
  .just({ foo: 'cold' });

cold.subscribe(log); // <-- Logs three times
cold.subscribe(log);
cold.subscribe(log);

var coldShare = Rx.Observable
  .just({ foo: 'cold share' })
  .share();

coldShare.subscribe(log); // <-- Only logs once
coldShare.subscribe(log);
coldShare.subscribe(log);

Both streams only emit one event, but the un-shared one can be subscribed to three times. Why is this?

I need to "fork" a stream but share its value (and then combine the forked streams).

How can I share the value of a stream but also subscribe to it multiple times?

I realize that this is probably related to the concept of "cold" and "hot" observables. However:

• Is the stream created by Rx.Observable.just() cold or hot?
• How is one supposed to determine the answer to the previous question?

Answer 1

Is the stream created by Rx.Observable.just() cold or hot?

Cold.

How is one supposed to determine the answer to the previous question?

I guess the documentation is the only guide.

How can I share the value of a stream but also subscribe to it multiple times?

You are looking for the idea of a connectable observable. By example:

var log = function(x) { console.log(x); };
var coldShare = Rx.Observable
  .just({ foo: 'cold share' })
  .publish();

coldShare.subscribe(log); // Does nothing
coldShare.subscribe(log); // Does nothing
coldShare.subscribe(log); // Does nothing

coldShare.connect(); // Emits one value to its three subscribers (logs three times)

var log = function(x) {
  document.write(JSON.stringify(x));
  document.write("<br>");
};

var coldShare = Rx.Observable
  .just({ foo: 'cold share' })
  .publish();

coldShare.subscribe(log); // <-- Only logs once
coldShare.subscribe(log);
coldShare.subscribe(log);

coldShare.connect();

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/4.0.7/rx.all.min.js"></script>

The example above logs three times. Using publish and connect, you essentially "pause" the observable until the call to connect.

See also:

• How do I share an observable with publish and connect?
• Are there 'hot' and 'cold' operators?

Answer 2

I don-t understand your first question, but about the last one, as I have been having problem getting that one too:

• Rxjs implementation of Observables/Observers is based on the observer pattern, which is similar to the good old callback mechanism.

• To exemplify, here is the basic form of creating an observable (taken from the doc at https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/create.md)

  var source = Rx.Observable.create(function (observer) {
      observer.onNext(42);
      observer.onCompleted();
  
      // Note that this is optional, you do not have to return this if you require no cleanup
      return function () {
          console.log('disposed');
      };
  });
  

• Rx.Observable.create takes as argument a function (say factory_fn to be original) which takes an observer. Your values are generated by a computation of your choice in the body of factory_fn, and because you have the observer in parameter you can process/push the generated values when you see fit. BUT factory_fn is not executed, it is just registered (like a callback would). It will be called everytime there is a subscribe(observer) on the related observable (i.e. the one returned by Rx.Observable.create(factory_fn).

• Once subscription is done (creation callback called), values flow to your observer according to the logic in the factory function and it remains that way till your observable completes or the observer unsubscribes (supposing you did implement an action to cancel value flow as the return value of factory_fn).
• What that basically means is by default, Rx.Observables are cold.
• My conclusion after using quite a bit of the library, is that unless it is duely documented, the only way to know FOR SURE the temperature of an observable is to eye the source code. Or add a side effect somewhere, subscribe twice and see if the side effect happens twice or only once (which is what you did). That, or ask on stackoverflow.
• For instance, Rx.fromEvent produce hot observables, as you can see from the last line in the code (return new EventObservable(element, eventName, selector).publish().refCount();). (code here : https://github.com/Reactive-Extensions/RxJS/blob/master/src/core/linq/observable/fromevent.js). The publish operator is among those operators which turns a cold observable into a hot one. How that works is out of scope so I won-t detail it here.
• But Rx.DOM.fromWebSocket does not produce hot observables (https://github.com/Reactive-Extensions/RxJS-DOM/blob/master/src/dom/websocket.js). Cf. How to buffer stream using fromWebSocket Subject
• Confusion often comes I think from the fact that we conflate the actual source (say stream of button clicks) and its representation (Rx.Observable). It is unfortunate when that happens but what we imagine as hot sources can end up being represented by a cold Rx.Observable.

So, yes, Rx.Observable.just creates cold observables.