index.js

/**
 * Highland: the high-level streams library
 *
 * Highland may be freely distributed under the Apache 2.0 license.
 * http://github.com/caolan/highland
 * Copyright (c) Caolan McMahon
 *
 */


var inherits = require('util').inherits;
var EventEmitter = require('events').EventEmitter;
var Decoder = require('string_decoder').StringDecoder;

var Queue = require('./queue');
var IntMap = require('./intMap');

// Create quick slice reference variable for speed
var slice = Array.prototype.slice;
var hasOwn = Object.prototype.hasOwnProperty;

// Set up the global object.
var _global = this;
if (typeof global !== 'undefined') {
    _global = global;
}
else if (typeof window !== 'undefined') {
    _global = window;
}

/**
 * The Stream constructor, accepts an array of values or a generator function
 * as an optional argument. This is typically the entry point to the Highland
 * APIs, providing a convenient way of chaining calls together.
 *
 * **Arrays -** Streams created from Arrays will emit each value of the Array
 * and then emit a [nil](#nil) value to signal the end of the Stream.
 *
 * **Generators -** These are functions which provide values for the Stream.
 * They are lazy and can be infinite, they can also be asynchronous (for
 * example, making a HTTP request). You emit values on the Stream by calling
 * `push(err, val)`, much like a standard Node.js callback. Once it has been
 * called, the generator function will not be called again unless you call
 * `next()`. This call to `next()` will signal you've finished processing the
 * current data and allow for the generator function to be called again. If the
 * Stream is still being consumed the generator function will then be called
 * again.
 *
 * You can also redirect a generator Stream by passing a new source Stream
 * to read from to next. For example: `next(other_stream)` - then any subsequent
 * calls will be made to the new source.
 *
 * **Node Readable Stream -** Pass in a Node Readable Stream object to wrap
 * it with the Highland API. Reading from the resulting Highland Stream will
 * begin piping the data from the Node Stream to the Highland Stream.
 *
 * A stream constructed in this way relies on `Readable#pipe` to end the
 * Highland Stream once there is no more data. Not all Readable Streams do
 * this. For example, `IncomingMessage` will only emit `close` when the client
 * aborts communications and will *not* properly call `end`. In this case, you
 * can provide an optional `onFinished` function with the signature
 * `onFinished(readable, callback)` as the second argument.
 *
 * This function will be passed the Readable and a callback that should called
 * when the Readable ends. If the Readable ended from an error, the error
 * should be passed as the first argument to the callback. `onFinished` should
 * bind to whatever listener is necessary to detect the Readable's completion.
 * If the callback is called multiple times, only the first invocation counts.
 * If the callback is called *after* the Readable has already ended (e.g., the
 * `pipe` method already called `end`), it will be ignored.
 *
 * The `onFinished` function may optionally return one of the following:
 *
 * - A cleanup function that will be called when the stream ends. It should
 * unbind any listeners that were added.
 * - An object with the following optional properties:
 *    - `onDestroy` - the cleanup function.
 *    - `continueOnError` - Whether or not to continue the stream when an
 *      error is passed to the callback. Set this to `true` if the Readable
 *      may continue to emit values after errors. Default: `false`.
 *
 * See [this issue](https://github.com/caolan/highland/issues/490) for a
 * discussion on why Highland cannot reliably detect stream completion for
 * all implementations and why the `onFinished` function is required.
 *
 * **EventEmitter / jQuery Elements -** Pass in both an event name and an
 * event emitter as the two arguments to the constructor and the first
 * argument emitted to the event handler will be written to the new Stream.
 *
 * You can pass a mapping hint as the third argument, which specifies how
 * event arguments are pushed into the stream. If no mapping hint is provided,
 * only the first value emitted with the event to the will be pushed onto the
 * Stream.
 *
 * If `mappingHint` is a number, an array of that length will be pushed onto
 * the stream, containing exactly that many parameters from the event. If it's
 * an array, it's used as keys to map the arguments into an object which is
 * pushed to the tream. If it is a function, it's called with the event
 * arguments, and the returned value is pushed.
 *
 * **Promise -** Accepts an ES6 / jQuery style promise and returns a
 * Highland Stream which will emit a single value (or an error). In case you use
 * [bluebird cancellation](http://bluebirdjs.com/docs/api/cancellation.html) Highland Stream will be empty for a cancelled promise.
 *
 * **Iterator -** Accepts an ES6 style iterator that implements the [iterator protocol](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Iteration_protocols#The_.22iterator.22_protocol):
 * yields all the values from the iterator using its `next()` method and terminates when the
 * iterator's done value returns true. If the iterator's `next()` method throws, the exception will be emitted as an error,
 * and the stream will be ended with no further calls to `next()`.
 *
 * **Iterable -** Accepts an object that implements the [iterable protocol](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Iteration_protocols#The_.22iterable.22_protocol),
 * i.e., contains a method that returns an object that conforms to the iterator protocol. The stream will use the
 * iterator defined in the `Symbol.iterator` property of the iterable object to generate emitted values.
 *
 * @id _(source)
 * @section Stream Objects
 * @name _(source)
 * @param {Array | Function | Iterator | Iterable | Promise | Readable Stream | String} source - (optional) source to take values from from
 * @param {Function} onFinished - (optional) a function that detects when the readable completes. Second argument. Only valid if `source` is a Readable.
 * @param {EventEmitter | jQuery Element} eventEmitter - (optional) An event emitter. Second argument. Only valid if `source` is a String.
 * @param {Array | Function | Number} mappingHint - (optional) how to pass the
 * arguments to the callback. Only valid if `source` is a String.
 * @api public
 *
 * // from an Array
 * _([1, 2, 3, 4]);
 *
 * // using a generator function
 * _(function (push, next) {
 *     push(null, 1);
 *     push(err);
 *     next();
 * });
 *
 * // a stream with no source, can pipe node streams through it etc.
 * var through = _();
 *
 * // wrapping a Node Readable Stream so you can easily manipulate it
 * _(readable).filter(hasSomething).pipe(writeable);
 *
 * // wrapping a Readable that may signify completion by emitting `close`
 * // (e.g., IncomingMessage).
 * _(req, function (req, callback) {
 *     req.on('end', callback)
 *         .on('close', callback)
 *         .on('error', callback);
 *
 *     return function () {
 *         req.removeListener('end', callback);
 *         req.removeListener('close', callback);
 *         req.removeListener('error', callback);
 *     };
 * }).pipe(writable);
 *
 * // wrapping a Readable that may emit values after errors.
 * _(req, function (req, callback) {
 *     req.on('error', callback);
 *
 *     return {
 *         onDestroy: function () {
 *             req.removeListener('error', callback);
 *         },
 *         continueOnError: true
 *     };
 * }).pipe(writable);
 *
 * // creating a stream from events
 * _('click', btn).each(handleEvent);
 *
 * // creating a stream from events with a mapping array
 * _('request', httpServer, ['req', 'res']).each(handleEvent);
 * //=> { req: IncomingMessage, res: ServerResponse }
 *
 * // creating a stream from events with a mapping function
 * _('request', httpServer, function(req, res) {
 *     return res;
 * }).each(handleEvent);
 * //=> IncomingMessage
 *
 * // from a Promise object
 * var foo = _($.getJSON('/api/foo'));
 *
 * //from an iterator
 * var map = new Map([['a', 1], ['b', 2]]);
 * var bar = _(map.values()).toArray(_.log);
 * //=> [1, 2]
 *
 * //from an iterable
 * var set = new Set([1, 2, 2, 3, 4]);
 * var bar = _(set).toArray(_.log);
 * //=> [ 1, 2, 3, 4]
 */

/*eslint-disable no-multi-spaces */
var _ = exports = module.exports = __(Stream);
function __(StreamCtor) {
    return function (/*optional*/xs, /*optional*/secondArg, /*optional*/ mappingHint) {
        /*eslint-enable no-multi-spaces */
        var s = null;
        if (_.isUndefined(xs)) {
            // nothing else to do
            s = new StreamCtor();
            s.writable = true;
        }
        else if (_.isStream(xs)) {
            if (!(xs instanceof StreamCtor)) { // different subclass or version
                var ret = new StreamCtor();
                xs.on('error', ret.write.bind(ret));
                s = xs.pipe(ret);
            }
            else {
                s = xs;
            }
        }
        else if (_.isArray(xs)) {
            s = new StreamCtor();
            s._outgoing.enqueueAll(xs);
            s._outgoing.enqueue(_.nil);
        }
        else if (_.isFunction(xs)) {
            s = new StreamCtor(xs);
        }
        else if (_.isObject(xs)) {
            // check to see if we have a readable stream
            if (_.isFunction(xs.on) && _.isFunction(xs.pipe)) {
                var onFinish = _.isFunction(secondArg) ? secondArg : defaultReadableOnFinish;
                s = new StreamCtor();
                s.writable = true;
                pipeReadable(xs, onFinish, s);
                // s has to be writable so that the pipe works
                // return a non-writable stream
                return s.map(function (x) { return x; });
            }
            else if (_.isFunction(xs.then)) {
                // probably a promise
                s = promiseStream(StreamCtor, xs);
            }
            // must check iterators and iterables in this order
            // because generators are both iterators and iterables:
            // their Symbol.iterator method returns the `this` object
            // and an infinite loop would result otherwise
            else if (_.isFunction(xs.next)) {
                //probably an iterator
                return iteratorStream(StreamCtor, xs);
            }
            else if (!_.isUndefined(_global.Symbol) && xs[_global.Symbol.iterator]) {
                //probably an iterable
                return iteratorStream(StreamCtor, xs[_global.Symbol.iterator]());
            }
            else {
                throw new Error(
                    'Object was not a stream, promise, iterator or iterable: ' + (typeof xs)
                );
            }
        }
        else if (_.isString(xs)) {
            var mapper = hintMapper(mappingHint);

            s = new StreamCtor();
            secondArg.on(xs, function () {
                var ctx = mapper.apply(this, arguments);
                s.write(ctx);
            });
        }
        else {
            throw new Error(
                'Unexpected argument type to Stream constructor: ' + (typeof xs)
            );
        }

        return s;
    };
}
/*eslint-enable no-use-before-define */

// ES5 detected value, used for switch between ES5 and ES3 code
var isES5 = (function () {
    'use strict';
    return Function.prototype.bind && !this;
}());


_.isUndefined = function (x) {
    return typeof x === 'undefined';
};

_.isFunction = function (x) {
    return typeof x === 'function';
};

_.isObject = function (x) {
    return typeof x === 'object' && x !== null;
};

_.isString = function (x) {
    return typeof x === 'string';
};

_.isArray = Array.isArray || function (x) {
    return Object.prototype.toString.call(x) === '[object Array]';
};

// setImmediate browser fallback
if (typeof setImmediate === 'undefined') {
    _.setImmediate = function (fn) {
        setTimeout(fn, 0);
    };
}
else {
    // We don't use a direct alias since some tests depend
    // on allowing Sinon.Js to override the global
    // setImmediate.
    _.setImmediate = function (fn) {
        setImmediate(fn);
    };
}

/**
 * The end of stream marker. This is sent along the data channel of a Stream
 * to tell consumers that the Stream has ended. See the example map code for
 * an example of detecting the end of a Stream.
 *
 * Note: `nil` is setup as a global where possible. This makes it convenient
 * to access, but more importantly lets Streams from different Highland
 * instances work together and detect end-of-stream properly. This is mostly
 * useful for NPM where you may have many different Highland versions installed.
 *
 * @id nil
 * @section Utils
 * @name _.nil
 * @api public
 *
 * var map = function (iter, source) {
 *     return source.consume(function (err, val, push, next) {
 *         if (err) {
 *             push(err);
 *             next();
 *         }
 *         else if (val === _.nil) {
 *             push(null, val);
 *         }
 *         else {
 *             push(null, iter(val));
 *             next();
 *         }
 *     });
 * };
 */

// set up a global nil object in cases where you have multiple Highland
// instances installed (often via npm)
if (!_global.nil) {
    _global.nil = {};
}
var nil = _.nil = _global.nil;

/**
 * Transforms a function with specific arity (all arguments must be
 * defined) in a way that it can be called as a chain of functions until
 * the arguments list is saturated.
 *
 * This function is not itself curryable.
 *
 * @id curry
 * @name _.curry(fn, [*arguments])
 * @section Functions
 * @param {Function} fn - the function to curry
 * @param args.. - any number of arguments to pre-apply to the function
 * @returns Function
 * @api public
 *
 * fn = curry(function (a, b, c) {
 *     return a + b + c;
 * });
 *
 * fn(1)(2)(3) == fn(1, 2, 3)
 * fn(1, 2)(3) == fn(1, 2, 3)
 * fn(1)(2, 3) == fn(1, 2, 3)
 */

_.curry = function (fn /* args... */) {
    var args = slice.call(arguments);
    return _.ncurry.apply(this, [fn.length].concat(args));
};

/**
 * Same as `curry` but with a specific number of arguments. This can be
 * useful when functions do not explicitly define all its parameters.
 *
 * This function is not itself curryable.
 *
 * @id ncurry
 * @name _.ncurry(n, fn, [args...])
 * @section Functions
 * @param {Number} n - the number of arguments to wait for before apply fn
 * @param {Function} fn - the function to curry
 * @param args... - any number of arguments to pre-apply to the function
 * @returns Function
 * @api public
 *
 * fn = ncurry(3, function () {
 *     return Array.prototype.join.call(arguments, '.');
 * });
 *
 * fn(1, 2, 3) == '1.2.3';
 * fn(1, 2)(3) == '1.2.3';
 * fn(1)(2)(3) == '1.2.3';
 */

_.ncurry = function (n, fn /* args... */) {
    var largs = slice.call(arguments, 2);
    if (largs.length >= n) {
        return fn.apply(this, largs.slice(0, n));
    }

    return _.partial.apply(this, [_.ncurry, n, fn].concat(largs));
};

/**
 * Partially applies the function (regardless of whether it has had curry
 * called on it). This will always postpone execution until at least the next
 * call of the partially applied function.
 *
 * @id partial
 * @name _.partial(fn, args...)
 * @section Functions
 * @param {Function} fn - function to partial apply
 * @param args... - the arguments to apply to the function
 * @api public
 *
 * var addAll = function () {
 *     var args = Array.prototype.slice.call(arguments);
 *     return foldl1(add, args);
 * };
 * var f = partial(addAll, 1, 2);
 * f(3, 4) == 10
 */

_.partial = function (f /* args... */) {
    var args = slice.call(arguments, 1);
    return function () {
        return f.apply(this, args.concat(slice.call(arguments)));
    };
};

