javascript - RxJS:如何让一个 Observer 处理多个 Observables？

Question

我正在使用调用我实现的函数的框架。我希望将此函数的参数转换为 Observable，并通过一系列 Observers 发送。我以为我可以为此使用 Subject，但它的行为并不像我预期的那样。

为了澄清，我有类似下面的代码。我认为下面的 Option 1 会起作用，但到目前为止，我正在接受 Option 2，这看起来一点也不符合习惯。

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

正如 Brandon 已经解释过的，将 eventSubject 订阅到另一个 observable 意味着将 eventSubjects onNext、onError 和 onComplete 订阅到那个 observable onNext、onError 和 onComplete。从您的示例来看，您似乎只想订阅 onNext。

一旦第一个 eventSource 完成/出错，您的主题就会完成/出错 - 您的 eventSubject 会正确地忽略后续 eventSoures 对其触发的任何进一步的 onNext/onError。

有多种方法可以只订阅任何事件源的onNext:

1. 仅手动订阅 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. 使用只为您处理订阅 eventSources onNext/onError 的运算符。这是布兰登的建议。 请记住，这也会订阅 eventSources onError，在您的示例中您似乎不想要它。

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

3. 使用不为 eventSources onError/onComplete 调用 eventSubjects onError/onComplete 的观察者。您可以简单地覆盖 eventSubjects onComplete 作为一个肮脏的 hack，但创建一个新的观察者可能更好。

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