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


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


/**
 * 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. You call `next()`
 * to signal you've finished processing the current data. 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.
 *
 * **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 also pass as an optional third parameter a function, an array of strings
 * or a number. In this case the event handler will try to wrap the arguments emitted
 * to it and write this object to the new stream.
 *
 * **Promise -** Accepts an ES6 / jQuery style promise and returns a
 * Highland Stream which will emit a single value (or an error).
 *
 * @id _(source)
 * @section Streams
 * @name _(source)
 * @param {Array | Function | Readable Stream | Promise} source - (optional) source to take values from from
 * @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);
 *
 * // creating a stream from events
 * _('click', btn).each(handleEvent);
 *
 * // creating a stream from events with mapping
 * _('request', httpServer, ['req', 'res']).each(handleEvent);
 *
 * // from a Promise object
 * var foo = _($.getJSON('/api/foo'));
 */

exports = module.exports = function (/*optional*/xs, /*optional*/ee, /*optional*/ mappingHint) {
    return new Stream(xs, ee, mappingHint);
};

var _ = exports;


// Save bytes in the minified (but not gzipped) version:
var ArrayProto = Array.prototype,
    ObjProto = Object.prototype;

// Create quick reference variables for speed access to core prototypes.
var slice = ArrayProto.slice,
    toString = ObjProto.toString;


_.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 toString.call(x) === '[object Array]';
};

// setImmediate implementation with browser and older node fallbacks
if (typeof setImmediate === 'undefined') {
    if (typeof process === 'undefined' || !(process.nextTick)) {
        _.setImmediate = function (fn) {
            setTimeout(fn, 0);
        };
    }
    else {
        // use nextTick on old node versions
        _.setImmediate = process.nextTick;
    }
}
else if (typeof process === 'undefined' || !(process.stdout) /*browserify*/) {
    // modern browser - but not a direct alias for IE10 compatibility
    _.setImmediate = function (fn) {
        setImmediate(fn);
    };
}
else {
    _.setImmediate = setImmediate;
}


/**
 * 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 Streams
 * @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)
var _global = this;
if (typeof global !== 'undefined') {
    _global = global;
}
else if (typeof window !== 'undefined') {
    _global = window;
}
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 function () {
        var args = largs.concat(slice.call(arguments));
        if (args.length < n) {
            return _.ncurry.apply(this, [n, fn].concat(args));
        }
        return fn.apply(this, args.slice(0, n));
    };
};

/**
 * 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} f - 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. 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;
        }
        var r = fns[0].apply(this, arguments);
        for (var i = 1; i < fns.length; i++) {
            r = fns[i].call(this, r);
        }
        return r;
    };
};

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

function Stream(/*optional*/xs, /*optional*/ee, /*optional*/mappingHint) {
    if (xs && _.isStream(xs)) {
        // already a Stream
        return xs;
    }

    EventEmitter.call(this);
    var self = 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
    self.__HighlandStream__ = true;

    self.id = ('' + Math.random()).substr(2, 6);
    this.paused = true;
    this._incoming = [];
    this._outgoing = [];
    this._consumers = [];
    this._observers = [];
    this._destructors = [];
    this._send_events = false;
    this.source = null;

    // Old-style node Stream.pipe() checks for this
    this.writable = true;

    self.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
    self.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;
            }
        }
    });

    if (xs === undefined) {
        // nothing else to do
    }
    else if (Array.isArray(xs)) {
        self._incoming = xs.concat([nil]);
    }
    else if (typeof xs === 'function') {
        this._generator = xs;
        this._generator_push = function (err, x) {
            self.write(err ? new StreamError(err): x);
        };
        this._generator_next = function (s) {
            if (s) {
                // we MUST pause to get the redirect object into the _incoming
                // buffer otherwise it would be passed directly to _send(),
                // which does not handle StreamRedirect objects!
                var _paused = self.paused;
                if (!_paused) {
                    self.pause();
                }
                self.write(new StreamRedirect(s));
                if (!_paused) {
                    self.resume();
                }
            }
            else {
                self._generator_running = false;
            }
            if (!self.paused) {
                self.resume();
            }
        };
    }
    else if (_.isObject(xs)) {
        if (_.isFunction(xs.then)) {
            // probably a promise
            return _(function (push) {
                xs.then(function (value) {
                    push(null, value);
                    return push(null, nil);
                },
                function (err) {
                    push(err);
                    return push(null, nil);
                });
            });
        }
        else {
            // assume it's a pipeable stream as a source
            this._generator = function () {
                delete self._generator;
                xs.pipe(self);
            };
        }
    }
    else if (typeof xs === 'string') {
        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; };
        }

        ee.on(xs, function () {
            var ctx = mapper.apply(this, arguments);
            self.write(ctx);
        });
    }
    else {
        throw new Error(
            'Unexpected argument type to Stream(): ' + (typeof xs)
        );
    }
}
inherits(Stream, EventEmitter);

/**
 * adds a top-level _.foo(mystream) style export for Stream methods
 */

function exposeMethod(name) {
    var f = Stream.prototype[name];
    var n = f.length;
    _[name] = _.ncurry(n + 1, function () {
        var args = Array.prototype.slice.call(arguments);
        var s = _(args.pop());
        return f.apply(s, args);
    });
}

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

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

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

function StreamRedirect(to) {
    this.__HighlandStreamRedirect__ = true;
    this.to = to;
}

/**
 * Returns true if `x` is a Highland Stream.
 *
 * @id isStream
 * @section Streams
 * @name _.isStream(x)
 * @param x - the object to test
 * @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 (err, x) {
    if (x === nil) {
        this.ended = true;
    }
    if (this._consumers.length) {
        for (var i = 0, len = this._consumers.length; i < len; i++) {
            var c = this._consumers[i];
            if (err) {
                c.write(new StreamError(err));
            }
            else {
                c.write(x);
            }
        }
    }
    if (this._observers.length) {
        for (var j = 0, len2 = this._observers.length; j < len2; j++) {
            this._observers[j].write(x);
        }
    }
    if (this._send_events) {
        if (x === nil) {
            this.emit('end');
        }
        else {
            this.emit('data', x);
        }
    }
};

/**
 * Pauses the stream. All Highland Streams start in the paused state.
 *
 * @id pause
 * @section Streams
 * @name Stream.pause()
 * @api public
 *
 * var xs = _(generator);
 * xs.pause();
 */