/**
 * Evaluates the function `fn` with the argument positions swapped. Only
 * works with functions that accept two arguments.
 *
 * @id flip
 * @name _.flip(fn, [x, y])
 * @section Functions
 * @param {Function} fn - function to flip argument application for
 * @param x - parameter to apply to the right hand side of f
 * @param y - parameter to apply to the left hand side of f
 * @api public
 *
 * div(2, 4) == 0.5
 * flip(div, 2, 4) == 2
 * flip(div)(2, 4) == 2
 */

_.flip = _.curry(function (fn, x, y) { return fn(y, x); });

/**
 * Creates a composite function, which is the application of function1 to
 * the results of function2. You can pass an arbitrary number of arguments
 * and have them composed. This means you can't partially apply the compose
 * function itself.
 *
 * @id compose
 * @name _.compose(fn1, fn2, ...)
 * @section Functions
 * @api public
 *
 * var add1 = add(1);
 * var mul3 = mul(3);
 *
 * var add1mul3 = compose(mul3, add1);
 * add1mul3(2) == 9
 */

_.compose = function (/*functions...*/) {
    var fns = slice.call(arguments).reverse();
    return _.seq.apply(null, fns);
};

/**
 * The reversed version of [compose](#compose). Where arguments are in the
 * order of application.
 *
 * @id seq
 * @name _.seq(fn1, fn2, ...)
 * @section Functions
 * @api public
 *
 * var add1 = add(1);
 * var mul3 = mul(3);
 *
 * var add1mul3 = seq(add1, mul3);
 * add1mul3(2) == 9
 */

_.seq = function () {
    var fns = slice.call(arguments);
    return function () {
        if (!fns.length) {
            return null;
        }
        var r = fns[0].apply(this, arguments);
        for (var i = 1; i < fns.length; i++) {
            r = fns[i].call(this, r);
        }
        return r;
    };
};

function nop() {
    // Do nothing.
}

function defaultReadableOnFinish(readable, callback) {
    // It's possible that `close` is emitted *before* `end`, so we simply
    // cannot handle that case. See
    // https://github.com/caolan/highland/issues/490 for details.

    // pipe already pushes on end, so no need to bind to `end`.

    // write any errors into the stream
    readable.once('error', callback);

    return function () {
        readable.removeListener('error', callback);
    };
}

function pipeReadable(xs, onFinish, stream) {
    var response = onFinish(xs, streamEndCb);
    var unbound = false;

    var cleanup = null;
    var endOnError = true;

    if (_.isFunction(response)) {
        cleanup = response;
    }
    else if (response != null) {
        cleanup = response.onDestroy;
        endOnError = !response.continueOnError;
    }

    xs.pipe(stream);

    stream.onDestroy(unbind);

    function streamEndCb(error) {
        if (stream._nil_pushed) {
            return;
        }

        if (error) {
            stream.write(new StreamError(error));
        }

        if (error == null || endOnError) {
            unbind();
            stream.end();
        }
    }

    function unbind() {
        if (unbound) {
            return;
        }

        unbound = true;

        if (cleanup) {
            cleanup();
        }

        if (xs.unpipe) {
            xs.unpipe(stream);
        }
    }
}

function newPullFunction(xs) {
    return function pull(cb) {
        xs.pull(cb);
    };
}

function newDelegateGenerator(pull) {
    return function delegateGenerator(push, next) {
        var self = this;
        pull(function (err, x) {
            // Minor optimization to immediately call the
            // generator if requested.
            var old = self._defer_run_generator;
            self._defer_run_generator = true;

            push(err, x);
            if (x !== nil) {
                next();
            }

            self._defer_run_generator = old;
            if (!old && self._run_generator_deferred) {
                self._runGenerator();
            }
        });
    };
}

function promiseStream(StreamCtor, promise) {
    if (_.isFunction(promise['finally'])) { // eslint-disable-line dot-notation
        // Using finally handles also bluebird promise cancellation
        return new StreamCtor(function (push) {
            promise.then(function (value) {
                return push(null, value);
            },
                function (err) {
                    return push(err);
                })['finally'](function () { // eslint-disable-line dot-notation
                    return push(null, nil);
                });
        });
    }
    else {
        // Sticking to promise standard only
        return new StreamCtor(function (push) {
            promise.then(function (value) {
                push(null, value);
                return push(null, nil);
            },
            function (err) {
                push(err);
                return push(null, nil);
            });
        });
    }
}

function iteratorStream(StreamCtor, it) {
    return new StreamCtor(function (push, next) {
        var iterElem, iterErr;
        try {
            iterElem = it.next();
        }
        catch (err) {
            iterErr = err;
        }

        if (iterErr) {
            push(iterErr);
            push(null, _.nil);
        }
        else if (iterElem.done) {
            if (!_.isUndefined(iterElem.value)) {
                // generators can return a final
                // value on completion using return
                // keyword otherwise value will be
                // undefined
                push(null, iterElem.value);
            }
            push(null, _.nil);
        }
        else {
            push(null, iterElem.value);
            next();
        }

    });
}

function hintMapper(mappingHint) {
    var mappingHintType = (typeof mappingHint);
    var mapper;

    if (mappingHintType === 'function') {
        mapper = mappingHint;
    }
    else if (mappingHintType === 'number') {
        mapper = function () {
            return slice.call(arguments, 0, mappingHint);
        };
    }
    else if (_.isArray(mappingHint)) {
        mapper = function () {
            var args = arguments;
            return mappingHint.reduce(function (ctx, hint, idx) {
                ctx[hint] = args[idx];
                return ctx;
            }, {});
        };
    }
    else {
        mapper = function (x) { return x; };
    }

    return mapper;
}

function pipeStream(src, dest, write, end, passAlongErrors) {
    var resume = null;
    var s = src.consume(function (err, x, push, next) {
        var canContinue;
        if (err) {
            if (passAlongErrors) {
                canContinue = write.call(dest, new StreamError(err));
            }
            else {
                src.emit('error', err);
                canContinue = true;
            }
        }
        else if (x === nil) {
            end.call(dest);
            return;
        }
        else {
            canContinue = write.call(dest, x);
        }

        if (canContinue !== false) {
            next();
        }
        else {
            resume = next;
        }
    });

    dest.on('drain', onConsumerDrain);

    // Since we don't keep a reference to piped-to streams,
    // save a callback that will unbind the event handler.
    src.onDestroy(function () {
        dest.removeListener('drain', onConsumerDrain);
    });

    dest.emit('pipe', src);

    s.resume();
    return dest;

    function onConsumerDrain() {
        if (resume) {
            resume();
            resume = null;
        }
    }
}

/**
 * Actual Stream constructor wrapped the the main exported function
 */

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;

    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(self.resume.bind(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;
            }
        }
    });
}
inherits(Stream, EventEmitter);

function _addMethod(proto, topLevel) {
    return function(name, f) {
        proto[name] = f;
        var n = f.length;
        function relevel(coerce) {
            return _.ncurry(n + 1, function () {
                var args = slice.call(arguments);
                var s = coerce(args.pop());
                return f.apply(s, args);
            });
        }
        topLevel[name] = relevel(topLevel);
        topLevel[name]._relevel = relevel;
    };
}

var addMethod = _addMethod(Stream.prototype, _);

function _addMethods(proto, topLevel, methods) {
    for (var p in methods) {
        if (hasOwn.call(methods, p)) {
            _addMethod(proto, topLevel)(p, methods[p]);
        }
    }
}

function _addToplevelMethod(topLevel) {
    return function (name, fn) {
        function relevel(_topLevel) {
            var bound = fn.bind(_topLevel);
            bound._relevel = relevel;
            return bound;
        }
        topLevel[name] = relevel(topLevel);
    };
}

var addToplevelMethod = _addToplevelMethod(_);

function _addToplevelMethods(topLevel, methods) {
    for (var p in methods) {
        if (hasOwn.call(methods, p)) {
            _addToplevelMethod(topLevel)(p, methods[p]);
        }
    }
}

function use(Super, originalTopLevel) {
    return function(methods, toplevelMethods) {
        function Sub() {
            Stream.apply(this, arguments);
        }
        inherits(Sub, Super);

        function topLevel() {
            return __(Sub).apply(null, arguments);
        }
        for (var p in originalTopLevel) {
            if (hasOwn.call(originalTopLevel, p)) {
                var fn = originalTopLevel[p];
                topLevel[p] = (typeof fn._relevel === 'function') ? fn._relevel(topLevel) : fn;
            }
        }
        _addMethods(Sub.prototype, topLevel, methods || {});
        _addToplevelMethods(topLevel, toplevelMethods || {});
        topLevel.use = use(Sub, topLevel);

        return topLevel;
    };
}

_.use = use(Stream, _);

/**
 * Creates a stream that sends a single value then ends.
 *
 * @id of
 * @name _.of(x)
 * @param x - the value to send
 * @section Utils
 *
 * _.of(1).toArray(_.log); // => [1]
 */

addToplevelMethod('of', function (x) {
    return _([x]);
});

/**
 * Creates a stream that sends a single error then ends.
 *
 * @id fromError
 * @name _.fromError(err)
 * @param error - the error to send
 * @section Utils
 *
 * _.fromError(new Error('Single Error')).toCallback(function (err, result) {
 *     // err contains Error('Single Error') object
 * }
 */

addToplevelMethod('fromError', function (error) {
    return _(function (push) {
        push(error);
        push(null, _.nil);
    });
});

Stream.prototype.create = function () {
    return __(this.constructor).apply(null, arguments);
};

Stream.prototype.createChild = function createChild(/*varargs*/) {
    var child = this.create.apply(this, arguments);
    child.onDestroy(this.destroy.bind(this));
    return child;
};

/**
 * Used as an Error marker when writing to a Stream's incoming buffer
 */

function StreamError(err) {
    this.__HighlandStreamError__ = true;
    this.error = err;
}

/**
 * Returns true if `x` is a Highland Stream.
 *
 * @id isStream
 * @section Utils
 * @name _.isStream(x)
 * @param x - the object to test
 * @returns {Boolean}
 * @api public
 *
 * _.isStream('foo')  // => false
 * _.isStream(_([1,2,3]))  // => true
 */

_.isStream = function (x) {
    return _.isObject(x) && !!x.__HighlandStream__;
};

_._isStreamError = function (x) {
    return _.isObject(x) && !!x.__HighlandStreamError__;
};

_._isStreamRedirect = function (x) {
    return _.isObject(x) && !!x.__HighlandStreamRedirect__;
};

/**
 * Sends errors / data to consumers, observers and event handlers
 */

Stream.prototype._send = function (token) {
    // console.log(this.id, '_send', token, this._send_events);

    var err = null,
        x;

    if (_._isStreamError(token)) {
        err = token.error;
        this.readable = false;
    }
    else {
        x = token;
    }

    if (x === nil) {
        // Per https://nodejs.org/docs/v0.8.28/api/stream.html#stream_stream_readable
        // streams stop being readable when they end or are destroyed
        this.readable = false;
    }

    if (this._request) {
        // Allow pull to be called within the callback function.
        var cb = this._request;
        this._request = null;

        // If we have a request, then it was from a pull() and we need to
        // pause.
        this.pause();
        cb(err, x);
    }

    // this._observers may be changed from under us,
    // so we keep a copy.
    var observers = this._observers;
    for (var j = 0, len2 = observers.length; j < len2; j++) {
        observers[j].write(token);
    }

    if (this._send_events) {
        if (err) {
            this.emit('error', err);
        }
        else if (x === nil) {
            this.emit('end');
        }
        else {
            this.emit('data', x);
        }
    }

    if (x === nil) {
        this._onEnd();
    }
};

Stream.prototype._onEnd = function _onEnd() {
    if (this.ended) {
        return;
    }

    this.pause();

    this.ended = true;

    // Don't use _removeObserver for efficiency reasons.
    var i, len, observer;
    for (i = 0, len = this._observers.length; i < len; i++) {
        observer = this._observers[i];
        if (observer.source === this) {
            observer.source = null;
        }
    }

    for (i = 0, len = this._destructors.length; i < len; i++) {
        this._destructors[i].call(this);
    }

    this._outgoing.clear();
    this._observers = [];
    this._destructors = [];

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

    this._generator = null;
    this._generator_requested = false;
    this._defer_run_generator = false;
    this._run_generator_deferred = false;
};

/**
 * Pauses the stream. All Highland Streams start in the paused state.
 *
 * It is unlikely that you will need to manually call this method.
 *
 * @id pause
 * @section Stream Objects
 * @name Stream.pause()
 * @api public
 *
 * var xs = _(generator);
 * xs.pause();
 */

addMethod('pause', function () {
    //console.log(['pause', this.id]);
    this.paused = true;
});

/*
 * Emit as many buffered token as possible, but not to exceed num.
 * If num is null, then emit as much as possible.
 */

Stream.prototype._emitNext = function (num) {
    var emitted = 0;

    while ((num == null || emitted < num) &&
            this._outgoing.length &&
            !this.paused) {
        this._send(this._outgoing.dequeue());
        emitted++;
    }
};

/**
 * Resumes a paused Stream. This will either read from the Stream's incoming
 * buffer or request more data from an upstream source. Never call this method
 * on a stream that has been consumed (via a call to [consume](#consume) or any
 * other transform).
 *
 * @id resume
 * @section Stream Objects
 * @name Stream.resume()
 * @api public
 *
 * var xs = _(generator);
 * xs.resume();
 */

addMethod('resume', function () {
    // console.log(this.id, 'resume', this.paused);
    if (!this.paused ||
            (this._consumer && this._consumer.paused) ||
            (this._multiplexer && this._multiplexer.paused)) {
        return;
    }

    this.paused = false;

    // Emit all pending tokens in _outgoing.
    if (this._outgoing.length) {
        this._emitNext();

        // The emit may have caused the stream to pause.
        if (this.paused) {
            return;
        }
    }

    if (this._generator) {
        this._runGenerator();
    }
    else {
        // perhaps a node stream is being piped in
        this.emit('drain');
    }
});

/**
 * Ends a Stream. This is the same as sending a [nil](#nil) value as data.
 * You shouldn't need to call this directly, rather it will be called by
 * any [Node Readable Streams](http://nodejs.org/api/stream.html#stream_class_stream_readable)
 * you pipe in.
 *
 * Only call this function on streams that were constructed with no source
 * (i.e., with `_()`).
 *
 * @id end
 * @section Stream Objects
 * @name Stream.end()
 * @aprototype.ppi public
 *
 * mystream.end();
 */

