function Stream(generator) {
    EventEmitter.call(this);

    // used to detect Highland Streams using isStream(x), this
    // will work even in cases where npm has installed multiple
    // versions, unlike an instanceof check
    this.__HighlandStream__ = true;

    this.id = ('' + Math.random()).substr(2, 6);

    this.paused = true;
    this.ended = false;

    // Old-style node Stream.pipe() checks for writable, and gulp checks for
    // readable. Discussion at https://github.com/caolan/highland/pull/438.
    this.readable = true;
    this.writable = false;

    this._outgoing = new Queue();
    this._observers = [];
    this._destructors = [];
    this._send_events = false;

    this._nil_pushed = false; // Sets to true when a nil has been queued.
    this._explicitly_destroyed = false; // Sets to true when destroy() is called.

    this._request = null;
    this._multiplexer = null;
    this._consumer = null;

    this._generator = generator;
    this._generator_requested = true;
    this._defer_run_generator = false;
    this._run_generator_deferred = false;

    // Signals whether or not a call to write() returned false, and thus we can
    // drain. This is only relevant for streams constructed with _().
    this._can_drain = false;

    var self = this;

    // These are defined here instead of on the prototype
    // because bind is super slow.
    this._push_fn = function (err, x) {
        if (x === nil) {
            // It's possible that next was called before the
            // nil, causing the generator to be deferred. This
            // is allowed since push can be called at any time.
            // We have to cancel the deferred call to preserve the
            // generator contract.
            self._run_generator_deferred = false;
        }

        // This will set _nil_pushed if necessary.
        self._writeOutgoing(err ? new StreamError(err) : x);
    };

    this._next_fn = function (xs) {
        if (self._explicitly_destroyed) {
            return;
        }

        // It's possible to get into a situation where a call to next() is
        // scheduled asynchonously, but before it is run, destroy() is called,
        // usually by a downstream consumer like take(1). The call to next()
        // still completes, and there is nothing the original caller can do can
        // do. We do not want to throw in that situation.
        if (self._nil_pushed) {
            throw new Error('Cannot call next after nil');
        }

        self._generator_requested = true;
        if (xs) {
            xs = self.create(xs);
            var pull = newPullFunction(xs);
            self._generator = newDelegateGenerator(pull);
        }

        if (!self.paused) {
            if (self._defer_run_generator) {
                self._run_generator_deferred = true;
            }
            else {
                _.setImmediate(function () {
                    self._runGenerator();
                });
            }
        }
    };

    this.on('newListener', function (ev) {
        if (ev === 'data') {
            self._send_events = true;
            _.setImmediate(bindContext(self.resume, self));
        }
        else if (ev === 'end') {
            // this property avoids us checking the length of the
            // listners subscribed to each event on each _send() call
            self._send_events = true;
        }
    });

    // TODO: write test to cover this removeListener code
    this.on('removeListener', function (ev) {
        if (ev === 'end' || ev === 'data') {
            var end_listeners = self.listeners('end').length;
            var data_listeners = self.listeners('data').length;
            if (end_listeners + data_listeners === 0) {
                // stop emitting events
                self._send_events = false;
            }
        }
    });
}
Actual Stream constructor wrapped the the main exported function