Stream.prototype.pause = function () {
    //console.log([this.id, 'pause']);
    this.paused = true;
    if (this.source) {
        this.source._checkBackPressure();
    }
};

/**
 * When there is a change in downstream consumers, it will often ask
 * the parent Stream to re-check it's state and pause/resume accordingly.
 */

Stream.prototype._checkBackPressure = function () {
    if (!this._consumers.length) {
        return this.pause();
    }
    for (var i = 0, len = this._consumers.length; i < len; i++) {
        if (this._consumers[i].paused) {
            return this.pause();
        }
    }
    return this.resume();
};

/**
 * Starts pull values out of the incoming buffer and sending them downstream,
 * this will exit early if this causes a downstream consumer to pause.
 */

Stream.prototype._readFromBuffer = function () {
    var len = this._incoming.length;
    var i = 0;
    while (i < len && !this.paused) {
        var x = this._incoming[i];
        if (_._isStreamError(x)) {
            this._send(x.error);
        }
        else if (_._isStreamRedirect(x)) {
            this._redirect(x.to);
        }
        else {
            this._send(null, x);
        }
        i++;
    }
    // remove processed data from _incoming buffer
    this._incoming.splice(0, i);
};

/**
 * Starts pull values out of the incoming buffer and sending them downstream,
 * this will exit early if this causes a downstream consumer to pause.
 */

Stream.prototype._sendOutgoing = function () {
    var len = this._outgoing.length;
    var i = 0;
    while (i < len && !this.paused) {
        var x = this._outgoing[i];
        if (_._isStreamError(x)) {
            Stream.prototype._send.call(this, x.error);
        }
        else if (_._isStreamRedirect(x)) {
            this._redirect(x.to);
        }
        else {
            Stream.prototype._send.call(this, null, x);
        }
        i++;
    }
    // remove processed data from _outgoing buffer
    this._outgoing.splice(0, i);
};

/**
 * Resumes a paused Stream. This will either read from the Stream's incoming
 * buffer or request more data from an upstream source.
 *
 * @id resume
 * @section Streams
 * @name Stream.resume()
 * @api public
 *
 * var xs = _(generator);
 * xs.resume();
 */

Stream.prototype.resume = function () {
    //console.log([this.id, 'resume']);
    if (this._resume_running) {
        // already processing _incoming buffer, ignore resume call
        this._repeat_resume = true;
        return;
    }
    this._resume_running = true;
    do {
        // use a repeat flag to avoid recursing resume() calls
        this._repeat_resume = false;
        this.paused = false;

        // send values from outgoing buffer first
        this._sendOutgoing();

        // send values from incoming buffer before reading from source
        this._readFromBuffer();

        // we may have paused while reading from buffer
        if (!this.paused) {
            // ask parent for more data
            if (this.source) {
                this.source._checkBackPressure();
            }
            // run _generator to fill up _incoming buffer
            else if (this._generator) {
                this._runGenerator();
            }
            else {
                // perhaps a node stream is being piped in
                this.emit('drain');
            }
        }
    } while (this._repeat_resume);
    this._resume_running = false;
};

/**
 * 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.
 *
 * @id end
 * @section Streams
 * @name Stream.end()
 * @api public
 *
 * mystream.end();
 */