addMethod('end', function () {
    if (this._nil_pushed) {
        // Allow ending multiple times.
        return;
    }

    this.write(nil);
});

/**
 * Pipes a Highland Stream to a [Node Writable Stream](http://nodejs.org/api/stream.html#stream_class_stream_writable)
 * (Highland Streams are also Node Writable Streams). This will pull all the
 * data from the source Highland Stream and write it to the destination,
 * automatically managing flow so that the destination is not overwhelmed
 * by a fast source.
 *
 * Users may optionally pass an object that may contain any of these fields:
 *
 * - `end` - Ends the destination when this stream ends. Default: `true`. This
 *   option has no effect if the destination is either `process.stdout` or
 *   `process.stderr`. Those two streams are never ended.
 *
 * Like [Readable#pipe](https://nodejs.org/api/stream.html#stream_readable_pipe_destination_options),
 * this function will throw errors if there is no `error` handler installed on
 * the stream. Use [through](#through) if you are piping to another Highland
 * stream and want errors as well as values to be propagated.
 *
 * This function returns the destination so you can chain together pipe calls.
 *
 * @id pipe
 * @section Consumption
 * @name Stream.pipe(dest, options)
 * @param {Writable Stream} dest - the destination to write all data to
 * @param {Object} options - (optional) pipe options.
 * @api public
 *
 * var source = _(generator);
 * var dest = fs.createWriteStream('myfile.txt')
 * source.pipe(dest);
 *
 * // chained call
 * source.pipe(through).pipe(dest);
 */

addMethod('pipe', function (dest, options) {
    options = options || {};

    // stdout and stderr are special case writables that cannot be closed
    var canClose = dest !== process.stdout && dest !== process.stderr && options.end !== false;

    var end;
    if (canClose) {
        end = dest.end;
    }
    else {
        end = nop;
    }

    return pipeStream(this, dest, dest.write, end, false);
});

/**
 * Destroys a stream by unlinking it from any consumers and sources. This will
 * stop all consumers from receiving events from this stream and removes this
 * stream as a consumer of any source stream.
 *
 * This function calls end() on the stream and unlinks it from any piped-to streams.
 *
 * @id destroy
 * @section Stream Objects
 * @name Stream.destroy()
 * @api public
 */

addMethod('destroy', function () {
    this._explicitly_destroyed = true;

    // Already destroyed.
    if (this.ended) {
        return;
    }

    this.readable = this.writable = false;

    if (!this._nil_pushed) {
        this.end();
    }

    this._onEnd();
});

Stream.prototype._writeOutgoing = function _writeOutgoing(token) {
    if (this._explicitly_destroyed) {
        return;
    }

    if (this._nil_pushed) {
        throw new Error('Cannot write to stream after nil');
    }

    if (token === _.nil) {
        this._nil_pushed = true;
    }

    //console.log('_writeOutgoing', token, this.id);
    if (token === nil || _._isStreamError(token)) {
        // Per https://nodejs.org/docs/v0.8.28/api/stream.html#stream_stream_writable
        // writable should turn false after end is called or an error occurs
        this.writable = false;
    }

    if (this.paused) {
        this._outgoing.enqueue(token);
    }
    else {
        this._send(token);
    }
};

/**
 * Runs the generator function for this Stream. If the generator is already
 * running (it has been called and not called next() yet) then this function
 * will do nothing.
 */

Stream.prototype._runGenerator = function () {
    //console.log(this.id, '_runGenerator');
    // if _generator already running, exit
    if (!this._generator_requested) {
        return;
    }

    if (this._defer_run_generator) {
        this._run_generator_deferred = true;
        return;
    }

    this._defer_run_generator = true;
    do {
        this._run_generator_deferred = false;
        this._generator_requested = false;

        // this._generator(this._push_fn, this._next_fn);

        var gen = this._generator;

        // Special handling for consume generators.
        if (gen.__ConsumeGenerator__) {
            var source = gen.source;
            if (source._outgoing.length) {
                var token = source._outgoing.dequeue();

                // Inline the pull -> resume -> _emitNext call chain.
                source._request = gen.pullCb;
                source.paused = false;
                source._send(token);
            }
            else {
                // Inline pull.
                source._request = gen.pullCb;
                source.resume();
            }
        }
        else {
            gen.call(this, this._push_fn, this._next_fn);
        }
    } while (this._run_generator_deferred);
    this._defer_run_generator = false;
};

/**
 * Removes an observer from this Stream.
 */

Stream.prototype._removeObserver = function (s) {
    this._observers = this._observers.filter(function (o) {
        return o !== s;
    });
    if (s.source === this) {
        s.source = null;
    }
};

/**
 * Consumes values from a Stream (once resumed) and returns a new Stream for
 * you to optionally push values onto using the provided push / next functions.
 *
 * This function forms the basis of many higher-level Stream operations.
 * It will not cause a paused stream to immediately resume, but behaves more
 * like a 'through' stream, handling values as they are read.
 *
 * @id consume
 * @section Transforms
 * @name Stream.consume(f)
 * @param {Function} f - the function to handle errors and values
 * @api public
 *
 * var filter = function (f, source) {
 *     return source.consume(function (err, x, push, next) {
 *         if (err) {
 *             // pass errors along the stream and consume next value
 *             push(err);
 *             next();
 *         }
 *         else if (x === _.nil) {
 *             // pass nil (end event) along the stream
 *             push(null, x);
 *         }
 *         else {
 *             // pass on the value only if the value passes the predicate
 *             if (f(x)) {
 *                 push(null, x);
 *             }
 *             next();
 *         }
 *     });
 * };
 */

addMethod('consume', function (f) {
    var source = this,
        consumer;

    if (this._consumer) {
        throw new Error(
            'Stream already being consumed, you must either fork() or observe()'
        );
    }

    if (this._multiplexer) {
        throw new Error(
            'Stream has been forked. You must either fork() or observe().'
        );
    }

    // This should be a subclass, but prototype resolution is slow, and
    // consume is on the critical path, so we inline it.
    var gen = function () {
        // Important. next may not be called outside of
        // pullCb.
        source.pull(pullCb);
    };

    gen.__ConsumeGenerator__ = true;
    gen.pullCb = pullCb;
    gen.source = source;

    consumer = this._consumer = this.createChild(gen);

    consumer.id = 'consume:' + consumer.id;

    return consumer;

    function pullCb(err, x) {
        // Minor optimization to immediately call the
        // generator if requested.
        if (consumer._defer_run_generator) {
            f(err, x, consumer._push_fn, consumer._next_fn);
        }
        else {
            consumer._defer_run_generator = true;
            f(err, x, consumer._push_fn, consumer._next_fn);
            consumer._defer_run_generator = false;

            if (consumer._run_generator_deferred) {
                consumer._runGenerator();
            }
        }
    }
});

/**
 * Consumes a single item from the Stream. Unlike consume, this function will
 * not provide a new stream for you to push values onto, and it will unsubscribe
 * as soon as it has a single error, value or nil from the source.
 *
 * You probably won't need to use this directly, but it is used internally by
 * some functions in the Highland library.
 *
 * @id pull
 * @section Consumption
 * @name Stream.pull(f)
 * @param {Function} f - the function to handle data
 * @api public
 *
 * xs.pull(function (err, x) {
 *     // do something
 * });
 */

addMethod('pull', function (f) {
    // console.log(this.id, 'pull', this._outgoing.toArray(), this.paused);
    if (f == null) {
        throw new Error('Cannot pull() with a null callback.');
    }

    if (this._request) {
        f(new Error('Cannot service a second pull() request while one is in progress.'));
    }

    // Register the callback and send a message.
    // Don't need to pause. _send will do it for us.
    this._request = f;
    this.resume();
});

/**
 * Writes a value to the Stream. If the Stream is paused it will go into the
 * Stream's incoming buffer, otherwise it will be immediately processed and
 * sent to the Stream's consumers (if any). Returns false if the Stream is
 * paused, true otherwise. This lets Node's pipe method handle back-pressure.
 *
 * You shouldn't need to call this yourself, but it may be called by Node
 * functions which treat Highland Streams as a [Node Writable Stream](http://nodejs.org/api/stream.html#stream_class_stream_writable).
 *
 * Only call this function on streams that were constructed with no source
 * (i.e., with `_()`).

 * @id write
 * @section Stream Objects
 * @name Stream.write(x)
 * @param x - the value to write to the Stream
 * @api public
 *
 * var xs = _();
 * xs.write(1);
 * xs.write(2);
 * xs.end();
 *
 * xs.toArray(function (ys) {
 *     // ys will be [1, 2]
 * });
 *
 * // Do *not* do this.
 * var xs2 = _().toArray(_.log);
 * xs2.write(1); // This call is illegal.
 */

addMethod('write', function (x) {
    // console.log(this.id, 'write', x, this.paused);
    this._writeOutgoing(x);
    return !this.paused;
});

/**
 * Forks a stream, allowing you to add additional consumers with shared
 * back-pressure. A stream forked to multiple consumers will only pull values
 * from its source as fast as the slowest consumer can handle them.
 *
 * @id fork
 * @section Higher-order Streams
 * @name Stream.fork()
 * @api public
 *
 * var xs = _([1, 2, 3, 4]);
 * var ys = xs.fork();
 * var zs = xs.fork();
 *
 * // no values will be pulled from xs until zs also resume
 * ys.resume();
 *
 * // now both ys and zs will get values from xs
 * zs.resume();
 */

addMethod('fork', function () {
    if (this._requests) {
        throw new Error('Cannot fork a stream with an outstanding pull() request.');
    }

    if (this._consumer) {
        throw new Error('Cannot fork a stream that has already been consumed().');
    }

    if (!this._multiplexer) {
        this._multiplexer = new StreamMultiplexer(this);
    }

    var s = this._multiplexer.newStream();
    s.id = 'fork:' + s.id;
    return s;
});

/**
 * Observes a stream, allowing you to handle values as they are emitted, without
 * adding back-pressure or causing data to be pulled from the source. This can
 * be useful when you are performing two related queries on a stream where one
 * would block the other. Just be aware that a slow observer could fill up its
 * buffer and cause memory issues. Where possible, you should use [fork](#fork).
 *
 * @id observe
 * @section Higher-order Streams
 * @name Stream.observe()
 * @api public
 *
 * var xs = _([1, 2, 3, 4]);
 * var ys = xs.fork();
 * var zs = xs.observe();
 *
 * // now both zs and ys will receive data as fast as ys can handle it
 * ys.resume();
 */

addMethod('observe', function () {
    var s = this.create();
    s.id = 'observe:' + s.id;
    s.writable = false;

    s.onDestroy(this._removeObserver.bind(this, s));

    // s.source = this;
    this._observers.push(s);
    return s;
});

/**
 * Extracts errors from a Stream and applies them to an error handler
 * function. Returns a new Stream with the errors removed (unless the error
 * handler chooses to rethrow them using `push`). Errors can also be
 * transformed and put back onto the Stream as values.
 *
 * @id errors
 * @section Transforms
 * @name Stream.errors(f)
 * @param {Function} f - the function to pass all errors to
 * @api public
 *
 * getDocument.errors(function (err, push) {
 *     if (err.statusCode === 404) {
 *         // not found, return empty doc
 *         push(null, {});
 *     }
 *     else {
 *         // otherwise, re-throw the error
 *         push(err);
 *     }
 * });
 */

addMethod('errors', function (f) {
    return this.consume(function (err, x, push, next) {
        if (err) {
            f(err, push);
            next();
        }
        else if (x === nil) {
            push(null, nil);
        }
        else {
            push(null, x);
            next();
        }
    });
});

/**
 * Like the [errors](#errors) method, but emits a Stream end marker after
 * an Error is encountered.
 *
 * @id stopOnError
 * @section Transforms
 * @name Stream.stopOnError(f)
 * @param {Function} f - the function to handle an error
 * @api public
 *
 * brokenStream.stopOnError(function (err) {
 *     //console.error('Something broke: ' + err);
 * });
 */

addMethod('stopOnError', function (f) {
    return this.consume(function (err, x, push, next) {
        if (err) {
            f(err, push);
            push(null, nil);
        }
        else if (x === nil) {
            push(null, nil);
        }
        else {
            push(null, x);
            next();
        }
    });
});

/**
 * Iterates over every value from the Stream, calling the iterator function
 * on each of them. This method consumes the Stream.
 *
 * If an error from the Stream reaches this call, it will emit an `error` event
 * (i.e., it will call `emit('error')` on the stream being consumed).  This
 * event will cause an error to be thrown if unhandled.
 *
 * While `each` consumes the stream, it is possible to chain [done](#done) (and
 * *only* `done`) after it.
 *
 * @id each
 * @section Consumption
 * @name Stream.each(f)
 * @param {Function} f - the iterator function
 * @api public
 *
 * _([1, 2, 3, 4]).each(function (x) {
 *     // will be called 4 times with x being 1, 2, 3 and 4
 * });
 */

addMethod('each', function (f) {
    var self = this;
    var s = this.consume(function (err, x, push, next) {
        if (err) {
            self.emit('error', err);
        }
        else if (x === nil) {
            push(null, nil);
        }
        else {
            f(x);
            next();
        }
    });
    s.resume();
    return s;
});

/**
 * Applies all values from a Stream as arguments to a function. This method consumes the stream.
 * `f` will always be called when the `nil` token is encountered, even when the stream is empty.
 *
 * @id apply
 * @section Consumption
 * @name Stream.apply(f)
 * @param {Function} f - the function to apply arguments to
 * @api public
 *
 * _([1, 2, 3]).apply(function (a, b, c) {
 *     // a === 1
 *     // b === 2
 *     // c === 3
 * });
 *
 * _([1, 2, 3]).apply(function (a) {
 *     // arguments.length === 3
 *     // a === 1
 * });
 */

addMethod('apply', function (f) {
    return this.toArray(function (args) {
        f.apply(null, args);
    });
});

/**
 * Collects all values from a Stream into an Array and calls a function with
 * the result. This method consumes the stream.
 *
 * If an error from the Stream reaches this call, it will emit an `error` event
 * (i.e., it will call `emit('error')` on the stream being consumed).  This
 * event will cause an error to be thrown if unhandled.
 *
 * @id toArray
 * @section Consumption
 * @name Stream.toArray(f)
 * @param {Function} f - the callback to provide the completed Array to
 * @api public
 *
 * _([1, 2, 3, 4]).toArray(function (x) {
 *     // parameter x will be [1,2,3,4]
 * });
 */

