title: RxJs Dynamically add events from another EventEmitter
body: <p>I have an Observable coming from an EventEmitter which is really just a http connection, streaming events.</p>
  
  <p>Occasionally I have to disconnect from the underlying stream and reconnect. I am not sure how to handle this with rxjs.</p>
  
  <p>I am not sure if i can complete a source and then dynamically add other "source" to the source, or if I have to do something like i have at the very bottom.  </p>
  
  <pre><code>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')
  </code></pre>
  
  <p>Puesdo code that is more of what i am doing, I am creating a second stream connection before I close the first, so i don't "lose" any data. Here i am not sure how to stop the Observable without "losing" records due to onNext not being called due to the buffer.</p>
  
  <pre><code>  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 &gt; 0) {
        streams.pop().stop(); // stop the old stream
      }
  
      while(notifiers.length &gt; 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);
    }
  </code></pre>