Stream.prototype.end = function () {
    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.
 *
 * This function returns the destination so you can chain together pipe calls.
 *
 * @id pipe
 * @section Streams
 * @name Stream.pipe(dest)
 * @param {Writable Stream} dest - the destination to write all data to
 * @api public
 *
 * var source = _(generator);
 * var dest = fs.createWriteStream('myfile.txt')
 * source.pipe(dest);
 *
 * // chained call
 * source.pipe(through).pipe(dest);
 */

Stream.prototype.pipe = function (dest) {
    var self = this;

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

    var s = self.consume(function (err, x, push, next) {
        if (err) {
            self.emit('error', err);
            return;
        }
        if (x === nil) {
            if (canClose) {
                dest.end();
            }
        }
        else if (dest.write(x) !== false) {
            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.
    this._destructors.push(function () {
        dest.removeListener('drain', onConsumerDrain);
    });

    s.resume();
    return dest;

    function onConsumerDrain() {
        s.resume();
    }
};

/**
 * 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 Streams
 * @name Stream.destroy()
 * @api public
 */

Stream.prototype.destroy = function () {
    var self = this;
    this.end();
    _(this._consumers).each(function (consumer) {
        self._removeConsumer(consumer);
    });
    if (this.source) {
        this.source._removeConsumer(this);
    }
    _(this._destructors).each(function (destructor) {
        destructor();
    });
};

/**
 * 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 () {
    // if _generator already running, exit
    if (this._generator_running) {
        return;
    }
    this._generator_running = true;
    this._generator(this._generator_push, this._generator_next);
};

/**
 * Performs the redirect from one Stream to another. In order for the
 * redirect to happen at the appropriate time, it is put on the incoming
 * buffer as a StreamRedirect object, and this function is called
 * once it is read from the buffer.
 */

Stream.prototype._redirect = function (to) {
    // coerce to Stream
    to = _(to);

    to._consumers = this._consumers.map(function (c) {
        c.source = to;
        return c;
    });
    // TODO: copy _observers
    this._consumers = [];
    this.consume = function () {
        return to.consume.apply(to, arguments);
    };
    this._removeConsumer = function () {
        return to._removeConsumer.apply(to, arguments);
    };
    if (this.paused) {
        to.pause();
    }
    else {
        this.pause();
        to._checkBackPressure();
    }
};

/**
 * Adds a new consumer Stream, which will accept data and provide backpressure
 * to this Stream. Adding more than one consumer will cause an exception to be
 * thrown as the backpressure strategy must be explicitly chosen by the
 * developer (through calling fork or observe).
 */

Stream.prototype._addConsumer = function (s) {
    if (this._consumers.length) {
        throw new Error(
            'Stream already being consumed, you must either fork() or observe()'
        );
    }
    s.source = this;
    this._consumers.push(s);
    this._checkBackPressure();
};

/**
 * Removes a consumer from this Stream.
 */

Stream.prototype._removeConsumer = function (s) {
    this._consumers = this._consumers.filter(function (c) {
        return c !== s;
    });
    if (s.source === this) {
        s.source = null;
    }
    this._checkBackPressure();
};

/**
 * 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 Streams
 * @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();
 *         }
 *     });
 * };
 */

Stream.prototype.consume = function (f) {
    var self = this;
    var s = new Stream();
    var _send = s._send;
    var push = function (err, x) {
        //console.log(['push', err, x, s.paused]);
        if (x === nil) {
            // ended, remove consumer from source
            self._removeConsumer(s);
        }
        if (s.paused) {
            if (err) {
                s._outgoing.push(new StreamError(err));
            }
            else {
                s._outgoing.push(x);
            }
        }
        else {
            _send.call(s, err, x);
        }
    };
    var async;
    var next_called;
    var next = function (s2) {
        //console.log(['next', async]);
        if (s2) {
            // we MUST pause to get the redirect object into the _incoming
            // buffer otherwise it would be passed directly to _send(),
            // which does not handle StreamRedirect objects!
            var _paused = s.paused;
            if (!_paused) {
                s.pause();
            }
            s.write(new StreamRedirect(s2));
            if (!_paused) {
                s.resume();
            }
        }
        else if (async) {
            s.resume();
        }
        else {
            next_called = true;
        }
    };
    s._send = function (err, x) {
        async = false;
        next_called = false;
        f(err, x, push, next);
        async = true;
        if (!next_called) {
            s.pause();
        }
    };
    self._addConsumer(s);
    return s;
};

/**
 * 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 Streams
 * @name Stream.pull(f)
 * @param {Function} f - the function to handle data
 * @api public
 *
 * xs.pull(function (err, x) {
 *     // do something
 * });
 */

Stream.prototype.pull = function (f) {
    var s = this.consume(function (err, x) {
        s.source._removeConsumer(s);
        f(err, x);
    });
    s.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).
 *
 * @id write
 * @section Streams
 * @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]
 * });
 */

Stream.prototype.write = function (x) {
    if (this.paused) {
        this._incoming.push(x);
    }
    else {
        if (_._isStreamError(x)) {
            this._send(x.error);
        }
        else {
            this._send(null, 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 it's source as fast as the slowest consumer can handle them.
 *
 * @id fork
 * @section 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();
 */

Stream.prototype.fork = function () {
    var s = new Stream();
    s.id = 'fork:' + s.id;
    s.source = this;
    this._consumers.push(s);
    this._checkBackPressure();
    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 it's
 * buffer and cause memory issues. Where possible, you should use [fork](#fork).
 *
 * @id observe
 * @section 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 recieve data as fast as ys can handle it
 * ys.resume();
 */

Stream.prototype.observe = function () {
    var s = new Stream();
    s.id = 'observe:' + s.id;
    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 Streams
 * @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);
 *     }
 * });
 */

Stream.prototype.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();
        }
    });
};
exposeMethod('errors');

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

Stream.prototype.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();
        }
    });
};
exposeMethod('stopOnError');

/**
 * Iterates over every value from the Stream, calling the iterator function
 * on each of them. This function causes a **thunk**.
 *
 * If an error from the Stream reaches the `each` call, it will emit an
 * error event (which will cause it to throw if unhandled).
 *
 * @id each
 * @section Streams
 * @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
 * });
 */

Stream.prototype.each = function (f) {
    var self = this;
    return this.consume(function (err, x, push, next) {
        if (err) {
            self.emit('error', err);
        }
        else if (x !== nil) {
            f(x);
            next();
        }
    }).resume();
};
exposeMethod('each');

/**
 * Applies results from a Stream as arguments to a function
 *
 * @id apply
 * @section Streams
 * @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
 * });
 */

Stream.prototype.apply = function (f) {
    return this.toArray(function (args) {
        f.apply(null, args);
    });
};
exposeMethod('apply');

/**
 * Collects all values from a Stream into an Array and calls a function with
 * once with the result. This function causes a **thunk**.
 *
 * If an error from the Stream reaches the `toArray` call, it will emit an
 * error event (which will cause it to throw if unhandled).
 *
 * @id toArray
 * @section Streams
 * @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]
 * });
 */

// TODO: implement using collect()?
Stream.prototype.toArray = function (f) {
    var self = this;
    var xs = [];
    return this.consume(function (err, x, push, next) {
        if (err) {
            self.emit('error', err);
        }
        else if (x === nil) {
            f(xs);
        }
        else {
            xs.push(x);
            next();
        }
    }).resume();
};

/**
 * Creates a new Stream of transformed values by applying a function to each
 * value from the source. The transformation function can be replaced with
 * a non-function value for convenience, and it will emit that value
 * for every data event on the source Stream.
 *
 * @id map
 * @section Streams
 * @name Stream.map(f)
 * @param f - the transformation function or value to map to
 * @api public
 *
 * var doubled = _([1, 2, 3, 4]).map(function (x) {
 *     return x * 2;
 * });
 *
 * _([1, 2, 3]).map('hi')  // => 'hi', 'hi', 'hi'
 */

Stream.prototype.map = function (f) {
    if (!_.isFunction(f)) {
        var val = f;
        f = function () {
            return val;
        };
    }
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            push(err, x);
        }
        else {
            push(null, f(x));
            next();
        }
    });
};
exposeMethod('map');

/**
 * Creates a new Stream of values by applying each item in a Stream to an
 * iterator function which may return a Stream. Each item on these result
 * Streams are then emitted on a single output Stream.
 *
 * The same as calling `stream.map(f).flatten()`.
 *
 * @id flatMap
 * @section Streams
 * @name Stream.flatMap(f)
 * @param {Function} f - the iterator function
 * @api public
 *
 * filenames.flatMap(readFile)
 */