addMethod('toArray', function (f) {
    var self = this;
    return this.collect().pull(function (err, x) {
        if (err) {
            self.emit('error', err);
        }
        else {
            f(x);
        }
    });
});

/**
 * Calls a function once the Stream has ended. This method consumes the stream.
 * If the Stream has already ended, the function is called immediately.
 *
 * If an error from the Stream reaches this call, it will emit an `error` event
 * (i.e., it will call `emit('error')` on the stream being consumed).  This
 * event will cause an error to be thrown if unhandled.
 *
 * As a special case, it is possible to chain `done` after a call to
 * [each](#each) even though both methods consume the stream.
 *
 * @id done
 * @section Consumption
 * @name Stream.done(f)
 * @param {Function} f - the callback
 * @api public
 *
 * var total = 0;
 * _([1, 2, 3, 4]).each(function (x) {
 *     total += x;
 * }).done(function () {
 *     // total will be 10
 * });
 */

addMethod('done', function (f) {
    if (this.ended) {
        f();
        return null;
    }
    var self = this;
    return this.consume(function (err, x, push, next) {
        if (err) {
            self.emit('error', err);
        }
        else if (x === nil) {
            f();
        }
        else {
            next();
        }
    }).resume();
});

/**
 * Registers a callback that will be called when the stream is destroyed. The
 * callback will have its `this` variable bound to the stream being destroyed.
 *
 * This method can be used to free any resources that were allocated when
 * creating the stream.
 *
 * @id onDestroy
 * @section Consumption
 * @name Stream.onDestroy(f)
 * @param {Function} f - the callback
 * @api public
 *
 * var _([1]).onDestroy(function () {
 *     console.log('freed');
 * }).each(console.log);
 * // => 1
 * // => freed
 */

Stream.prototype.onDestroy = function onDestroy(f) {
    this._destructors.push(f);
    return this;
};

/**
 * Returns the result of a stream to a nodejs-style callback function.
 *
 * If the stream contains a single value, it will call `cb`
 * with the single item emitted by the stream (if present).
 * If the stream is empty, `cb` will be called without any arguments.
 * If an error is encountered in the stream, this function will stop
 * consumption and call `cb` with the error.
 * If the stream contains more than one item, it will stop consumption
 * and call `cb` with an error.
 *
 * @id toCallback
 * @section Consumption
 * @name Stream.toCallback(cb)
 * @param {Function} cb - the callback to provide the error/result to
 * @api public
 *
 * _([1, 2, 3, 4]).collect().toCallback(function (err, result) {
 *     // parameter result will be [1,2,3,4]
 *     // parameter err will be null
 * });
 */

addMethod('toCallback', function (cb) {
    var value;
    var hasValue = false; // In case an emitted value === null or === undefined.

    this.consume(function (err, x, push, next) {
        if (err) {
            push(null, nil);
            if (hasValue) {
                cb(new Error('toCallback called on stream emitting multiple values'));
            }
            else {
                cb(err);
            }
        }
        else if (x === nil) {
            if (hasValue) {
                cb(null, value);
            }
            else {
                cb();
            }
        }
        else {
            if (hasValue) {
                push(null, nil);
                cb(new Error('toCallback called on stream emitting multiple values'));
            }
            else {
                value = x;
                hasValue = true;
                next();
            }
        }
    }).resume();
});

/**
 * Creates a new Stream of transformed values by applying a function to each
 * value from the source.
 *
 * @id map
 * @section Transforms
 * @name Stream.map(f)
 * @param {Function} f - the transformation function
 * @api public
 *
 * var doubled = _([1, 2, 3, 4]).map(function (x) {
 *     return x * 2;
 * }); // => 2, 4, 6, 8
 */

addMethod('map', function (f) {
    if (!_.isFunction(f)) {
        throw new Error('map expects a function as its only argument.');
    }
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            push(err, x);
        }
        else {
            var fnVal, fnErr;
            try {
                fnVal = f(x);
            }
            catch (e) {
                fnErr = e;
            }
            push(fnErr, fnVal);
            next();
        }
    });
});

/**
 * Creates a new Stream which applies a function to each value from the source
 * and re-emits the source value. Useful when you want to mutate the value or
 * perform side effects
 *
 * @id doto
 * @section Transforms
 * @name Stream.doto(f)
 * @param {Function} f - the function to apply
 * @api public
 *
 * var appended = _([[1], [2], [3], [4]]).doto(function (x) {
 *     x.push(1);
 * });
 *
 * _([1, 2, 3]).doto(console.log)
 * // 1
 * // 2
 * // 3
 * // => 1, 2, 3
 */

addMethod('doto', function (f) {
    return this.map(function (x) {
        f(x);
        return x;
    });
});

/**
 * An alias for the [doto](#doto) method.
 *
 * @id tap
 * @section Transforms
 * @name Stream.tap(f)
 * @param {Function} f - the function to apply
 * @api public
 *
 * _([1, 2, 3]).tap(console.log)
 */

Stream.prototype.tap = Stream.prototype.doto;
_.tap = _.doto;

/**
 * Limits number of values through the stream to a maximum of number of values
 * per window. Errors are not limited but allowed to pass through as soon as
 * they are read from the source.
 *
 * @id ratelimit
 * @section Transforms
 * @name Stream.ratelimit(num, ms)
 * @param {Number} num - the number of operations to perform per window
 * @param {Number} ms - the window of time to limit the operations in (in ms)
 * @api public
 *
 * _([1, 2, 3, 4, 5]).ratelimit(2, 100);
 *
 * // after 0ms => 1, 2
 * // after 100ms => 1, 2, 3, 4
 * // after 200ms => 1, 2, 3, 4, 5
 */

addMethod('ratelimit', function (num, ms) {
    if (num < 1) {
        throw new Error('Invalid number of operations per ms: ' + num);
    }
    var sent = 0;
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            push(null, nil);
        }
        else {
            if (sent < num) {
                sent++;
                push(null, x);
                next();
            }
            else {
                setTimeout(function () {
                    sent = 1;
                    push(null, x);
                    next();
                }, ms);
            }
        }
    });
});

/**
 * Creates a new Stream of values by applying each item in a Stream to an
 * iterator function which must return a (possibly empty) Stream. Each item on
 * these result Streams are then emitted on a single output Stream.
 *
 * This transform is functionally equivalent to `.map(f).sequence()`.
 *
 * @id flatMap
 * @section Higher-order Streams
 * @name Stream.flatMap(f)
 * @param {Function} f - the iterator function
 * @api public
 *
 * var readFile = _.wrapCallback(fs.readFile);
 * filenames.flatMap(readFile)
 */

addMethod('flatMap', function (f) {
    return this.map(f).sequence();
});

/**
 * Retrieves values associated with a given property from all elements in
 * the collection.
 *
 * @id pluck
 * @section Transforms
 * @name Stream.pluck(property)
 * @param {String} prop - the property to which values should be associated
 * @api public
 *
 * var docs = [
 *     {type: 'blogpost', title: 'foo'},
 *     {type: 'blogpost', title: 'bar'},
 *     {type: 'comment', title: 'baz'}
 * ];
 *
 * _(docs).pluck('title').toArray(function (xs) {
 *    // xs is now ['foo', 'bar', 'baz']
 * });
 */

addMethod('pluck', function (prop) {
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            push(err, x);
        }
        else if (_.isObject(x)) {
            push(null, x[prop]);
            next();
        }
        else {
            push(new Error(
                'Expected Object, got ' + (typeof x)
            ));
            next();
        }
    });
});

/**
 * Only applies the transformation strategy on Objects.
 * This helper is used in `pick` and `pickBy`
 **/

var objectOnly = _.curry(function(strategy, x) {
    if (_.isObject(x)) {
        return strategy(x);
    }
    else {
        throw new Error(
            'Expected Object, got ' + (typeof x)
        );
    }
});


/**
 *
 * Retrieves copies of all the elements in the collection
 * that satisfy a given predicate. Note: When using ES3,
 * only enumerable elements are selected. Both enumerable
 * and non-enumerable elements are selected when using ES5.
 *
 * @id pickBy
 * @section Transforms
 * @name Stream.pickBy(f)
 * @param {Function} f - the predicate function
 * @api public
 *
 *  var dogs = [
 *      {breed: 'chihuahua', name: 'Princess', age: 5},
 *      {breed: 'labrador', name: 'Rocky', age: 3},
 *      {breed: 'german-shepherd', name: 'Waffles', age: 9}
 *  ];

 *  _(dogs).pickBy(function (value, key) {
 *      return value > 4;
 *  }).toArray(function (xs) {
 *    // xs is now:
 *    [
 *      { age: 5 },
 *      {},
 *      { age: 9 }
 *    ]
 *  });
 */

addMethod('pickBy', function (f) {
    return this.map(objectOnly(function (x) {
        var out = {};

        // prevents testing overridden properties multiple times.
        var seen = isES5 ? Object.create(null) : {};
        var obj = x;  // variable used to traverse prototype chain
        function testAndAdd (prop) {
            if (seen[prop] !== true && f(x[prop], prop)) {
                out[prop] = x[prop];
                seen[prop] = true;
            }
        }
        if (isES5) {
            do {
                Object.getOwnPropertyNames(obj).forEach(testAndAdd);
                obj = Object.getPrototypeOf(obj);
            } while (obj);
        }
        else {
            for (var k in x) {
                testAndAdd(k);
            }
        }
        return out;
    }));
});

/**
 *
 * Retrieves copies of all elements in the collection,
 * with only the whitelisted keys. If one of the whitelisted
 * keys does not exist, it will be ignored.
 *
 * @id pick
 * @section Transforms
 * @name Stream.pick(properties)
 * @param {Array} properties - property names to white filter
 * @api public
 *
 * var dogs = [
 *      {breed: 'chihuahua', name: 'Princess', age: 5},
 *      {breed: 'labrador', name: 'Rocky', age: 3},
 *      {breed: 'german-shepherd', name: 'Waffles', age: 9}
 * ];
 *
 * _(dogs).pick(['breed', 'age']).toArray(function (xs) {
 *       // xs is now:
 *       [
 *           {breed: 'chihuahua', age: 5},
 *           {breed: 'labrador', age: 3},
 *           {breed: 'german-shepherd', age: 9}
 *       ]
 * });
 *
 * _(dogs).pick(['owner']).toArray(function (xs) {
 *      // xs is now:
 *      [
 *          {},
 *          {},
 *          {}
 *      ]
 * });*/

addMethod('pick', function (properties) {
    return this.map(objectOnly(function(x) {
        var out = {};
        for (var i = 0, length = properties.length; i < length; i++) {
            var p = properties[i];
            if (p in x) {
                out[p] = x[p];
            }
        }
        return out;
    }));
});

/**
 * Creates a new Stream that includes only the values that pass a truth test.
 *
 * @id filter
 * @section Transforms
 * @name Stream.filter(f)
 * @param {Function} f - the truth test function
 * @api public
 *
 * var evens = _([1, 2, 3, 4]).filter(function (x) {
 *     return x % 2 === 0;
 * });
 */

addMethod('filter', function (f) {
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            push(err, x);
        }
        else {
            var fnVal, fnErr;
            try {
                fnVal = f(x);
            }
            catch (e) {
                fnErr = e;
            }

            if (fnErr) {
                push(fnErr);
            }
            else if (fnVal) {
                push(null, x);
            }
            next();
        }
    });
});

/**
 * Filters using a predicate which returns a Stream. If you need to check
 * against an asynchronous data source when filtering a Stream, this can
 * be convenient. The Stream returned from the filter function should have
 * a Boolean as its first value (all other values on the Stream will be
 * disregarded).
 *
 * @id flatFilter
 * @section Higher-order Streams
 * @name Stream.flatFilter(f)
 * @param {Function} f - the truth test function which returns a Stream
 * @api public
 *
 * var checkExists = _.wrapCallback(fs.access);
 *
 * filenames.flatFilter(checkExists)
 */

addMethod('flatFilter', function (f) {
    return this.flatMap(function (x) {
        return f(x).take(1).otherwise(errorStream())
        .flatMap(function (bool) {
            return _(bool ? [x] : []);
        });
    });

    function errorStream() {
        return _(function (push) {
            push(new Error('Stream returned by function was empty.'));
            push(null, _.nil);
        });
    }
});

/**
 * The inverse of [filter](#filter).
 *
 * @id reject
 * @section Transforms
 * @name Stream.reject(f)
 * @param {Function} f - the truth test function
 * @api public
 *
 * var odds = _([1, 2, 3, 4]).reject(function (x) {
 *     return x % 2 === 0;
 * });
 */

addMethod('reject', function (f) {
    return this.filter(_.compose(_.not, f));
});

/**
 * A convenient form of [filter](#filter), which returns the first object from a
 * Stream that passes the provided truth test.
 *
 * @id find
 * @section Transforms
 * @name Stream.find(f)
 * @param {Function} f - the truth test function which returns a Stream
 * @api public
 *
 * var docs = [
 *     {type: 'blogpost', title: 'foo'},
 *     {type: 'blogpost', title: 'bar'},
 *     {type: 'comment', title: 'foo'}
 * ];
 *
 * var f = function (x) {
 *     return x.type == 'blogpost';
 * };
 *
 * _(docs).find(f);
 * // => {type: 'blogpost', title: 'foo'}
 *
 * // example with partial application
 * var firstBlogpost = _.find(f);
 *
 * firstBlogpost(docs)
 * // => {type: 'blogpost', title: 'foo'}
 */

addMethod('find', function (f) {
    return this.filter(f).take(1);
});

/**
 * A convenient form of [where](#where), which returns the first object from a
 * Stream that matches a set of property values. findWhere is to [where](#where) as [find](#find) is to [filter](#filter).
 *
 * @id findWhere
 * @section Transforms
 * @name Stream.findWhere(props)
 * @param {Object} props - the properties to match against
 * @api public
 *
 * var docs = [
 *     {type: 'blogpost', title: 'foo'},
 *     {type: 'blogpost', title: 'bar'},
 *     {type: 'comment', title: 'foo'}
 * ];
 *
 * _(docs).findWhere({type: 'blogpost'})
 * // => {type: 'blogpost', title: 'foo'}
 *
 * // example with partial application
 * var firstBlogpost = _.findWhere({type: 'blogpost'});
 *
 * firstBlogpost(docs)
 * // => {type: 'blogpost', title: 'foo'}
 */

addMethod('findWhere', function (props) {
    return this.where(props).take(1);
});


