RxJs 动态添加来自另一个事件发射器的事件

Question

我有一个来自EventEmitter的可观察量，它实际上只是一个http连接，流事件。

有时，我必须断开与基础流的连接并重新连接。我不确定如何使用 rxjs 处理这个问题。

我

不确定我是否可以完成一个源，然后动态地将其他"源"添加到源中，或者我是否必须像我在最底部那样做一些事情。

var Rx = require('rx'),
    EventEmitter = require('events').EventEmitter;
  var eventEmitter = new EventEmitter();
  var eventEmitter2 = new EventEmitter();
  var source = Rx.Observable.fromEvent(eventEmitter, 'data')
  var subscription = source.subscribe(function (data) {
    console.log('data: ' + data);
  });
  setInterval(function() {
    eventEmitter.emit('data', 'foo');
  }, 500);
  // eventEmitter stop emitting data, underlying connection closed
  // now attach seconds eventemitter (new connection)
  // something like this but obvouisly doesn't work
  source
    .fromEvent(eventEmitter2, 'data')

Puesdo 代码更像是我正在做的事情，我在关闭第一个流连接之前正在创建第二个流连接，所以我不会"丢失"任何数据。在这里，我不确定如何在不"丢失"记录的情况下停止可观察量，因为由于缓冲区未调用onNext。

  var streams = [], notifiers = [];
  // create initial stream
  createNewStream();
  setInterval(function() {
    if (params of stream have changed) createNewStream();
  }, $1minutes / 3);
  function createNewStream() {
    var stream = new eventEmitterStream();
    stream.once('connected', function() {
      stopOthers();
      streams.push(stream);
      createSource(stream, 'name', 'id');
    });
  }
  function stopOthers() {
    while(streams.length > 0) {
      streams.pop().stop(); // stop the old stream
    }
    while(notifiers.length > 0) {
      // if i call this, the buffer may lose records, before onNext() called
      //notifiers.pop()(Rx.Notification.createOnCompleted());
    }
  }
  function createObserver(tag) {
    return Rx.Observer.create(
      function (x) {
        console.log('Next: ', tag, x.length, x[0], x[x.length-1]);
      },
      function (err) {
        console.log('Error: ', tag, err);
      },
      function () {
        console.log('Completed', tag);
      });
  }
  function createSource(stream, event, id) {
    var source = Rx.Observable
      .fromEvent(stream, event)
      .bufferWithTimeOrCount(time, max);
    var subscription = source.subscribe(createObserver(id));
    var notifier = subscription.toNotifier();
    notifiers.push(notifier);
  }

Answer 1

首先，您需要确保可以从以前的"死"发射器中删除所有侦听器。否则，您将创建一个泄漏的应用程序。

似乎您知道事件发射器已死亡的唯一方法是观察频率，除非您有一个在错误或完成时触发的事件（用于断开连接）。后者更可取。

无论如何，Rx的秘诀是确保将您的数据流创建和拆卸包装在您的可观察对象中。如果将发射器的创建包装在您的可观察量中，以及将其拆除的方法，您将能够使用很棒的东西，例如retry运算符来重新创建该可观察量。

因此，如果您无法知道它是否死亡，并且想要重新连接它，则可以使用如下所示的内容：

// I'll presume you have some function to get an EventEmitter that 
// is already set up
function getEmitter() {
  var emitter = new EventEmitter();
  setInterval(function(){
    emitter.emit('data', 'foo');
  }, 500)
  return emitter;
}

var emitterObservable = Observable.create(function(observer) {
  // setup the data stream
  var emitter = getEmitter();
  var handler = function(d) {
    observer.onNext(d);
  };
  emitter.on('data', handler);
  return function() {
    // tear down the data stream in your disposal function
    emitter.removeListener('on', handler);
  };
});
// Now you can do Rx magic!
emitterObservable
  // if it doesn't emit in 700ms, throw a timeout error
  .timeout(700)
  // catch all* errors and retry
  // this means the emitter will be torn down and recreated 
  // if it times out!
  .retry()
  // do something with the values
  .subscribe(function(x) { console.log(x); });

* 注意：重试会捕获所有错误，因此您可能需要在其上方添加一个catch来处理非超时错误。轮到你了。