Stream.prototype.flatMap = function (f) {
    return this.map(f).flatten();
};
exposeMethod('flatMap');

/**
 * Retrieves values associated with a given property from all elements in
 * the collection.
 *
 * @id pluck
 * @section Streams
 * @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']
 * });
 */

Stream.prototype.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();
        }
    });
};
exposeMethod('pluck');

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

Stream.prototype.filter = function (f) {
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            push(err, x);
        }
        else {
            if (f(x)) {
                push(null, x);
            }
            next();
        }
    });
};
exposeMethod('filter');

/**
 * 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 it's first value (all other values on the Stream will be
 * disregarded).
 *
 * @id flatFilter
 * @section Streams
 * @name Stream.flatFilter(f)
 * @param {Function} f - the truth test function which returns a Stream
 * @api public
 *
 * var checkExists = _.wrapCallback(fs.exists);
 * filenames.flatFilter(checkExists)
 */

Stream.prototype.flatFilter = function (f) {
    var xs = this.observe();
    var ys = this.flatMap(function (x) {
        return f(x).take(1);
    });
    return xs.zip(ys)
        .filter(function (pair) {
            return pair[1];
        })
        .map(function (pair) {
            return pair[0];
        });
};
exposeMethod('flatFilter');

/**
 * The inverse of [filter](#filter).
 *
 * @id reject
 * @section Streams
 * @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;
 * });
 */

Stream.prototype.reject = function (f) {
    return this.filter(_.compose(_.not, f));
};
exposeMethod('reject');

/**
 * A convenient form of filter, which returns the first object from a
 * Stream that passes the provided truth test
 *
 * @id find
 * @section Streams
 * @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'}]
 */

Stream.prototype.find = function (f) {
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            push(err, x);
        }
        else {
            if (f(x)) {
                push(null, x);
                push(null, nil);
            }
            next();
        }
    }.bind(this));
};
exposeMethod('find');

/**
 * A convenient form of reduce, which groups items based on a function or property name
 *
 * @id group
 * @section Streams
 * @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'}]
 * // =>  }
 *
 */

Stream.prototype.group = function (f) {
    var lambda = _.isString(f) ? _.get(f) : f;
    return this.reduce({}, function (m, o) {
        var key = lambda(o);
        if (!m.hasOwnProperty(key)) { m[key] = []; }
        m[key].push(o);
        return m;
    }.bind(this));
};
exposeMethod('group');

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

Stream.prototype.compact = function () {
    return this.filter(function (x) {
        return x;
    });
};
exposeMethod('compact');

/**
 * A convenient form of filter, which returns all objects from a Stream
 * match a set of property values.
 *
 * @id where
 * @section Streams
 * @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'}
 */

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

/**
 * Takes two Streams and returns a Stream of corresponding pairs.
 *
 * @id zip
 * @section 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]
 */

Stream.prototype.zip = function (ys) {
    ys = _(ys);
    var xs = this;
    var returned = 0;
    var z = [];
    function nextValue(index, max, src, push, next) {
        src.pull(function (err, x) {
            if (err) {
                push(null, err);
                nextValue(index, max, src, push, next);
            }
            else if (x === _.nil) {
                push(null, nil);
            }
            else {
                returned++;
                z[index] = x;
                if (returned === max) {
                    push(null, z);
                    next();
                }
            }
        });
    }
    return _(function (push, next) {
        returned = 0;
        z = [];
        nextValue(0, 2, xs, push, next);
        nextValue(1, 2, ys, push, next);
    });
};
exposeMethod('zip');

/**
 * Creates a new Stream with the first `n` values from the source.
 *
 * @id take
 * @section Streams
 * @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
 */

// TODO: test that errors don't count in take 'n' calls
Stream.prototype.take = function (n) {
    if (n === 0) {
        return _([]);
    }
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            if (n > 0) {
                next();
            }
            else {
                push(null, nil);
            }
        }
        else if (x === nil) {
            push(null, nil);
        }
        else {
            n--;
            push(null, x);
            if (n > 0) {
                next();
            }
            else {
                push(null, nil);
            }
        }
    });
};
exposeMethod('take');

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

Stream.prototype.head = function () {
    return this.take(1);
};
exposeMethod('head');

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

Stream.prototype.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();
        }
    });
};
exposeMethod('last');

/**
 * Passes the current Stream to a function, returning the result. Can also
 * be used to pipe the current Stream through another Stream. It will always
 * return a Highland Stream (instead of the piped to target directly as in
 * Node.js).
 *
 * @id through
 * @section Streams
 * @name Stream.through(target)
 * @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) {
 *     // ...
 * });
 */

Stream.prototype.through = function (target) {
    if (_.isFunction (target)) {
        return target(this);
    }
    else {
        var output = _();
        target.pause();
        this.pipe(target).pipe(output);
        return output;
    }
};
exposeMethod('through');

