RxJS multiple signatures for Observable.Interval

Question

Is there any solution similar to the following for RxJS? Is it possible to activate OnNexts subscribers for different threads in Rx?

PS My first, naive approach (in CoffeeScript) obviously failed:

hObs = Rx.Observable.interval(35000) .startWith(-1) .select(moment().format("D MMMM, HH:mm:ss")) .publish() hObs.subscribe((x)->console.log(x)) hObs.connect() hObs.subscribe((x)->console.log(x, 1)) hObs.connect() 

The second subscription does not return anything for an interval of 35 seconds, etc.

Answer 1

.select expects a function, not a value. The following works:

 (function() { var list = document.getElementById("list"); var stream = Rx.Observable.interval(35000) .startWith(-1) .select(function(){ return moment().format("D MMMM, HH:mm:ss") }); stream.subscribe(function(value) { var li = document.createElement("li"); li.innerHTML = "subscriber 1: "+value; list.appendChild(li); }); stream.subscribe(function(value) { var li = document.createElement("li"); li.innerHTML = "subscriber 2: "+value; list.appendChild(li); }); })(); 

http://jsfiddle.net/9EjSQ/43/

Note that you do not need to call connect() twice, as a rule, you only call it once. It is better to use an "automatic" connect() by calling .publish().refCount() at the end of the chain you are observing. This created a ConnectableObservable, in other words, a "hot observable" (see https://github.com/Reactive-Extensions/RxJS/blob/master/doc/gettingstarted/creating.md#cold-vs-hot-observables ). In this case, we did not need hot observation.

In coffeescript:

 list = document.getElementById("list") stream = Rx.Observable.interval(35000) .startWith(-1) .select(-> moment().format("D MMMM, HH:mm:ss")) stream.subscribe((value) -> li = document.createElement("li") li.innerHTML = "subscriber 1: " + value list.appendChild(li) ) stream.subscribe((value) -> li = document.createElement("li") li.innerHTML = "subscriber 2: " + value list.appendChild(li) ) 

http://jsfiddle.net/9EjSQ/44/