RxJS: How to have one Observer process multiple Observables?

Question

I am working with a framework that calls a function I implement. I would like the parameter of this function to be converted to an Observable, and sent through a sequence of Observers. I thought I could use a Subject for this, but it isn't behaving as I expected.

To clarify, I have something like the following code. I thought Option 1 below would work, but so far I am settling for Option 2, which doesn't seem idiomatic at all.

var eventSubject = new Rx.Subject();
var resultSource = eventSubject.map(processEvent);
var subscription = resultSource.subscribe(
    function(event) {
        console.log("got event", event);
    },
    function(e) {
        log.error(e);
    },
    function() {
        console.log('eventSubject onCompleted');
    }
);

// The framework calls this method
function onEvent(eventArray) {

   var eventSource = Rx.Observable.from(eventArray);

   // Option 1: I thought this would work, but it doesn't
   // eventSource.subscribe(eventSubject);

   // Option 2: This does work, but its obviously clunky
  eventSource.subscribe(
      function(event) {
          log.debug("sending to subject");
          eventSubject.onNext(event);
      },
      function(e) {
          log.error(e);
      },
      function() {
          console.log('eventSource onCompleted');
      }
  );
}

Answer 1

As Brandon already explained, subscribing the eventSubject to another observable means subscribing the eventSubjects onNext, onError and onComplete to that observables onNext, onError and onComplete. From your example you seem to only want to subscribe to onNext.

Your subject completes/errros once the first eventSource completes/errors - your eventSubject correctly ignores any further onNext/onError fired on it by subsequent eventSoures.

There are multiple ways to only subscribe to the onNext of any eventSource:

1. Manually subscribe to only onNext.

   resultSource = eventSubject
     .map(processEvent);
   
   eventSource.subscribe(
       function(event) {
           eventSubject.onNext(event);
       },
       function(error) {
           // don't subscribe to onError
       },
       function() {
           // don't subscribe to onComplete
       }
   );
   

2. Use an operator that handles only subscribing to the eventSources onNext/onError for you. This is what Brandon suggested. Keep in mind this also subscribes to the eventSources onError, which you don't seem to want as of your example.

   resultSource = eventSubject
     .mergeAll()
     .map(processEvent);
   
   eventSubject.onNext(eventSource);
   

3. Use an observer that doesn't call the eventSubjects onError/onComplete for the eventSources onError/onComplete. You could simply overwrite the eventSubjects onComplete as a dirty hack, but it's probably better to create a new observer.

   resultSource = eventSubject
     .map(processEvent);
   
   var eventObserver = Rx.Observer.create(
     function (event) {
       eventSubject.onNext(event);
     }
   );
   
   eventSubject.subscribe(eventObserver);
   

Answer 2

It is just that when you subscribe the whole Subject to an observable, you end up subscribing the onComplete event of that observable to the subject. Which means when the observable completes, it will complete your subject. So when you get the next set of events, they do nothing because the subject is already complete.

One solution is exactly what you did. Just subscribe the onNext event to the subject.

A second solution, and arguably more "rx like" is to treat your incoming data as an observable of observables and flatten this observable stream with mergeAll:

var eventSubject = new Rx.Subject(); // observes observable sequences
var resultSource = eventSubject
  .mergeAll() // subscribe to each inner observable as it arrives
  .map(processEvent);
var subscription = resultSource.subscribe(
    function(event) {
        console.log("got event", event);
    },
    function(e) {
        log.error(e);
    },
    function() {
        console.log('eventSubject onCompleted');
    }
);

// The framework calls this method
function onEvent(eventArray) {

   var eventSource = Rx.Observable.from(eventArray);
   // send a new inner observable sequence
   // into the subject
   eventSubject.onNext(eventSource);
}

Update: Here is a running example