_.pipeline = function (/*through...*/) {
    var start = _(arguments[0]);
    var rest = slice.call(arguments, 1);
    var end = rest.reduce(function (src, dest) {
        return src.through(dest);
    }, start);
    var wrapper = _(function (push, next) {
        end.pull(function (err, x) {
            if (err) {
                wrapper._send(err);
                next();
            }
            else if (x === nil) {
                wrapper._send(null, nil);
            }
            else {
                wrapper._send(null, x);
                next();
            }
        });
    });
    wrapper.write = function (x) {
        start.write(x);
    };<commit_msg>add some docs for pipeline<commit_after>/**
 * Highland: the high-level streams library
 *
 * Highland may be freely distributed under the Apache 2.0 license.
 * http://github.com/caolan/highland
 * Copyright (c) <NAME>
 *
 */


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


/**
 * 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. You call `next()`
 * to signal you've finished processing the current data. 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.
 *
 * **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 also pass as an optional third parameter a function, an array of strings
 * or a number. In this case the event handler will try to wrap the arguments emitted
 * to it and write this object to the new stream.
 *
 * **Promise -** Accepts an ES6 / jQuery style promise and returns a
 * Highland Stream which will emit a single value (or an error).
 *
 * @id _(source)
 * @section Streams
 * @name _(source)
 * @param {Array | Function | Readable Stream | Promise} source - (optional) source to take values from from
 * @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);
 *
 * // creating a stream from events
 * _('click', btn).each(handleEvent);
 *
 * // creating a stream from events with mapping
 * _('request', httpServer, ['req', 'res']).each(handleEvent);
 *
 * // from a Promise object
 * var foo = _($.getJSON('/api/foo'));
 */

exports = module.exports = function (/*optional*/xs, /*optional*/ee, /*optional*/ mappingHint) {
    return new Stream(xs, ee, mappingHint);
};

var _ = exports;


// Save bytes in the minified (but not gzipped) version:
var ArrayProto = Array.prototype,
    ObjProto = Object.prototype;

// Create quick reference variables for speed access to core prototypes.
var slice = ArrayProto.slice,
    toString = ObjProto.toString;


_.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 toString.call(x) === '[object Array]';
};

// setImmediate implementation with browser and older node fallbacks
if (typeof setImmediate === 'undefined') {
    if (typeof process === 'undefined' || !(process.nextTick)) {
        _.setImmediate = function (fn) {
            setTimeout(fn, 0);
        };
    }
    else {
        // use nextTick on old node versions
        _.setImmediate = process.nextTick;
    }
}
else if (typeof process === 'undefined' || !(process.stdout) /*browserify*/) {
    // modern browser - but not a direct alias for IE10 compatibility
    _.setImmediate = function (fn) {
        setImmediate(fn);
    };
}
else {
    _.setImmediate = setImmediate;
}


/**
 * 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 Streams
 * @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)
var _global = this;
if (typeof global !== 'undefined') {
    _global = global;
}
else if (typeof window !== 'undefined') {
    _global = window;
}
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 function () {
        var args = largs.concat(slice.call(arguments));
        if (args.length < n) {
            return _.ncurry.apply(this, [n, fn].concat(args));
        }
        return fn.apply(this, args.slice(0, n));
    };
};

/**
 * 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} f - 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. 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;
        }
        var r = fns[0].apply(this, arguments);
        for (var i = 1; i < fns.length; i++) {
            r = fns[i].call(this, r);
        }
        return r;
    };
};

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

function Stream(/*optional*/xs, /*optional*/ee, /*optional*/mappingHint) {
    if (xs && _.isStream(xs)) {
        // already a Stream
        return xs;
    }

    EventEmitter.call(this);
    var self = 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
    self.__HighlandStream__ = true;

    self.id = ('' + Math.random()).substr(2, 6);
    this.paused = true;
    this._incoming = [];
    this._outgoing = [];
    this._consumers = [];
    this._observers = [];
    this._destructors = [];
    this._send_events = false;
    this.source = null;

    // Old-style node Stream.pipe() checks for this
    this.writable = true;

    self.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
    self.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;
            }
        }
    });

    if (xs === undefined) {
        // nothing else to do
    }
    else if (Array.isArray(xs)) {
        self._incoming = xs.concat([nil]);
    }
    else if (typeof xs === 'function') {
        this._generator = xs;
        this._generator_push = function (err, x) {
            self.write(err ? new StreamError(err): x);
        };
        this._generator_next = function (s) {
            if (s) {
                // we MUST pause to get the redirect object into the _incoming
                // buffer otherwise it would be passed directly to _send(),
                // which does not handle StreamRedirect objects!
                var _paused = self.paused;
                if (!_paused) {
                    self.pause();
                }
                self.write(new StreamRedirect(s));
                if (!_paused) {
                    self.resume();
                }
            }
            else {
                self._generator_running = false;
            }
            if (!self.paused) {
                self.resume();
            }
        };
    }
    else if (_.isObject(xs)) {
        if (_.isFunction(xs.then)) {
            // probably a promise
            return _(function (push) {
                xs.then(function (value) {
                    push(null, value);
                    return push(null, nil);
                },
                function (err) {
                    push(err);
                    return push(null, nil);
                });
            });
        }
        else {
            // assume it's a pipeable stream as a source
            this._generator = function () {
                delete self._generator;
                xs.pipe(self);
            };
        }
    }
    else if (typeof xs === 'string') {
        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; };
        }

        ee.on(xs, function () {
            var ctx = mapper.apply(this, arguments);
            self.write(ctx);
        });
    }
    else {
        throw new Error(
            'Unexpected argument type to Stream(): ' + (typeof xs)
        );
    }
}
inherits(Stream, EventEmitter);

/**
 * adds a top-level _.foo(mystream) style export for Stream methods
 */

function exposeMethod(name) {
    var f = Stream.prototype[name];
    var n = f.length;
    _[name] = _.ncurry(n + 1, function () {
        var args = Array.prototype.slice.call(arguments);
        var s = _(args.pop());
        return f.apply(s, args);
    });
}

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

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

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

function StreamRedirect(to) {
    this.__HighlandStreamRedirect__ = true;
    this.to = to;
}

/**
 * Returns true if `x` is a Highland Stream.
 *
 * @id isStream
 * @section Streams
 * @name _.isStream(x)
 * @param x - the object to test
 * @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 (err, x) {
    if (x === nil) {
        this.ended = true;
    }
    if (this._consumers.length) {
        for (var i = 0, len = this._consumers.length; i < len; i++) {
            var c = this._consumers[i];
            if (err) {
                c.write(new StreamError(err));
            }
            else {
                c.write(x);
            }
        }
    }
    if (this._observers.length) {
        for (var j = 0, len2 = this._observers.length; j < len2; j++) {
            this._observers[j].write(x);
        }
    }
    if (this._send_events) {
        if (x === nil) {
            this.emit('end');
        }
        else {
            this.emit('data', x);
        }
    }
};

/**
 * Pauses the stream. All Highland Streams start in the paused state.
 *
 * @id pause
 * @section Streams
 * @name Stream.pause()
 * @api public
 *
 * var xs = _(generator);
 * xs.pause();
 */