/**
 * A convenient form of [reduce](#reduce), which groups items based on a function or property name
 *
 * @id group
 * @section Transforms
 * @name Stream.group(f)
 * @param {Function | String} f - the function or property name on which to group,
 *                              toString() is called on the result of a function.
 * @api public
 *
 * var docs = [
 *     {type: 'blogpost', title: 'foo'},
 *     {type: 'blogpost', title: 'bar'},
 *     {type: 'comment', title: 'foo'}
 * ];
 *
 * var f = function (x) {
 *     return x.type;
 * };
 *
 * _(docs).group(f); OR _(docs).group('type');
 * // => {
 * // =>    'blogpost': [{type: 'blogpost', title: 'foo'}, {type: 'blogpost', title: 'bar'}]
 * // =>    'comment': [{type: 'comment', title: 'foo'}]
 * // =>  }
 *
 */

addMethod('group', function (f) {
    var lambda = _.isString(f) ? _.get(f) : f;
    return this.reduce(function (m, o) {
        var key = lambda(o);
        if (!hasOwn.call(m, key)) { m[key] = []; }
        m[key].push(o);
        return m;
    }, {});
});

/**
 * Filters a Stream to drop all non-truthy values.
 *
 * @id compact
 * @section Transforms
 * @name Stream.compact()
 * @api public
 *
 * var compacted = _([0, 1, false, 3, null, undefined, 6]).compact();
 * // => 1, 3, 6
 */

addMethod('compact', function () {
    return this.filter(function (x) {
        return x;
    });
});

/**
 * A convenient form of [filter](#filter), which returns all objects from a Stream
 * which match a set of property values.
 *
 * @id where
 * @section Transforms
 * @name Stream.where(props)
 * @param {Object} props - the properties to match against
 * @api public
 *
 * var docs = [
 *     {type: 'blogpost', title: 'foo'},
 *     {type: 'blogpost', title: 'bar'},
 *     {type: 'comment', title: 'foo'}
 * ];
 *
 * _(docs).where({title: 'foo'})
 * // => {type: 'blogpost', title: 'foo'}
 * // => {type: 'comment', title: 'foo'}
 *
 * // example with partial application
 * var getBlogposts = _.where({type: 'blogpost'});
 *
 * getBlogposts(docs)
 * // => {type: 'blogpost', title: 'foo'}
 * // => {type: 'blogpost', title: 'bar'}
 */

addMethod('where', function (props) {
    return this.filter(function (x) {
        for (var k in props) {
            if (x[k] !== props[k]) {
                return false;
            }
        }
        return true;
    });
});

/**
 * Filters out all duplicate values from the stream and keeps only the first
 * occurence of each value, using the provided function to define equality.
 *
 * Note:
 *
 * - Memory: In order to guarantee that each unique item is chosen only once,
 *   we need to keep an internal buffer of all unique values. This may outgrow
 *   the available memory if you are not cautious about the size of your stream
 *   and the number of unique objects you may receive on it.
 * - Errors: The comparison function should never throw an error. However, if
 *   it does, this transform will emit an error for each all that throws. This
 *   means that one value may turn into multiple errors.
 *
 * @id uniqBy
 * @section Transforms
 * @name Stream.uniqBy(compare)
 * @param {Function} compare - custom equality predicate
 * @api public
 *
 * var colors = [ 'blue', 'red', 'red', 'yellow', 'blue', 'red' ]
 *
 * _(colors).uniqBy(function(a, b) { return a[1] === b[1]; })
 * // => 'blue'
 * // => 'red'
 *
 */

addMethod('uniqBy', function (compare) {
    var uniques = [];
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            push(err, x);
        }
        else {
            var seen = false;
            var hasErr;
            for (var i = 0, len = uniques.length; i < len; i++) {
                try {
                    seen = compare(x, uniques[i]);
                }
                catch (e) {
                    hasErr = e;
                    seen = true;
                }
                if (seen) {
                    break;
                }
            }
            if (!seen) {
                uniques.push(x);
                push(null, x);
            }
            if (hasErr) {
                push(hasErr);
            }
            next();
        }
    });
});

/**
 * Filters out all duplicate values from the stream and keeps only the first
 * occurence of each value, using `===` to define equality.
 *
 * Like [uniqBy](#uniqBy), this transform needs to store a buffer containing
 * all unique values that has been encountered. Be careful about using this
 * transform on a stream that has many unique values.
 *
 * @id uniq
 * @section Transforms
 * @name Stream.uniq()
 * @api public
 *
 * var colors = [ 'blue', 'red', 'red', 'yellow', 'blue', 'red' ]
 *
 * _(colors).uniq()
 * // => 'blue'
 * // => 'red'
 * // => 'yellow'
 */

addMethod('uniq', function () {
    if (!_.isUndefined(_global.Set)) {
        var uniques = new _global.Set(),
            size = uniques.size;

        return this.consume(function (err, x, push, next) {
            if (err) {
                push(err);
                next();
            }
            else if (x === nil) {
                push(err, x);
            }
            // pass NaN through as Set does not respect strict
            // equality in this case.
            else if (x !== x) {
                push(null, x);
                next();
            }
            else {
                uniques.add(x);
                if (uniques.size > size) {
                    size = uniques.size;
                    push(null, x);
                }
                next();
            }
        });
    }
    return this.uniqBy(function (a, b) {
        return a === b;
    });
});

/**
 * Takes a `finite` stream of streams and returns a stream where the first
 * element from each separate stream is combined into a single data event,
 * followed by the second elements of each stream and so on until the shortest
 * input stream is exhausted.
 *
 * @id zipAll
 * @section Higher-order Streams
 * @name Stream.zipAll()
 * @api public
 *
 * _([
 *     _([1, 2, 3]),
 *     _([4, 5, 6]),
 *     _([7, 8, 9]),
 *     _([10, 11, 12])
 * ]).zipAll()
 * // => [ [ 1, 4, 7, 10 ], [ 2, 5, 8, 11 ], [ 3, 6, 9, 12 ] ]
 *
 * // shortest stream determines length of output stream
 * _([
 *     _([1, 2, 3, 4]),
 *     _([5, 6, 7, 8]),
 *     _([9, 10, 11, 12]),
 *     _([13, 14])
 * ]).zipAll()
 * // => [ [ 1, 5, 9, 13 ], [ 2, 6, 10, 14 ] ]
 */

addMethod('zipAll', function () {
    var returned = 0;
    var z = [];
    var finished = false;

    function nextValue(index, max, src, push, next) {
        src.pull(function (err, x) {
            if (err) {
                push(err);
                nextValue(index, max, src, push, next);
            }
            else if (x === _.nil) {
                if (!finished) {
                    finished = true;
                    push(null, nil);
                }
            }
            else {
                returned++;
                z[index] = x;
                if (returned === max) {
                    push(null, z);
                    next();
                }
            }
        });
    }

    var self = this;
    return self.collect().flatMap(function (array) {
        if (!array.length) {
            return self.create([]);
        }
        return self.create(function (push, next) {
            returned = 0;
            z = [];
            for (var i = 0, length = array.length; i < length; i++) {
                nextValue(i, length, array[i], push, next);
            }
        });
    });

});

/**
 * Takes a stream and a `finite` stream of `N` streams
 * and returns a stream of the corresponding `(N+1)`-tuples.
 *
 * @id zipEach
 * @section Higher-order Streams
 * @name Stream.zipEach(ys)
 * @param {Array | Stream} ys - the array of streams to combine values with
 * @api public
 *
 * _([1,2,3]).zipEach([[4, 5, 6], [7, 8, 9], [10, 11, 12]])
 * // => [ [ 1, 4, 7, 10 ], [ 2, 5, 8, 11 ], [ 3, 6, 9, 12 ] ]
 *
 * // shortest stream determines length of output stream
 * _([1, 2, 3, 4]).zipEach([[5, 6, 7, 8], [9, 10, 11, 12], [13, 14]])
 * // => [ [ 1, 5, 9, 13 ], [ 2, 6, 10, 14 ] ]
 */

addMethod('zipEach', function (ys) {
    return this.create([this]).concat(this.create(ys).map(this.create.bind(this))).zipAll();
});

/**
 * Takes two Streams and returns a Stream of corresponding pairs. The size of
 * the resulting stream is the smaller of the two source streams.
 *
 * @id zip
 * @section Higher-order Streams
 * @name Stream.zip(ys)
 * @param {Array | Stream} ys - the other stream to combine values with
 * @api public
 *
 * _(['a', 'b', 'c']).zip([1, 2, 3])  // => ['a', 1], ['b', 2], ['c', 3]
 *
 * _(['a', 'b', 'c']).zip(_([1]))  // => ['a', 1]
 */

addMethod('zip', function (ys) {
    return this.create([this, this.create(ys)]).zipAll();
});

/**
 * Takes one Stream and batches incoming data into arrays of given length
 *
 * @id batch
 * @section Transforms
 * @name Stream.batch(n)
 * @param {Number} n - length of the array to batch
 * @api public
 *
 * _([1, 2, 3, 4, 5]).batch(2)  // => [1, 2], [3, 4], [5]
 */

addMethod('batch', function (n) {
    return this.batchWithTimeOrCount(-1, n);
});

/**
 * Takes one Stream and batches incoming data within a maximum time frame
 * into arrays of a maximum length.
 *
 * @id batchWithTimeOrCount
 * @section Transforms
 * @name Stream.batchWithTimeOrCount(ms, n)
 * @param {Number} ms - the maximum milliseconds to buffer a batch
 * @param {Number} n - the maximum length of the array to batch
 * @api public
 *
 * _(function (push) {
 *     push(1);
 *     push(2);
 *     push(3);
 *     setTimeout(push, 20, 4);
 * }).batchWithTimeOrCount(10, 2)
 *
 * // => [1, 2], [3], [4]
 */

addMethod('batchWithTimeOrCount', function (ms, n) {
    var batched = [],
        timeout;

    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            if (batched.length > 0) {
                push(null, batched);
                clearTimeout(timeout);
            }

            push(null, nil);
        }
        else {
            batched.push(x);

            if (batched.length === n) {
                push(null, batched);
                batched = [];
                clearTimeout(timeout);
            }
            else if (batched.length === 1 && ms >= 0) {
                timeout = setTimeout(function () {
                    push(null, batched);
                    batched = [];
                }, ms);
            }

            next();
        }
    });
});

/**
 * Creates a new Stream with the separator interspersed between the elements of the source.
 *
 * `intersperse` is effectively the inverse of [splitBy](#splitBy).
 *
 * @id intersperse
 * @section Transforms
 * @name Stream.intersperse(sep)
 * @param {String} sep - the value to intersperse between the source elements
 * @api public
 *
 * _(['ba', 'a', 'a']).intersperse('n')  // => ba, n, a, n, a
 * _(['mississippi']).splitBy('ss').intersperse('ss')  // => mi, ss, i, ss, ippi
 * _(['foo']).intersperse('bar')  // => foo
 */

addMethod('intersperse', function (separator) {
    var started = false;
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            push(null, nil);
        }
        else {
            if (started) {
                push(null, separator);
            }
            else {
                started = true;
            }
            push(null, x);
            next();
        }
    });
});

/**
 * Splits the source Stream by a separator and emits the pieces in between, much like splitting a string.
 *
 * `splitBy` is effectively the inverse of [intersperse](#intersperse).
 *
 * @id splitBy
 * @section Transforms
 * @name Stream.splitBy(sep)
 * @param {String | RegExp} sep - the separator to split on
 * @api public
 *
 * _(['mis', 'si', 's', 'sippi']).splitBy('ss')  // => mi, i, ippi
 * _(['ba', 'a', 'a']).intersperse('n').splitBy('n')  // => ba, a, a
 * _(['foo']).splitBy('bar')  // => foo
 */

addMethod('splitBy', function (sep) {
    var decoder = new Decoder();
    var buffer = false;

    function drain(x, push) {
        buffer = (buffer || '') + decoder.write(x);
        var pieces = buffer.split(sep);
        buffer = pieces.pop();

        pieces.forEach(function (piece) {
            push(null, piece);
        });
    }

    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            if (_.isString(buffer)) {
                drain(decoder.end(), push);
                push(null, buffer);
            }
            push(null, nil);
        }
        else {
            drain(x, push);
            next();
        }
    });
});

/**
 * [splitBy](#splitBy) over newlines.
 *
 * @id split
 * @section Transforms
 * @name Stream.split()
 * @api public
 *
 * _(['a\n', 'b\nc\n', 'd', '\ne']).split()  // => a, b, c, d, e
 * _(['a\r\nb\nc']]).split()  // => a, b, c
 */

addMethod('split', function () {
    return this.splitBy(/\r?\n/);
});

/**
 * Creates a new Stream with the values from the source in the range of `start` (inclusive) to `end` (exclusive).
 * `start` and `end` must be of type `Number`, if `start` is not a `Number` it will default to `0`
 * and, likewise, `end` will default to `Infinity`: this could result in the whole stream being be
 * returned.
 *
 * @id slice
 * @section Transforms
 * @name Stream.slice(start, end)
 * @param {Number} start - integer representing index to start reading from source (inclusive)
 * @param {Number} stop - integer representing index to stop reading from source (exclusive)
 * @api public
 *
 * _([1, 2, 3, 4]).slice(1, 3) // => 2, 3
 */

addMethod('slice', function(start, end) {
    var index = 0;
    start = typeof start != 'number' || start < 0 ? 0 : start;
    end = typeof end != 'number' ? Infinity : end;

    if (start === 0 && end === Infinity) {
        return this;
    }
    else if (start >= end) {
        return _([]);
    }
    var s = this.consume(function (err, x, push, next) {
        var done = x === nil;
        if (err) {
            push(err);
        }
        else if (!done && index++ >= start) {
            push(null, x);
        }

        if (!done && index < end) {
            next();
        }
        else {
            push(null, nil);
        }
    });
    s.id = 'slice:' + s.id;
    return s;
});

/**
 * Creates a new Stream with the first `n` values from the source. `n` must be of type `Number`,
 * if not the whole stream will be returned.
 *
 * @id take
 * @section Transforms
 * @name Stream.take(n)
 * @param {Number} n - integer representing number of values to read from source
 * @api public
 *
 * _([1, 2, 3, 4]).take(2) // => 1, 2
 */

addMethod('take', function (n) {
    var s = this.slice(0, n);
    s.id = 'take:' + s.id;
    return s;
});

