gen = function () {
        // Important. next may not be called outside of
        // pullCb.
        source.pull(pullCb);
    }
Consumes values from a Stream (once resumed) and returns a new Stream for
you to optionally push values onto using the provided push / next functions.

This function forms the basis of many higher-level Stream operations.
It will not cause a paused stream to immediately resume, but behaves more
like a 'through' stream, handling values as they are read.

@id consume
@section Transforms
@name Stream.consume(f)
@param {Function} f - the function to handle errors and values
@api public

var filter = function (f, source) {
    return source.consume(function (err, x, push, next) {
        if (err) {
            // pass errors along the stream and consume next value
            push(err);
            next();
        }
        else if (x === _.nil) {
            // pass nil (end event) along the stream
            push(null, x);
        }
        else {
            // pass on the value only if the value passes the predicate
            if (f(x)) {
                push(null, x);
            }
            next();
        }
    });
};