javascript - RxJS: How to have one Observer process multiple Observables? - Stack Overflow

admin管理员组

文章数量:1314573

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');
      }
  );
}

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');
      }
  );
}

• javascript
• reactive-programming
• rxjs

Share Improve this question Follow asked Dec 14, 2015 at 20:10 JBCPJBCP 13.5k99 gold badges7575 silver badges112112 bronze badges 7

• is that onEvent handler something you register yourself? – user3743222 Commented Dec 14, 2015 at 20:18
• In any case, what I can think about uses Subject.create(observer, observable) and does not really result in something less clunky as the observer you will pass will do exactly the same than the one you passed to the eventSource.subscribe, so let's see other people's propositions. – user3743222 Commented Dec 14, 2015 at 20:38
• @user3743222 - onEvent is a function I write, my framework (loopback) knows the function exists and calls it from its own code. I can't use the fromEvent() or fromEventPattern() methods since they don't match up to the framework's method of registering handlers. – JBCP Commented Dec 14, 2015 at 20:42
• ok, my idea was to write your own fromLoopbackEvent, but that will not likely result in less code. You can have a look at the implementation of Rx.DOM.fromWebSocket for an example of using the Subject.create(observer, observable) form. Or the documentation : github./Reactive-Extensions/RxJS/blob/master/doc/api/…. The point is you need to specify the observer, there is probably no default observer. So doing it your way, or the other way is about the same I think but let's see. – user3743222 Commented Dec 14, 2015 at 20:48
• The Rx.Observable.prototype.subscribe expects an observer or three functions, and a subject is not an observer. So alternatively you might try to convert the subject to a proper observer with eventSource.subscribe(eventSubject.asObserver()). Cf.github./Reactive-Extensions/RxJS/blob/master/doc/api/core/…. The default behaviour is an onNext method which sends to the observable linked to the subject. – user3743222 Commented Dec 14, 2015 at 21:07

 |  Show 2 more ments

2 Answers 2

Sorted by: Reset to default 3

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 pletes/errros once the first eventSource pletes/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);
   

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 pletes, it will plete your subject. So when you get the next set of events, they do nothing because the subject is already plete.

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 ining 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

本文标签： javascriptRxJS How to have one Observer process multiple ObservablesStack Overflow