How to buffer stream using fromWebSocket Subject

Question

This RxJava buffer example (with marble chart!) describes the desired result perfectly:

collect items in buffers during the bursty periods and emit them at the end of each burst, by using the debounce operator to emit a buffer closing indicator to the buffer operator

Edit: having reviewed How to create a RxJS buffer that groups elements in NodeJS but that does not rely on forever running interval?, my issue appears related to using a Subject as opposed to straight Observable.

Using the socket stream to generate window close events (as follows) results in 2 sockets opened and no events streaming out:

ws = Rx.DOM.fromWebSocket(wsURI, null, wsOpenObserver, wsCloseObserver);
var closer = ws.flatMapFirst(Rx.Observable.timer(250));
ws.buffer(closer)
    .subscribe(function(e) { console.log(e, 'socket messages');});

Let's start with 'how to reference the source observable itself in the bufferClosingSelector.' I find no examples on this subject. — mygzi
That was helpful - my issue turns out to be one layer deeper. See edits. — mygzi
that should allow to discriminate what is happening. My intuition tells me there is a hot vs. cold issue here, so try to add .share()to ws. Hopefully only one socket is created then. Then tap ws to see what is emitted, i.e. something like wsRx.DOM.fromWebSocket(wsURI, null, wsOpenObserver, wsCloseObserver).tap(*log_fn*).share(). — user3743222

user3743222 user3743222 · Accepted Answer · 2015-11-04T19:20:40

Summarizing the findings issue here :

Rx.DOM.fromWebSocket returns a Rx.subject which wraps around the websocket. That subject is made from one observer and one observable (via new Rx.Subject(observer, observable). From what I understood, that observer allows to write to the socket via its onNext method, while the observable allows to read from the socket.
you always read that subjects are hot sources, but apparently here that only means that the observer will immediately push its value to the subject which here pushes it to the socket. In normal cases(new Rx.Subject()), the default observer and observable are so that the observable listens to the observer, hence the default observable is hot. Here however, the observable is a cold source and then any subscription will reexecute the callback creating another websocket. Hence the creation of two sockets.
this does not happen for instance with Rx.dom.fromEvent because the created (cold) observable is shared (via publish().refCount()).
thus by doing the same here, the duplication issue can be solved. That means in this particular case, use in your code ws = Rx.DOM.fromWebSocket(wsURI, null, wsOpenObserver, wsCloseObserver).share();, share being an alias for publish().refCount().
I have to wonder whether that behaviour of Rx.DOM.fromWebSocket should be reported as a bug

Code for both methods:

How to buffer stream using fromWebSocket Subject

2 Answers