/**
 * Acts as the inverse of [`take(n)`](#take) - instead of returning the first `n` values, it ignores the
 * first `n` values and then emits the rest. `n` must be of type `Number`, if not the whole stream will
 * be returned. All errors (even ones emitted before the nth value) will be emitted.
 *
 * @id drop
 * @section Transforms
 * @name Stream.drop(n)
 * @param {Number} n - integer representing number of values to read from source
 * @api public
 *
 * _([1, 2, 3, 4]).drop(2) // => 3, 4
 */

addMethod('drop', function (n) {
    return this.slice(n, Infinity);
});

/**
 * Creates a new Stream with only the first value from the source.
 *
 * @id head
 * @section Transforms
 * @name Stream.head()
 * @api public
 *
 * _([1, 2, 3, 4]).head() // => 1
 */

addMethod('head', function () {
    return this.take(1);
});

/**
 * Drops all values from the Stream apart from the last one (if any).
 *
 * @id last
 * @section Transforms
 * @name Stream.last()
 * @api public
 *
 * _([1, 2, 3, 4]).last()  // => 4
 */

addMethod('last', function () {
    var nothing = {};
    var prev = nothing;
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            if (prev !== nothing) {
                push(null, prev);
            }
            push(null, nil);
        }
        else {
            prev = x;
            next();
        }
    });
});

/**
 * Collects all values together then emits each value individually in sorted
 * order. The method for sorting the elements is defined by the comparator
 * function supplied as a parameter.
 *
 * The comparison function takes two arguments `a` and `b` and should return
 *
 * - a negative number if `a` should sort before `b`.
 * - a positive number if `a` should sort after `b`.
 * - zero if `a` and `b` may sort in any order (i.e., they are equal).
 *
 * This function must also define a [partial
 * order](https://en.wikipedia.org/wiki/Partially_ordered_set). If it does not,
 * the resulting ordering is undefined.
 *
 * @id sortBy
 * @section Transforms
 * @name Stream.sortBy(f)
 * @param {Function} f - the comparison function
 * @api public
 *
 * var sorts = _([3, 1, 4, 2]).sortBy(function (a, b) {
 *     return b - a;
 * }).toArray(_.log);
 *
 * //=> [4, 3, 2, 1]
 */

addMethod('sortBy', function (f) {
    return this.collect().invoke('sort', [f]).sequence();
});

/**
 * Collects all values together then emits each value individually but in sorted order.
 * The method for sorting the elements is ascending lexical.
 *
 * @id sort
 * @section Transforms
 * @name Stream.sort()
 * @api public
 *
 * var sorted = _(['b', 'z', 'g', 'r']).sort().toArray(_.log);
 * // => ['b', 'g', 'r', 'z']
 */

addMethod('sort', function () {
    return this.sortBy();
});


/**
 * Transforms a stream using an arbitrary target transform.
 *
 * If `target` is a function, this transform passes the current Stream to it,
 * returning the result.
 *
 * If `target` is a [Duplex
 * Stream](https://nodejs.org/api/stream.html#stream_class_stream_duplex_1),
 * this transform pipes the current Stream through it. It will always return a
 * Highland Stream (instead of the piped to target directly as in
 * [pipe](#pipe)). Any errors emitted will be propagated as Highland errors.
 *
 * @id through
 * @section Higher-order Streams
 * @name Stream.through(target)
 * @param {Function | Duplex Stream} target - the stream to pipe through or a
 * function to call.
 * @api public
 *
 * function oddDoubler(s) {
 *     return s.filter(function (x) {
 *         return x % 2; // odd numbers only
 *     })
 *     .map(function (x) {
 *         return x * 2;
 *     });
 * }
 *
 * _([1, 2, 3, 4]).through(oddDoubler).toArray(function (xs) {
 *     // xs will be [2, 6]
 * });
 *
 * // Can also be used with Node Through Streams
 * _(filenames).through(jsonParser).map(function (obj) {
 *     // ...
 * });
 *
 * // All errors will be propagated as Highland errors
 * _(['zz{"a": 1}']).through(jsonParser).errors(function (err) {
 *   console.log(err); // => SyntaxError: Unexpected token z
 * });
 */

addMethod('through', function (target) {
    var output;

    if (_.isFunction(target)) {
        return target(this);
    }
    else {
        target.pause();
        output = this.createChild();
        this.on('error', writeErr);
        target.on('error', writeErr);
        return this.pipe(target).pipe(output);
    }

    function writeErr(err) {
        output.write(new StreamError(err));
    }
});

/**
 * Creates a 'Through Stream', which passes data through a pipeline
 * of functions or other through Streams. This is particularly useful
 * when combined with partial application of Highland functions to expose a
 * Node-compatible Through Stream.
 *
 * This is not a method on a Stream, and it only exposed at the top-level
 * as `_.pipeline`. It takes an arbitrary number of arguments.
 *
 * @id pipeline
 * @section Higher-order Streams
 * @name _.pipeline(...)
 * @api public
 *
 * var through = _.pipeline(
 *     _.map(parseJSON),
 *     _.filter(isBlogpost),
 *     _.reduce(collectCategories)
 *     _.through(otherPipeline)
 * );
 *
 * readStream.pipe(through).pipe(outStream);
 *
 * // Alternatively, you can use pipeline to manipulate a stream in
 * // the chained method call style:
 *
 * var through2 = _.pipeline(function (s) {
 *     return s.map(parseJSON).filter(isBlogpost); // etc.
 * });
 */

addToplevelMethod('pipeline', function (/*through...*/) {
    if (!arguments.length) {
        return this();
    }
    var start = arguments[0], rest, startHighland;
    if (!_.isStream(start) && !_.isFunction(start.resume)) {
        // not a Highland stream or Node stream, start with empty stream
        start = this();
        startHighland = start;
        rest = slice.call(arguments);
    }
    else {
        // got a stream as first argument, co-erce to Highland stream
        startHighland = this(start);
        rest = slice.call(arguments, 1);
    }

    var end = rest.reduce(function (src, dest) {
        return src.through(dest);
    }, startHighland);

    var wrapper = this(function (push, next) {
        end.pull(function (err, x) {
            push(err, x);
            if (x !== nil) {
                next();
            }
        });
    });

    wrapper.writable = true;

    wrapper.write = function (x) {
        return start.write(x);
    };

    wrapper.end = function () {
        wrapper.writable = false;
        return start.end();
    };

    start.on('drain', function () {
        wrapper.emit('drain');
    });

    return wrapper;
});

/**
 * Reads values from a Stream of Streams or Arrays, emitting them on a single
 * output Stream. This can be thought of as a [flatten](#flatten), just one
 * level deep, often used for resolving asynchronous actions such as a HTTP
 * request or reading a file.
 *
 * @id sequence
 * @section Higher-order Streams
 * @name Stream.sequence()
 * @api public
 *
 * var nums = _([
 *     _([1, 2, 3]),
 *     _([4, 5, 6])
 * ]);
 *
 * nums.sequence()  // => 1, 2, 3, 4, 5, 6
 *
 * // using sequence to read from files in series
 * var readFile = _.wrapCallback(fs.readFile);
 * filenames.map(readFile).sequence()
 */

addMethod('sequence', function () {
    var original = this;
    var curr = this;
    return this.create(function (push, next) {
        curr.pull(function (err, x) {
            if (err) {
                push(err);
                next();
            }
            else if (_.isArray(x)) {
                if (onOriginalStream()) {
                    // just send all values from array directly
                    x.forEach(function (y) {
                        push(null, y);
                    });
                }
                else {
                    push(null, x);
                }
                next();
            }
            else if (_.isStream(x)) {
                if (onOriginalStream()) {
                    // switch to reading new stream
                    curr = x;
                    next();
                }
                else {
                    // sequence only goes 1 level deep
                    push(null, x);
                    next();
                }
            }
            else if (x === nil) {
                if (onOriginalStream()) {
                    push(null, nil);
                }
                else {
                    // resume reading from original
                    curr = original;
                    next();
                }
            }
            else {
                if (onOriginalStream()) {
                    // we shouldn't be getting non-stream (or array)
                    // values from the top-level stream
                    push(new Error(
                        'Expected Stream, got ' + (typeof x)
                    ));
                    next();
                }
                else {
                    push(null, x);
                    next();
                }
            }
        });
    });

    function onOriginalStream() {
        return curr === original;
    }
});

/**
 * An alias for the [sequence](#sequence) method.
 *
 * @id series
 * @section Higher-order Streams
 * @name Stream.series()
 * @api public
 *
 * var readFile = _.wrapCallback(fs.readFile);
 * filenames.map(readFile).series()
 */

Stream.prototype.series = Stream.prototype.sequence;
_.series = _.sequence;

/**
 * Recursively reads values from a Stream which may contain nested Streams
 * or Arrays. As values or errors are encountered, they are emitted on a
 * single output Stream.
 *
 * @id flatten
 * @section Higher-order Streams
 * @name Stream.flatten()
 * @api public
 *
 * _([1, [2, 3], [[4]]]).flatten();  // => 1, 2, 3, 4
 *
 * var nums = _(
 *     _([1, 2, 3]),
 *     _([4, _([5, 6]) ])
 * );
 *
 * nums.flatten();  // => 1, 2, 3, 4, 5, 6
 */

addMethod('flatten', function () {
    var curr = this;
    var stack = [];
    return this.createChild(function (push, next) {
        curr.pull(function (err, x) {
            if (err) {
                push(err);
                next();
                return;
            }
            if (_.isArray(x)) {
                x = _(x);
            }
            if (_.isStream(x)) {
                stack.push(curr);
                curr = x;
                next();
            }
            else if (x === nil) {
                if (stack.length) {
                    curr = stack.pop();
                    next();
                }
                else {
                    push(null, nil);
                }
            }
            else {
                push(null, x);
                next();
            }
        });
    });
});

/**
 * Takes a Stream of Streams and reads from them in parallel, buffering
 * the results until they can be returned to the consumer in their original
 * order.
 *
 * @id parallel
 * @section Higher-order Streams
 * @name Stream.parallel(n)
 * @param {Number} n - the maximum number of concurrent reads/buffers
 * @api public
 *
 * var readFile = _.wrapCallback(fs.readFile);
 * var filenames = _(['foo.txt', 'bar.txt', 'baz.txt']);
 *
 * // read from up to 10 files at once
 * filenames.map(readFile).parallel(10);
 */

addMethod('parallel', function (n) {
    var source = this;
    var running = [];
    var ended = false;
    var reading_source = false;

    if (typeof n !== 'number') {
        throw new Error('Must specify a number to parallel().');
    }

    if (n <= 0) {
        throw new Error('The parallelism factor must be positive');
    }

    return this.create(function (push, next) {
        if (running.length < n && !ended && !reading_source) {
            // get another stream if not already waiting for one
            reading_source = true;
            source.pull(function (err, x) {
                reading_source = false;
                if (err) {
                    push(err);
                }
                else if (x === nil) {
                    ended = true;
                }
                else if (!_.isStream(x)) {
                    push(new Error('Expected Stream, got ' + (typeof x)));
                }
                else {
                    // got a new source, add it to the running array
                    var run = {stream: x, buffer: []};
                    running.push(run);
                    x.consume(function (_err, y, _push, _next) {
                        if (running[0] === run) {
                            // current output stream
                            if (y === nil) {
                                // remove self from running and check
                                // to see if we need to read from source again
                                running.shift();
                                flushBuffer();
                                next();

                            }
                            else {
                                // push directly onto parallel output stream
                                push(_err, y);
                            }
                        }
                        else {
                            // we're reading ahead, buffer the output
                            run.buffer.push([_err, y]);
                        }
                        if (y !== nil) {
                            // keep reading until we hit nil
                            _next();
                        }
                    }).resume();
                }
                // check if we need to get any more streams
                return next();
            });
        }
        else if (!running.length && ended) {
            // nothing more to do
            push(null, nil);
        }

        function flushBuffer() {
            while (running.length && running[0].buffer.length) {
                var buf = running[0].buffer;
                for (var i = 0; i < buf.length; i++) {
                    if (buf[i][1] === nil) {
                        // this stream has ended
                        running.shift();
                        break;
                    }
                    else {
                        // send the buffered output
                        push.apply(null, buf[i]);
                    }
                }
                buf.length = 0;
            }
        }
        // else wait for more data to arrive from running streams
    });
});

/**
 * Switches source to an alternate Stream if the current Stream is empty.
 *
 * @id otherwise
 * @section Higher-order Streams
 * @name Stream.otherwise(ys)
 * @param {Stream | Function} ys - alternate stream (or stream-returning function) to use if this stream is empty
 * @api public
 *
 * _([1,2,3]).otherwise(['foo'])  // => 1, 2, 3
 * _([]).otherwise(['foo'])       // => 'foo'
 *
 * _.otherwise(_(['foo']), _([1,2,3]))    // => 1, 2, 3
 * _.otherwise(_(['foo']), _([]))         // => 'foo'
 */

addMethod('otherwise', function (ys) {
    var xs = this;
    return xs.consume(function (err, x, push, next) {
        if (err) {
            // got an error, just keep going
            push(err);
            next();
        }
        else if (x === nil) {
            // hit the end without redirecting to xs, use alternative
            if (_.isFunction(ys)) {
                next(ys());
            }
            else {
                next(ys);
            }
        }
        else {
            // got a value, push it, then redirect to xs
            push(null, x);
            next(xs);
        }
    });
});

/**
 * Adds a value to the end of a Stream.
 *
 * @id append
 * @section Transforms
 * @name Stream.append(y)
 * @param y - the value to append to the Stream
 * @api public
 *
 * _([1, 2, 3]).append(4)  // => 1, 2, 3, 4
 */

addMethod('append', function (y) {
    return this.consume(function (err, x, push, next) {
        if (x === nil) {
            push(null, y);
            push(null, _.nil);
        }
        else {
            push(err, x);
            next();
        }
    });
});

/**
 * Boils down a Stream to a single value. The memo is the initial state
 * of the reduction, and each successive step of it should be returned by
 * the iterator function. The iterator is passed two arguments:
 * the memo and the next value.
 *
 * If the iterator throws an error, the reduction stops and the resulting
 * stream will emit that error instead of a value.
 *
 * @id reduce
 * @section Transforms
 * @name Stream.reduce(iterator, memo)
 * @param memo - the initial state of the reduction
 * @param {Function} iterator - the function which reduces the values
 * @api public
 *
 * var add = function (a, b) {
 *     return a + b;
 * };
 *
 * _([1, 2, 3, 4]).reduce(add, 0)  // => 10
 */