Stream.prototype.pause = function () {
    //console.log([this.id, 'pause']);
    this.paused = true;
    if (this.source) {
        this.source._checkBackPressure();
    }
};

/**
 * When there is a change in downstream consumers, it will often ask
 * the parent Stream to re-check it's state and pause/resume accordingly.
 */

Stream.prototype._checkBackPressure = function () {
    if (!this._consumers.length) {
        return this.pause();
    }
    for (var i = 0, len = this._consumers.length; i < len; i++) {
        if (this._consumers[i].paused) {
            return this.pause();
        }
    }
    return this.resume();
};

/**
 * Starts pull values out of the incoming buffer and sending them downstream,
 * this will exit early if this causes a downstream consumer to pause.
 */

Stream.prototype._readFromBuffer = function () {
    var len = this._incoming.length;
    var i = 0;
    while (i < len && !this.paused) {
        var x = this._incoming[i];
        if (_._isStreamError(x)) {
            this._send(x.error);
        }
        else if (_._isStreamRedirect(x)) {
            this._redirect(x.to);
        }
        else {
            this._send(null, x);
        }
        i++;
    }
    // remove processed data from _incoming buffer
    this._incoming.splice(0, i);
};

/**
 * Starts pull values out of the incoming buffer and sending them downstream,
 * this will exit early if this causes a downstream consumer to pause.
 */

Stream.prototype._sendOutgoing = function () {
    var len = this._outgoing.length;
    var i = 0;
    while (i < len && !this.paused) {
        var x = this._outgoing[i];
        if (_._isStreamError(x)) {
            Stream.prototype._send.call(this, x.error);
        }
        else if (_._isStreamRedirect(x)) {
            this._redirect(x.to);
        }
        else {
            Stream.prototype._send.call(this, null, x);
        }
        i++;
    }
    // remove processed data from _outgoing buffer
    this._outgoing.splice(0, i);
};

/**
 * Resumes a paused Stream. This will either read from the Stream's incoming
 * buffer or request more data from an upstream source.
 *
 * @id resume
 * @section Streams
 * @name Stream.resume()
 * @api public
 *
 * var xs = _(generator);
 * xs.resume();
 */

Stream.prototype.resume = function () {
    //console.log([this.id, 'resume']);
    if (this._resume_running) {
        // already processing _incoming buffer, ignore resume call
        this._repeat_resume = true;
        return;
    }
    this._resume_running = true;
    do {
        // use a repeat flag to avoid recursing resume() calls
        this._repeat_resume = false;
        this.paused = false;

        // send values from outgoing buffer first
        this._sendOutgoing();

        // send values from incoming buffer before reading from source
        this._readFromBuffer();

        // we may have paused while reading from buffer
        if (!this.paused) {
            // ask parent for more data
            if (this.source) {
                this.source._checkBackPressure();
            }
            // run _generator to fill up _incoming buffer
            else if (this._generator) {
                this._runGenerator();
            }
            else {
                // perhaps a node stream is being piped in
                this.emit('drain');
            }
        }
    } while (this._repeat_resume);
    this._resume_running = false;
};

/**
 * 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.
 *
 * @id end
 * @section Streams
 * @name Stream.end()
 * @api public
 *
 * mystream.end();
 */

