text: var Rx = require('./dist/rx.all');
  require('./dist/rx.testing');
  
  // Add specific Node functions
  var EventEmitter = require('events').EventEmitter,
    Observable = Rx.Observable;
  
  Rx.Node = {
    /**
     * @deprecated Use Rx.Observable.fromCallback from rx.async.js instead.
     *
     * Converts a callback function to an observable sequence.
     *
     * @param {Function} func Function to convert to an asynchronous function.
     * @param {Scheduler} [scheduler] Scheduler to run the function on. If not specified, defaults to Scheduler.timeout.
     * @param {Mixed} [context] The context for the func parameter to be executed.  If not specified, defaults to undefined.
     * @param {Function} [selector] A selector which takes the arguments from the event handler to produce a single item to yield on next.
     * @returns {Function} Asynchronous function.
     */
    fromCallback: function (func, scheduler, context, selector) {
      return Observable.fromCallback(func, scheduler, context, selector);
    },
  
    /**
     * @deprecated Use Rx.Observable.fromNodeCallback from rx.async.js instead.
     *
     * Converts a Node.js callback style function to an observable sequence.  This must be in function (err, ...) format.
     *
     * @param {Function} func The function to call
     * @param {Scheduler} [scheduler] Scheduler to run the function on. If not specified, defaults to Scheduler.timeout.
     * @param {Mixed} [context] The context for the func parameter to be executed.  If not specified, defaults to undefined.
     * @param {Function} [selector] A selector which takes the arguments from the event handler to produce a single item to yield on next.
     * @returns {Function} An async function which when applied, returns an observable sequence with the callback arguments as an array.
     */
    fromNodeCallback: function (func, scheduler, context, selector) {
      return Observable.fromNodeCallback(func, scheduler, context, selector);
    },
  
    /**
     * @deprecated Use Rx.Observable.fromNodeCallback from rx.async.js instead.
     *
     * Handles an event from the given EventEmitter as an observable sequence.
     *
     * @param {EventEmitter} eventEmitter The EventEmitter to subscribe to the given event.
     * @param {String} eventName The event name to subscribe
     * @param {Function} [selector] A selector which takes the arguments from the event handler to produce a single item to yield on next.
     * @returns {Observable} An observable sequence generated from the named event from the given EventEmitter.  The data will be returned as an array of arguments to the handler.
     */
    fromEvent: function (eventEmitter, eventName, selector) {
      return Observable.fromEvent(eventEmitter, eventName, selector);
    },
  
    /**
     * Converts the given observable sequence to an event emitter with the given event name.
     * The errors are handled on the 'error' event and completion on the 'end' event.
     * @param {Observable} observable The observable sequence to convert to an EventEmitter.
     * @param {String} eventName The event name to emit onNext calls.
     * @returns {EventEmitter} An EventEmitter which emits the given eventName for each onNext call in addition to 'error' and 'end' events.
     *   You must call publish in order to invoke the subscription on the Observable sequuence.
     */
    toEventEmitter: function (observable, eventName, selector) {
      var e = new EventEmitter();
      selector || (selector = function (x) { return x; });
  
      // Used to publish the events from the observable
      e.publish = function () {
        e.subscription = observable.subscribe(
          function (x) {
            e.emit(eventName, selector(x));
          },
          function (err) {
            e.emit('error', err);
          },
          function () {
            e.emit('end');
          });
      };
  
      return e;
    },
  
    /**
     * Converts a flowing stream to an Observable sequence.
     * @param {Stream} stream A stream to convert to a observable sequence.
     * @param {String} [finishEventName] Event that notifies about closed stream. ("end" by default)
     * @returns {Observable} An observable sequence which fires on each 'data' event as well as handling 'error' and finish events like `end` or `finish`.
     */
    fromStream: function (stream, finishEventName) {
      return Observable.create(function (observer) {
        function dataHandler (data) {
          observer.onNext(data);
        }
  
        function errorHandler (err) {
          observer.onError(err);
        }
  
        function endHandler () {
          observer.onCompleted();
        }
  
        if (!finishEventName) {
          finishEventName = 'end';
        }
  
        stream.addListener('data', dataHandler);
        stream.addListener('error', errorHandler);
        stream.addListener(finishEventName, endHandler);
  
        return function () {
          stream.removeListener('data', dataHandler);
          stream.removeListener('error', errorHandler);
          stream.removeListener(finishEventName, endHandler);
        };
      }).publish().refCount();
    },
  
    /**
     * Converts a flowing readable stream to an Observable sequence.
     * @param {Stream} stream A stream to convert to a observable sequence.
     * @returns {Observable} An observable sequence which fires on each 'data' event as well as handling 'error' and 'end' events.
     */
    fromReadableStream: function (stream) {
      return this.fromStream(stream, 'end');
    },
  
    /**
     * Converts a flowing writeable stream to an Observable sequence.
     * @param {Stream} stream A stream to convert to a observable sequence.
     * @returns {Observable} An observable sequence which fires on each 'data' event as well as handling 'error' and 'finish' events.
     */
    fromWritableStream: function (stream) {
      return this.fromStream(stream, 'finish');
    },
  
    /**
     * Converts a flowing transform stream to an Observable sequence.
     * @param {Stream} stream A stream to convert to a observable sequence.
     * @returns {Observable} An observable sequence which fires on each 'data' event as well as handling 'error' and 'finish' events.
     */
    fromTransformStream: function (stream) {
      return this.fromStream(stream, 'finish');
    },
  
    /**
     * Writes an observable sequence to a stream
     * @param {Observable} observable Observable sequence to write to a stream.
     * @param {Stream} stream The stream to write to.
     * @param {String} [encoding] The encoding of the item to write.
     * @returns {Disposable} The subscription handle.
     */
    writeToStream: function (observable, stream, encoding) {
      return observable.subscribe(
        function (x) {
          stream.write(x, encoding);
        },
        function (err) {
          stream.emit('error', err);
        }, function () {
          // Hack check because STDIO is not closable
          !stream._isStdio && stream.end();
        });
    }
  };
  
  module.exports = Rx;
  