Generalize RxJs Observable over an EventStream, specifying event type

sse.js

var rx = require('rx')

module.exports = {
/**
   * This method wraps an EventSource as an observable sequence.
   * 
   * Generalizes https://github.com/Reactive-Extensions/RxJS-DOM/blob/master/src/eventsource.js
   * by allowing different Observables on the same EventSource, based on the event type.
   *
   * Copyright Microsoft Open Technologies
   * 
   * Licensed under the Apache License, Version 2.0 (the "License"); 
   * you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0                                                                                              *
   * Unless required by applicable law or agreed to in writing, 
   * software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and limitations under the License.
   * 
   * @param {EventSource} source The event source
   * @param {String} event identifier of the event
   * @param {Observer} [openObserver] An optional observer for the 'open' event for the server side event.
   * @returns {Observable} An observable sequence which represents the data from a server-side event.
   */
  fromEventSource : function (source, event, openObserver) {

    return new rx.AnonymousObservable(function (observer) {
     
      function onOpen(e) {
        openObserver.onNext(e);
        openObserver.onCompleted();
        source.removeEventListener('open', onOpen, false);
      }

      function onError(e) {
        if (e.readyState === EventSource.CLOSED) {
          observer.onCompleted();
        } else {
          observer.onError(e);
        }
      }

      function onMessage(e) {
        observer.onNext(e);
      }

      openObserver && source.addEventListener('open', onOpen, false);
      source.addEventListener('error', onError, false);
      source.addEventListener(event, onMessage, false);

      return function () {
        source.removeEventListener('error', onError, false);
        source.removeEventListener(event, onMessage, false);
      };
    });
  }
  
};