Stream.prototype.end = function () {
    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.
 *
 * This function returns the destination so you can chain together pipe calls.
 *
 * @id pipe
 * @section Streams
 * @name Stream.pipe(dest)
 * @param {Writable Stream} dest - the destination to write all data to
 * @api public
 *
 * var source = _(generator);
 * var dest = fs.createWriteStream('myfile.txt')
 * source.pipe(dest);
 *
 * // chained call
 * source.pipe(through).pipe(dest);
 */

Stream.prototype.pipe = function (dest) {
    var self = this;

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

    var s = self.consume(function (err, x, push, next) {
        if (err) {
            self.emit('error', err);
            return;
        }
        if (x === nil) {
            if (canClose) {
                dest.end();
            }
        }
        else if (dest.write(x) !== false) {
            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.
    this._destructors.push(function () {
        dest.removeListener('drain', onConsumerDrain);
    });

    s.resume();
    return dest;

    function onConsumerDrain() {
        s.resume();
    }
};

/**
 * 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 Streams
 * @name Stream.destroy()
 * @api public
 */

Stream.prototype.destroy = function () {
    var self = this;
    this.end();
    _(this._consumers).each(function (consumer) {
        self._removeConsumer(consumer);
    });
    if (this.source) {
        this.source._removeConsumer(this);
    }
    _(this._destructors).each(function (destructor) {
        destructor();
    });
};

/**
 * 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 () {
    // if _generator already running, exit
    if (this._generator_running) {
        return;
    }
    this._generator_running = true;
    this._generator(this._generator_push, this._generator_next);
};

/**
 * Performs the redirect from one Stream to another. In order for the
 * redirect to happen at the appropriate time, it is put on the incoming
 * buffer as a StreamRedirect object, and this function is called
 * once it is read from the buffer.
 */

Stream.prototype._redirect = function (to) {
    // coerce to Stream
    to = _(to);

    to._consumers = this._consumers.map(function (c) {
        c.source = to;
        return c;
    });
    // TODO: copy _observers
    this._consumers = [];
    this.consume = function () {
        return to.consume.apply(to, arguments);
    };
    this._removeConsumer = function () {
        return to._removeConsumer.apply(to, arguments);
    };
    if (this.paused) {
        to.pause();
    }
    else {
        this.pause();
        to._checkBackPressure();
    }
};

/**
 * Adds a new consumer Stream, which will accept data and provide backpressure
 * to this Stream. Adding more than one consumer will cause an exception to be
 * thrown as the backpressure strategy must be explicitly chosen by the
 * developer (through calling fork or observe).
 */

Stream.prototype._addConsumer = function (s) {
    if (this._consumers.length) {
        throw new Error(
            'Stream already being consumed, you must either fork() or observe()'
        );
    }
    s.source = this;
    this._consumers.push(s);
    this._checkBackPressure();
};

/**
 * Removes a consumer from this Stream.
 */

Stream.prototype._removeConsumer = function (s) {
    this._consumers = this._consumers.filter(function (c) {
        return c !== s;
    });
    if (s.source === this) {
        s.source = null;
    }
    this._checkBackPressure();
};

/**
 * 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 Streams
 * @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();
 *         }
 *     });
 * };
 */

Stream.prototype.consume = function (f) {
    var self = this;
    var s = new Stream();
    var _send = s._send;
    var push = function (err, x) {
        //console.log(['push', err, x, s.paused]);
        if (x === nil) {
            // ended, remove consumer from source
            self._removeConsumer(s);
        }
        if (s.paused) {
            if (err) {
                s._outgoing.push(new StreamError(err));
            }
            else {
                s._outgoing.push(x);
            }
        }
        else {
            _send.call(s, err, x);
        }
    };
    var async;
    var next_called;
    var next = function (s2) {
        //console.log(['next', async]);
        if (s2) {
            // we MUST pause to get the redirect object into the _incoming
            // buffer otherwise it would be passed directly to _send(),
            // which does not handle StreamRedirect objects!
            var _paused = s.paused;
            if (!_paused) {
                s.pause();
            }
            s.write(new StreamRedirect(s2));
            if (!_paused) {
                s.resume();
            }
        }
        else if (async) {
            s.resume();
        }
        else {
            next_called = true;
        }
    };
    s._send = function (err, x) {
        async = false;
        next_called = false;
        f(err, x, push, next);
        async = true;
        if (!next_called) {
            s.pause();
        }
    };
    self._addConsumer(s);
    return s;
};

/**
 * 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 Streams
 * @name Stream.pull(f)
 * @param {Function} f - the function to handle data
 * @api public
 *
 * xs.pull(function (err, x) {
 *     // do something
 * });
 */

Stream.prototype.pull = function (f) {
    var s = this.consume(function (err, x) {
        s.source._removeConsumer(s);
        f(err, x);
    });
    s.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).
 *
 * @id write
 * @section Streams
 * @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]
 * });
 */

Stream.prototype.write = function (x) {
    if (this.paused) {
        this._incoming.push(x);
    }
    else {
        if (_._isStreamError(x)) {
            this._send(x.error);
        }
        else {
            this._send(null, 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 it's source as fast as the slowest consumer can handle them.
 *
 * @id fork
 * @section 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();
 */

Stream.prototype.fork = function () {
    var s = new Stream();
    s.id = 'fork:' + s.id;
    s.source = this;
    this._consumers.push(s);
    this._checkBackPressure();
    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 it's
 * buffer and cause memory issues. Where possible, you should use [fork](#fork).
 *
 * @id observe
 * @section 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 recieve data as fast as ys can handle it
 * ys.resume();
 */

Stream.prototype.observe = function () {
    var s = new Stream();
    s.id = 'observe:' + s.id;
    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 Streams
 * @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);
 *     }
 * });
 */

Stream.prototype.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();
        }
    });
};
exposeMethod('errors');

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

Stream.prototype.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();
        }
    });
};
exposeMethod('stopOnError');

/**
 * Iterates over every value from the Stream, calling the iterator function
 * on each of them. This function causes a **thunk**.
 *
 * If an error from the Stream reaches the `each` call, it will emit an
 * error event (which will cause it to throw if unhandled).
 *
 * @id each
 * @section Streams
 * @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
 * });
 */

Stream.prototype.each = function (f) {
    var self = this;
    return this.consume(function (err, x, push, next) {
        if (err) {
            self.emit('error', err);
        }
        else if (x !== nil) {
            f(x);
            next();
        }
    }).resume();
};
exposeMethod('each');

/**
 * Applies results from a Stream as arguments to a function
 *
 * @id apply
 * @section Streams
 * @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
 * });
 */

Stream.prototype.apply = function (f) {
    return this.toArray(function (args) {
        f.apply(null, args);
    });
};
exposeMethod('apply');

/**
 * Collects all values from a Stream into an Array and calls a function with
 * once with the result. This function causes a **thunk**.
 *
 * If an error from the Stream reaches the `toArray` call, it will emit an
 * error event (which will cause it to throw if unhandled).
 *
 * @id toArray
 * @section Streams
 * @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]
 * });
 */

// TODO: implement using collect()?
Stream.prototype.toArray = function (f) {
    var self = this;
    var xs = [];
    return this.consume(function (err, x, push, next) {
        if (err) {
            self.emit('error', err);
        }
        else if (x === nil) {
            f(xs);
        }
        else {
            xs.push(x);
            next();
        }
    }).resume();
};

/**
 * Creates a new Stream of transformed values by applying a function to each
 * value from the source. The transformation function can be replaced with
 * a non-function value for convenience, and it will emit that value
 * for every data event on the source Stream.
 *
 * @id map
 * @section Streams
 * @name Stream.map(f)
 * @param f - the transformation function or value to map to
 * @api public
 *
 * var doubled = _([1, 2, 3, 4]).map(function (x) {
 *     return x * 2;
 * });
 *
 * _([1, 2, 3]).map('hi')  // => 'hi', 'hi', 'hi'
 */

Stream.prototype.map = function (f) {
    if (!_.isFunction(f)) {
        var val = f;
        f = function () {
            return val;
        };
    }
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            push(err, x);
        }
        else {
            push(null, f(x));
            next();
        }
    });
};
exposeMethod('map');

/**
 * Creates a new Stream of values by applying each item in a Stream to an
 * iterator function which may return a Stream. Each item on these result
 * Streams are then emitted on a single output Stream.
 *
 * The same as calling `stream.map(f).flatten()`.
 *
 * @id flatMap
 * @section Streams
 * @name Stream.flatMap(f)
 * @param {Function} f - the iterator function
 * @api public
 *
 * filenames.flatMap(readFile)
 */

Stream.prototype.flatMap = function (f) {
    return this.map(f).flatten();
};
exposeMethod('flatMap');