addMethod('reduce', function (f, z) {
    // This can't be implemented with scan(), because we don't know if the
    // errors that we see from the scan were thrown by the iterator or just
    // passed through from the source stream.
    return this.consume(function (err, x, push, next) {
        if (x === nil) {
            push(null, z);
            push(null, _.nil);
        }
        else if (err) {
            push(err);
            next();
        }
        else {
            try {
                z = f(z, x);
            }
            catch (e) {
                push(e);
                push(null, _.nil);
                return;
            }

            next();
        }
    });
});

/**
 * Same as [reduce](#reduce), but uses the first element as the initial
 * state instead of passing in a `memo` value.
 *
 * @id reduce1
 * @section Transforms
 * @name Stream.reduce1(iterator)
 * @param {Function} iterator - the function which reduces the values
 * @api public
 *
 * _([1, 2, 3, 4]).reduce1(add)  // => 10
 */

addMethod('reduce1', function (f) {
    var self = this;
    return self.createChild(function (push, next) {
        self.pull(function (err, x) {
            if (err) {
                push(err);
                next();
            }
            else if (x === nil) {
                push(null, nil);
            }
            else {
                next(self.reduce(f, x));
            }
        });
    });
});

/**
 * Groups all values into an Array and passes down the stream as a single
 * data event. This is a bit like doing [toArray](#toArray), but instead
 * of accepting a callback and consuming the stream, it passes the value on.
 *
 * @id collect
 * @section Transforms
 * @name Stream.collect()
 * @api public
 *
 * _(['foo', 'bar']).collect().toArray(function (xs) {
 *     // xs will be [['foo', 'bar']]
 * });
 */

addMethod('collect', function () {
    var xs = [];
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            push(null, xs);
            push(null, nil);
        }
        else {
            xs.push(x);
            next();
        }
    });
});

/**
 * Like [reduce](#reduce), but emits each intermediate value of the
 * reduction as it is calculated.
 *
 * If the iterator throws an error, the scan will stop and the stream will
 * emit that error. Any intermediate values that were produced before the
 * error will still be emitted.
 *
 * @id scan
 * @section Transforms
 * @name Stream.scan(iterator, memo)
 * @param memo - the initial state of the reduction
 * @param {Function} iterator - the function which reduces the values
 * @api public
 *
 * _([1, 2, 3, 4]).scan(add, 0)  // => 0, 1, 3, 6, 10
 */

addMethod('scan', function (f, z) {
    var self = this;
    return this.create([z]).concat(
        self.consume(function (err, x, push, next) {
            if (x === nil) {
                push(null, _.nil);
            }
            else if (err) {
                push(err);
                next();
            }
            else {
                try {
                    z = f(z, x);
                }
                catch (e) {
                    push(e);
                    push(null, _.nil);
                    return;
                }

                push(null, z);
                next();
            }
        })
    );
});

/**
 * Same as [scan](#scan), but uses the first element as the initial
 * state instead of passing in a `memo` value.
 *
 * @id scan1
 * @section Transforms
 * @name Stream.scan1(iterator)
 * @param {Function} iterator - the function which reduces the values
 * @api public
 *
 * _([1, 2, 3, 4]).scan1(add)  // => 1, 3, 6, 10
 */

addMethod('scan1', function (f) {
    var self = this;
    return self.createChild(function (push, next) {
        self.pull(function (err, x) {
            if (err) {
                push(err);
                next();
            }
            else if (x === nil) {
                push(null, nil);
            }
            else {
                next(self.scan(f, x));
            }
        });
    });
});

function HighlandTransform(push) {
    this.push = push;
}

HighlandTransform.prototype['@@transducer/init'] = function () {
    return this.push;
};

HighlandTransform.prototype['@@transducer/result'] = function (push) {
    // Don't push nil here. Otherwise, we can't catch errors from `result`
    // and propagate them. The `transduce` implementation will do it.
    return push;
};

HighlandTransform.prototype['@@transducer/step'] = function (push, input) {
    push(null, input);
    return push;
};

/**
 * Applies the transformation defined by the the given *transducer* to the
 * stream. A transducer is any function that follows the
 * [Transducer Protocol](https://github.com/cognitect-labs/transducers-js#transformer-protocol).
 * See
 * [transduce-js](https://github.com/cognitect-labs/transducers-js#transducers-js)
 * for more details on what transducers actually are.
 *
 * The `result` object that is passed in through the
 * [Transformer Protocol](https://github.com/cognitect-labs/transducers-js#transformer-protocol)
 * will be the `push` function provided by the [consume](#consume) transform.
 *
 * Like [scan](#scan), if the transducer throws an exception, the transform
 * will stop and emit that error. Any intermediate values that were produced
 * before the error will still be emitted.
 *
 * @id transduce
 * @section Transforms
 * @name Stream.transduce(xf)
 * @param {Function} xf - The transducer.
 * @api public
 *
 * var xf = require('transducer-js').map(_.add(1));
 * _([1, 2, 3, 4]).transduce(xf);
 * // => [2, 3, 4, 5]
 */

addMethod('transduce', function transduce(xf) {
    var transform = null,
        memo = null;

    return this.consume(function (err, x, push, next) {
        if (transform == null) {
            transform = xf(new HighlandTransform(push));
            memo = transform['@@transducer/init']();
        }

        if (err) {
            // Pass through errors, like we always do.
            push(err);
            next();
        }
        else if (x === _.nil) {
            // Push may be different from memo depending on the transducer that
            // we get.
            runResult(push, memo);
        }
        else {
            var res = runStep(push, memo, x);

            if (!res) {
                return;
            }

            memo = res;
            if (memo['@@transducer/reduced']) {
                runResult(memo['@@transducer/value']);
            }
            else {
                next();
            }
        }
    });

    function runResult(push, _memo) {
        try {
            transform['@@transducer/result'](_memo);
        }
        catch (e) {
            push(e);
        }
        push(null, _.nil);
    }

    function runStep(push, _memo, x) {
        try {
            return transform['@@transducer/step'](_memo, x);
        }
        catch (e) {
            push(e);
            push(null, _.nil);
            return null;
        }
    }
});

/**
 * Concatenates a Stream to the end of this Stream.
 *
 * Be aware that in the top-level export, the args may be in the reverse
 * order to what you'd expect `_([a], [b]) => b, a`, as this follows the
 * convention of other top-level exported functions which do `x` to `y`.
 *
 * @id concat
 * @section Higher-order Streams
 * @name Stream.concat(ys)
 * @param {Stream | Array} ys - the values to concatenate onto this Stream
 * @api public
 *
 * _([1, 2]).concat([3, 4])  // => 1, 2, 3, 4
 * _.concat([3, 4], [1, 2])  // => 1, 2, 3, 4
 */

addMethod('concat', function (ys) {
    ys = _(ys);
    return this.consume(function (err, x, push, next) {
        if (x === nil) {
            next(ys);
        }
        else {
            push(err, x);
            next();
        }
    });
});

/**
 * Takes a Stream of Streams and merges their values and errors into a
 * single new Stream. The merged stream ends when all source streams have
 * ended.
 *
 * Note that no guarantee is made with respect to the order in which
 * values for each stream end up in the merged stream. Values in the
 * merged stream will, however, respect the order they were emitted from
 * their respective streams.
 *
 * @id merge
 * @section Higher-order Streams
 * @name Stream.merge()
 * @api public
 *
 * var readFile = _.wrapCallback(fs.readFile);
 *
 * var txt = _(['foo.txt', 'bar.txt']).map(readFile)
 * var md = _(['baz.md']).map(readFile)
 *
 * _([txt, md]).merge();
 * // => contents of foo.txt, bar.txt and baz.txt in the order they were read
 */

addMethod('merge', function () {
    var self = this;
    var srcs = [];

    var srcsNeedPull = [],
        first = true,
        async = false;

    return this.create(function (push, next) {
        if (first) {
            first = false;
            getSourcesSync(push, next);
        }

        if (srcs.length === 0) {
            push(null, nil);
        }
        else if (srcsNeedPull.length) {
            pullFromAllSources(push, next);
            next();
        }
        else {
            async = true;
        }
    });

    // Make a handler for the main merge loop.
    function srcPullHandler(push, next, src) {
        return function (err, x) {
            if (err) {
                push(err);
                srcsNeedPull.push(src);
            }
            else if (x === nil) {
                srcs = srcs.filter(function (s) {
                    return s !== src;
                });
            }
            else {
                if (src === self) {
                    srcs.push(x);
                    srcsNeedPull.push(x);
                    srcsNeedPull.unshift(self);
                }
                else {
                    push(null, x);
                    srcsNeedPull.push(src);
                }
            }

            if (async) {
                async = false;
                next();
            }
        };
    }


    function pullFromAllSources(push, next) {
        var _srcs = srcsNeedPull;
        srcsNeedPull = [];
        _srcs.forEach(function (src) {
            src.pull(srcPullHandler(push, next, src));
        });
    }

    // Pulls as many sources as possible from self synchronously.
    function getSourcesSync(push, next) {
        // Shadows the outer async variable.
        var asynchronous;
        var done = false;

        var pull_cb = function(err, x) {
            asynchronous = false;
            if (done) {
                // This means the pull was async. Handle like
                // regular async.
                srcPullHandler(push, next, self)(err, x);
            }
            else {
                if (err) {
                    push(err);
                }
                else if (x === nil) {
                    done = true;
                }
                else {
                    srcs.push(x);
                    srcsNeedPull.push(x);
                }
            }
        };

        while (!done) {
            asynchronous = true;
            self.pull(pull_cb);

            // Async behavior, record self as a src and return.
            if (asynchronous) {
                done = true;
                srcs.unshift(self);
            }
        }
    }
});

/**
 * Takes a Stream of Streams and merges their values and errors into a
 * single new Stream, limitting the number of unpaused streams that can
 * running at any one time.
 *
 * Note that no guarantee is made with respect to the order in which
 * values for each stream end up in the merged stream. Values in the
 * merged stream will, however, respect the order they were emitted from
 * their respective streams.
 *
 * @id mergeWithLimit
 * @section Higher-order Streams
 * @name Stream.mergeWithLimit(n)
 * @param {Number} n - the maximum number of streams to run in parallel
 * @api public
 *
 * var readFile = _.wrapCallback(fs.readFile);
 *
 * var txt = _(['foo.txt', 'bar.txt']).flatMap(readFile)
 * var md = _(['baz.md']).flatMap(readFile)
 * var js = _(['bosh.js']).flatMap(readFile)
 *
 * _([txt, md, js]).mergeWithLimit(2);
 * // => contents of foo.txt, bar.txt, baz.txt and bosh.js in the order
 * // they were read, but bosh.js is not read until either foo.txt and bar.txt
 * // has completely been read or baz.md has been read
 */


addMethod('mergeWithLimit', function mergeWithLimit(n){
    var self = this;
    var processCount = 0;
    var waiting = false;
    if (typeof n !== 'number' || n < 1) {
        throw new Error('mergeWithLimit expects a positive number, but got: ' + n);
    }

    if (n === Infinity) {
        return this.merge();
    }
    return this.create(function(push, next){
        self.pull(function(err, x){
            var done = x === nil;
            if (err){
                push(err);
                next();
            }
            else if (x === nil) {
                push(null, nil);
            }
            else {
                processCount++;
                push(err, x);
                // console.log('start', x.id);
                x.onDestroy(function() {
                    processCount--;
                    // console.log('end', x.id);
                    if (waiting) {
                        // console.log('get more');
                        waiting = false;
                        next();
                    }
                });
                if (!done && processCount < n) {
                    next();
                }
                else {
                    // console.log('wait till something ends');
                    waiting = true;
                }
            }
        });
    }).merge();
});

/**
 * Calls a named method on each object from the Stream - returning
 * a new stream with the result of those calls.
 *
 * @id invoke
 * @section Transforms
 * @name Stream.invoke(method, args)
 * @param {String} method - the method name to call
 * @param {Array} args - the arguments to call the method with
 * @api public
 *
 * _(['foo', 'bar']).invoke('toUpperCase', [])  // => FOO, BAR
 *
 * var readFile = _.wrapCallback(fs.readFile);
 * filenames.flatMap(readFile).invoke('toString', ['utf8']);
 */

addMethod('invoke', function (method, args) {
    return this.map(function (x) {
        return x[method].apply(x, args);
    });
});

/**
 * Takes a Stream of callback-accepting node-style functions,
 * [wraps](#wrapCallback) each one into a stream-returning function,
 * calls them with the arguments provided, and returns the results
 * as a Stream.
 *
 * This can be used as a control flow shortcut and draws parallels
 * with some control flow functions from [async](https://github.com/caolan/async).
 * A few rough correspondences include:
 *
 * - `.nfcall([]).series()` to `async.series()`
 * - `.nfcall([]).parallel(n)` to `async.parallelLimit(n)`
 * - `.nfcall(args)` to `async.applyEach(..., args)`
 * - `.nfcall(args).series()` to `async.applyEachSeries(..., args)`
 *
 * @id nfcall
 * @section Transforms
 * @name Stream.nfcall(args)
 * @param {Array} args - the arguments to call each function with
 * @api public
 *
 * _([
 *   function (callback) {
 *     setTimeout(function () {
 *       callback(null, 'one');
 *     }, 200);
 *   },
 *   function (callback) {
 *     setTimeout(function () {
 *       callback(null, 'two');
 *     }, 100);
 *   }
 * ]).nfcall([]).parallel(2).toArray(function (xs) {
 *   // xs is ['one', 'two'] even though second function had a shorter timeout
 * });
 *
 * _([enableSearch, updateSchema]).nfcall(['bucket']).toArray(callback);
 * // does roughly the same as
 * async.applyEach([enableSearch, updateSchema], 'bucket', callback);
 *
 * _([
 *   fs.appendFile,
 *   fs.appendFile
 * ]).nfcall(['example.txt', 'hello']).series().toArray(function() {
 *   // example.txt now contains 'hellohello'
 * });
 *
 */

addMethod('nfcall', function (args) {
    var wrap = _.wrapCallback._relevel(this.create.bind(this));
    return this.map(function (x) {
        return wrap(x).apply(x, args);
    });
});

/**
 * Ensures that only one data event is push downstream (or into the buffer)
 * every `ms` milliseconds, any other values are dropped.
 *
 * @id throttle
 * @section Transforms
 * @name Stream.throttle(ms)
 * @param {Number} ms - the minimum milliseconds between each value
 * @api public
 *
 * _('mousemove', document).throttle(1000);
 */

addMethod('throttle', function (ms) {
    var last = 0 - ms;
    return this.consume(function (err, x, push, next) {
        var now = new Date().getTime();
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            push(null, nil);
        }
        else if (now - ms >= last) {
            last = now;
            push(null, x);
            next();
        }
        else {
            next();
        }
    });
});