/**
 * Retrieves values associated with a given property from all elements in
 * the collection.
 *
 * @id pluck
 * @section Streams
 * @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']
 * });
 */

Stream.prototype.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();
        }
    });
};
exposeMethod('pluck');

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

Stream.prototype.filter = function (f) {
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            push(err, x);
        }
        else {
            if (f(x)) {
                push(null, x);
            }
            next();
        }
    });
};
exposeMethod('filter');

/**
 * 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 it's first value (all other values on the Stream will be
 * disregarded).
 *
 * @id flatFilter
 * @section Streams
 * @name Stream.flatFilter(f)
 * @param {Function} f - the truth test function which returns a Stream
 * @api public
 *
 * var checkExists = _.wrapCallback(fs.exists);
 * filenames.flatFilter(checkExists)
 */

Stream.prototype.flatFilter = function (f) {
    var xs = this.observe();
    var ys = this.flatMap(function (x) {
        return f(x).take(1);
    });
    return xs.zip(ys)
        .filter(function (pair) {
            return pair[1];
        })
        .map(function (pair) {
            return pair[0];
        });
};
exposeMethod('flatFilter');

/**
 * The inverse of [filter](#filter).
 *
 * @id reject
 * @section Streams
 * @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;
 * });
 */

Stream.prototype.reject = function (f) {
    return this.filter(_.compose(_.not, f));
};
exposeMethod('reject');

/**
 * A convenient form of filter, which returns the first object from a
 * Stream that passes the provided truth test
 *
 * @id find
 * @section Streams
 * @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'}]
 */

Stream.prototype.find = function (f) {
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            next();
        }
        else if (x === nil) {
            push(err, x);
        }
        else {
            if (f(x)) {
                push(null, x);
                push(null, nil);
            }
            next();
        }
    }.bind(this));
};
exposeMethod('find');

/**
 * A convenient form of reduce, which groups items based on a function or property name
 *
 * @id group
 * @section Streams
 * @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'}]
 * // =>  }
 *
 */

Stream.prototype.group = function (f) {
    var lambda = _.isString(f) ? _.get(f) : f;
    return this.reduce({}, function (m, o) {
        var key = lambda(o);
        if (!m.hasOwnProperty(key)) { m[key] = []; }
        m[key].push(o);
        return m;
    }.bind(this));
};
exposeMethod('group');

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

Stream.prototype.compact = function () {
    return this.filter(function (x) {
        return x;
    });
};
exposeMethod('compact');

/**
 * A convenient form of filter, which returns all objects from a Stream
 * match a set of property values.
 *
 * @id where
 * @section Streams
 * @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'}
 */

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

/**
 * Takes two Streams and returns a Stream of corresponding pairs.
 *
 * @id zip
 * @section 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]
 */

Stream.prototype.zip = function (ys) {
    ys = _(ys);
    var xs = this;
    var returned = 0;
    var z = [];
    function nextValue(index, max, src, push, next) {
        src.pull(function (err, x) {
            if (err) {
                push(null, err);
                nextValue(index, max, src, push, next);
            }
            else if (x === _.nil) {
                push(null, nil);
            }
            else {
                returned++;
                z[index] = x;
                if (returned === max) {
                    push(null, z);
                    next();
                }
            }
        });
    }
    return _(function (push, next) {
        returned = 0;
        z = [];
        nextValue(0, 2, xs, push, next);
        nextValue(1, 2, ys, push, next);
    });
};
exposeMethod('zip');

/**
 * Creates a new Stream with the first `n` values from the source.
 *
 * @id take
 * @section Streams
 * @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
 */

// TODO: test that errors don't count in take 'n' calls
Stream.prototype.take = function (n) {
    if (n === 0) {
        return _([]);
    }
    return this.consume(function (err, x, push, next) {
        if (err) {
            push(err);
            if (n > 0) {
                next();
            }
            else {
                push(null, nil);
            }
        }
        else if (x === nil) {
            push(null, nil);
        }
        else {
            n--;
            push(null, x);
            if (n > 0) {
                next();
            }
            else {
                push(null, nil);
            }
        }
    });
};
exposeMethod('take');

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

Stream.prototype.head = function () {
    return this.take(1);
};
exposeMethod('head');

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

Stream.prototype.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();
        }
    });
};
exposeMethod('last');

/**
 * Passes the current Stream to a function, returning the result. Can also
 * be used to pipe the current Stream through another Stream. It will always
 * return a Highland Stream (instead of the piped to target directly as in
 * Node.js).
 *
 * @id through
 * @section Streams
 * @name Stream.through(target)
 * @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) {
 *     // ...
 * });
 */

Stream.prototype.through = function (target) {
    if (_.isFunction (target)) {
        return target(this);
    }
    else {
        var output = _();
        target.pause();
        this.pipe(target).pipe(output);
        return output;
    }
};
exposeMethod('through');

/**
 * 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 Streams
 * @name _.pipeline(...)
 * @api public
 *
 * var through = _.pipeline(
 *     _.map(parseJSON),
 *     _.filter(isBlogpost),
 *     _.reduce(collectCategories)
 *     _.through(otherPipeline)
 * );
 *
 * readStream.pipe(through).pipe(outStream);
 */

_.pipeline = function (/*through...*/) {
    var start = _(arguments[0]);
    var rest = slice.call(arguments, 1);
    var end = rest.reduce(function (src, dest) {
        return src.through(dest);
    }, start);
    var wrapper = _(function (push, next) {
        end.pull(function (err, x) {
            if (err) {
                wrapper._send(err);
                next();
            }
            else if (x === nil) {
                wrapper._send(null, nil);
            }
            else {
                wrapper._send(null, x);
                next();
            }
        });
    });
    wrapper.write = function (x) {
        start.write(x);
    };
{'fraction_non_alphanumeric': 0.13698672911787665, 'fraction_numerical': 0.004829560239396305, 'mean_word_length': 2.1186418671081246, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5671829', 'n_tokens_mistral': 29399, 'n_tokens_neox': 28669, 'n_words': 9696}