/**
 * Holds off pushing data events downstream until there has been no more
 * data for `ms` milliseconds. Sends the last value that occurred before
 * the delay, discarding all other values.
 *
 * @id debounce
 * @section Transforms
 * @name Stream.debounce(ms)
 * @param {Number} ms - the milliseconds to wait before sending data
 * @api public
 *
 * // sends last keyup event after user has stopped typing for 1 second
 * $('keyup', textbox).debounce(1000);
 */

addMethod('debounce', function (ms) {
    var t = null;
    var nothing = {};
    var last = nothing;

    return this.consume(function (err, x, push, next) {
        if (err) {
            // let errors through regardless
            push(err);
            next();
        }
        else if (x === nil) {
            if (t) {
                clearTimeout(t);
            }
            if (last !== nothing) {
                push(null, last);
            }
            push(null, nil);
        }
        else {
            last = x;
            if (t) {
                clearTimeout(t);
            }
            t = setTimeout(push.bind(this, null, x), ms);
            next();
        }
    });
});

/**
 * Creates a new Stream, which when read from, only returns the last
 * seen value from the source. The source stream does not experience
 * back-pressure. Useful if you're using a Stream to model a changing
 * property which you need to query periodically.
 *
 * @id latest
 * @section Transforms
 * @name Stream.latest()
 * @api public
 *
 * // slowThing will always get the last known mouse position
 * // when it asks for more data from the mousePosition stream
 * mousePosition.latest().map(slowThing)
 */

addMethod('latest', function () {
    var nothing = {},
        latest = nothing,
        errors = [],
        ended = false,
        onValue = null;

    this.consume(function (err, x, push, next) {
        if (onValue != null) {
            var cb = onValue;
            onValue = null;
            cb(err, x);
        }

        if (err) {
            errors.push(err);
            next();
        }
        else if (x === nil) {
            ended = true;
        }
        else {
            latest = x;
            next();
        }
    }).resume();

    return this.create(function (push, next) {
        var oldErrors = errors;
        errors = [];

        if (!oldErrors.length && latest === nothing && !ended) {
            // We haven't gotten any data yet. We can't call next
            // because that might cause the stream to call the generator
            // again, resulting in an infinite loop. Thus, we stick a
            // a callback to be called whenever we get a value.
            onValue = function (err, x) {
                push(err, x);
                if (x !== nil) {
                    next();
                }
            };
        }
        else {
            oldErrors.forEach(push);
            if (latest !== nothing) {
                push(null, latest);
            }
            if (ended) {
                push(null, nil);
            }
            else {
                next();
            }
        }
    });
});

function StreamMultiplexer(stream) {
    this._stream = stream;

    this._consumers = new IntMap();
    this._requests = new IntMap();
    this._nonce = 0;
    this._emitting = false;
    this._repeatEmit = false;
    this._cached_value = null;
    this.paused = true;

    var self = this;
    this._pullCb = function _pullCb(err, x) {
        self.paused = true;
        if (self._requests.size === self._consumers.size) {
            self._send(err, x);
        }
        else {
            self._saved_token = [err, x];
        }
    };
}

/**
 * Emit downstream. The caller must guarantee that
 * this._consumers.size === this._requests.size.
 *
 * @param err - an error.
 * @param x - a value.
 */

StreamMultiplexer.prototype._send = function _send(err, x) {
    // Take a snapshot of the current request since calling the callbacks
    // may trigger more request to be registered.
    var requests = this._requests;
    this._requests = new IntMap();

    requests.forEach(function (requestCb) {
        requestCb(err, x);
    });
};

/**
 * Emit if we've met the backpressure requirements.
 */
StreamMultiplexer.prototype._resume = function _resume() {
    if (this._emitting) {
        this._repeatEmit = true;
        return;
    }

    this._emitting = true;
    do {
        // use a repeat flag to avoid recursing pull() calls
        this._repeatEmit = false;

        if (this._requests.size === this._consumers.size) {
            if (this._saved_token) {
                var token = this._saved_token;
                this._saved_token = null;
                this.paused = true;
                this._send(token[0], token[1]);
            }
            else if (this.paused) {
                this.paused = false;
                this._stream.pull(this._pullCb);
            }
        }
    } while (this._repeatEmit);
    this._emitting = false;
};

StreamMultiplexer.prototype.pull = function pull(id, cb) {
    if (!cb) {
        return;
    }

    if (this._requests.has(id)) {
        cb(new Error('Cannot service a second pull() request while one is in progress.'));
        return;
    }

    this._requests.set(id, cb);

    this._resume();
};

StreamMultiplexer.prototype.newStream = function newStream() {
    var self = this,
        id = self._nonce++;

    self._consumers.set(id, true);

    var pull = this.pull.bind(this, id);
    return new this._stream.constructor(newDelegateGenerator(pull))
        .onDestroy(function () {
            self.removeConsumer(id);
        });
};

StreamMultiplexer.prototype.removeConsumer = function removeConsumer(id) {
    if (!this._consumers.has(id)) {
        return;
    }

    this._consumers.delete(id);
    this._requests.delete(id);

    this._resume();
};

/**
 * Returns values from an Object as a Stream. Reads properties
 * lazily, so if you don't read from all keys on an object, not
 * all properties will be read from (may have an effect where getters
 * are used).
 *
 * @id values
 * @section Objects
 * @name _.values(obj)
 * @param {Object} obj - the object to return values from
 * @api public
 *
 * _.values({foo: 1, bar: 2, baz: 3})  // => 1, 2, 3
 */

_.values = function (obj) {
    return _.keys(obj).map(function (k) {
        return obj[k];
    });
};

/**
 * Returns keys from an Object as a Stream.
 *
 * @id keys
 * @section Objects
 * @name _.keys(obj)
 * @param {Object} obj - the object to return keys from
 * @api public
 *
 * _.keys({foo: 1, bar: 2, baz: 3})  // => 'foo', 'bar', 'baz'
 */

function keys (obj) {
    var keysArray = [];
    for (var k in obj) {
        if (hasOwn.call(obj, k)) {
            keysArray.push(k);
        }
    }
    return keysArray;
}

_.keys = function (obj) {
    return _(keys(obj));
};

/**
 * Returns key/value pairs for an Object as a Stream. Reads properties
 * lazily, so if you don't read from all keys on an object, not
 * all properties will be read from (may have an effect where getters
 * are used).
 *
 * @id pairs
 * @section Objects
 * @name _.pairs(obj)
 * @param {Object} obj - the object to return key/value pairs from
 * @api public
 *
 * _.pairs({foo: 1, bar: 2})  // => ['foo', 1], ['bar', 2]
 */

_.pairs = function (obj) {
    return _.keys(obj).map(function (k) {
        return [k, obj[k]];
    });
};

/**
 * Extends one object with the properties of another. **Note:** The
 * arguments are in the reverse order of other libraries such as
 * underscore. This is so it follows the convention of other functions in
 * this library and so you can more meaningfully partially apply it.
 *
 * @id extend
 * @section Objects
 * @name _.extend(a, b)
 * @param {Object} a - the properties to extend b with
 * @param {Object} b - the original object to extend
 * @api public
 *
 * _.extend({name: 'bar'}, {name: 'foo', price: 20})
 * // => {name: 'bar', price: 20}
 *
 * // example of partial application
 * var publish = _.extend({published: true});
 *
 * publish({title: 'test post'})
 * // => {title: 'test post', published: true}
 */

_.extend = _.curry(function (extensions, target) {
    for (var k in extensions) {
        if (hasOwn.call(extensions, k)) {
            target[k] = extensions[k];
        }
    }
    return target;
});

/**
 * Returns a property from an object.
 *
 * @id get
 * @section Objects
 * @name _.get(prop, obj)
 * @param {String} prop - the property to return
 * @param {Object} obj - the object to read properties from
 * @api public
 *
 * var obj = {foo: 'bar', baz: 123};
 * _.get('foo', obj) // => 'bar'
 *
 * // making use of partial application
 * var posts = [
 *   {title: 'one'},
 *   {title: 'two'},
 *   {title: 'three'}
 * ];
 *
 * _(posts).map(_.get('title'))  // => 'one', 'two', 'three'
 */

_.get = _.curry(function (prop, obj) {
    return obj[prop];
});

/**
 * Updates a property on an object, returning the updated object.
 *
 * @id set
 * @section Objects
 * @name _.set(prop, value, obj)
 * @param {String} prop - the property to return
 * @param value - the value to set the property to
 * @param {Object} obj - the object to set properties on
 * @api public
 *
 * var obj = {foo: 'bar', baz: 123};
 * _.set('foo', 'wheeee', obj) // => {foo: 'wheeee', baz: 123}
 *
 * // making use of partial application
 * var publish = _.set('published', true);
 *
 * publish({title: 'example'})  // => {title: 'example', published: true}
 */

_.set = _.curry(function (prop, val, obj) {
    obj[prop] = val;
    return obj;
});

/**
 * Logs values to the console, a simple wrapper around `console.log` that
 * it suitable for passing to other functions by reference without having to
 * call `bind`.
 *
 * @id log
 * @section Utils
 * @name _.log(args..)
 * @api public
 *
 * _.log('Hello, world!');
 *
 * _([1, 2, 3, 4]).each(_.log);
 */

_.log = function () {
    console.log.apply(console, arguments);
};

/**
 * Wraps a node-style async function which accepts a callback, transforming
 * it to a function which accepts the same arguments minus the callback and
 * returns a Highland Stream instead. The wrapped function keeps its context,
 * so you can safely use it as a method without binding (see the second
 * example below).
 *
 * `wrapCallback` also accepts an optional `mappingHint`, which specifies how
 * callback arguments are pushed to the stream. This can be used to handle
 * non-standard callback protocols that pass back more than one value.
 *
 * `mappingHint` can be a function, number, or array. See the documentation on
 * [EventEmitter Stream Objects](#Stream Objects) for details on the mapping
 * hint. If `mappingHint` is a function, it will be called with all but the
 * first argument that is passed to the callback. The first is still assumed to
 * be the error argument.
 *
 * @id wrapCallback
 * @section Utils
 * @name _.wrapCallback(f)
 * @param {Function} f - the node-style function to wrap
 * @param {Array | Function | Number} mappingHint - (optional) how to pass the
 * arguments to the callback
 * @api public
 *
 * var fs = require('fs');
 *
 * var readFile = _.wrapCallback(fs.readFile);
 *
 * readFile('example.txt').apply(function (data) {
 *     // data is now the contents of example.txt
 * });
 *
 * function Reader(file) {
 *     this.file = file;
 * }
 *
 * Reader.prototype.read = function(cb) {
 *     fs.readFile(this.file, cb);
 * };
 *
 * Reader.prototype.readStream = _.wrapCallback(Reader.prototype.read);
 */

/*eslint-disable no-multi-spaces */
addToplevelMethod('wrapCallback', function (f, /*optional*/mappingHint) {
    /*eslint-enable no-multi-spaces */
    var stream = this;
    var mapper = hintMapper(mappingHint);

    return function () {
        var self = this;
        var args = slice.call(arguments);
        return stream(function (push) {
            var cb = function (err) {
                if (err) {
                    push(err);
                }
                else {
                    var cbArgs = slice.call(arguments, 1);
                    var v = mapper.apply(this, cbArgs);
                    push(null, v);
                }
                push(null, nil);
            };
            f.apply(self, args.concat([cb]));
        });
    };
});

/**
 * Takes an object or a constructor function and returns that object or
 * constructor with streamified versions of its function properties.
 * Passed constructors will also have their prototype functions
 * streamified.  This is useful for wrapping many node style async
 * functions at once, and for preserving those functions' context.
 *
 * @id streamifyAll
 * @section Utils
 * @name _.streamifyAll(source)
 * @param {Object | Function} source - the function or object with
 * node-style function properties.
 * @api public
 *
 * var fs = _.streamifyAll(require('fs'));
 *
 * fs.readFileStream('example.txt').apply(function (data) {
 *     // data is now the contents of example.txt
 * });
 */

function isClass (fn) {
    if (!(typeof fn === 'function' && fn.prototype)) { return false; }
    var getKeys = isES5 ? Object.getOwnPropertyNames : keys;
    var allKeys = getKeys(fn.prototype);
    return allKeys.length > 0 && !(allKeys.length === 1 &&
            allKeys[0] === 'constructor');
}

function inheritedKeys (obj) {
    var allProps = {};
    var curr = obj;
    var handleProp = function (prop) {
        allProps[prop] = true;
    };
    while (Object.getPrototypeOf(curr)) {
        var props = Object.getOwnPropertyNames(curr);
        props.forEach(handleProp);
        curr = Object.getPrototypeOf(curr);
    }
    return keys(allProps);
}

function streamifyAll (wrapCallback, inp, suffix) {
    // will not streamify inherited functions in ES3
    var getKeys = isES5 ? inheritedKeys : keys;
    var allKeys = getKeys(inp);

    for (var i = 0, len = allKeys.length; i < len; i++) {
        var key = allKeys[i];
        var val;

        // will skip context aware getters
        try {
            val = inp[key];
        }
        catch (e) {
            // Ignore
        }

        if (val && typeof val === 'function' && !isClass(val) &&
                !val.__HighlandStreamifiedFunction__) {

            var streamified = wrapCallback(val);
            streamified.__HighlandStreamifiedFunction__ = true;
            inp[key + suffix] = streamified;
        }
    }
    return inp;
}

addToplevelMethod('streamifyAll', function (arg) {
    if (typeof arg !== 'function' && typeof arg !== 'object') {
        throw new TypeError('takes an object or a constructor function');
    }
    var suffix = 'Stream';

    var ret = streamifyAll(this.wrapCallback, arg, suffix);
    if (isClass(arg)) {
        ret.prototype = streamifyAll(this.wrapCallback, arg.prototype, suffix);
    }
    return ret;
});

/**
 * Add two values. Can be partially applied.
 *
 * @id add
 * @section Operators
 * @name _.add(a, b)
 * @api public
 *
 * _.add(1, 2) === 3
 * _.add(1)(5) === 6
 */

_.add = _.curry(function (a, b) {
    return a + b;
});

/**
 * Perform logical negation on a value. If `x` is truthy then returns false,
 * otherwise returns true.
 *
 * @id not
 * @section Operators
 * @name _.not(x)
 * @param x - the value to negate
 * @api public
 *
 * _.not(true)   // => false
 * _.not(false)  // => true
 */

_.not = function (x) {
    return !x;
};