<commit_before>/**
    ## `Stream(state)`

    * concat(b) - TODO
    * empty() - TODO
    * foreach(f) - TODO
    * filter(f) - TODO
    * map(f) - TODO
    * zip(s) - TODO
**/
function Stream(f) {
    var self = getInstance(this, Stream);
    // Not sure I like this subs array hanging off the stream.
    // Is there any alternatives?
    self.subs = [];

    f(function(a) {
        self.subs.forEach(function(s) {
            s(a);
        });
    });

    return self;
}

Stream.of = function(a, b) {
    var unbinder,
        bounce;

    return new Stream(function(state) {
        unbinder = a(function() {
            bounce = b();
            if (!bounce.isDone) {
                state(bounce.thunk());
            } else {
                state(bounce.result);
                unbinder();
            }
        });
    });
};

Stream.prototype.chain = function(f) {
    var env = this;
    return new Stream(function(state) {
        env.foreach(function(a) {
            f(a).fold(
                function(a) {
                    state(a);
                },
                function(){
                    // Do nothing.
                }
            );
        });
    });
};

Stream.prototype.concat = function(b) {
    return this.chain(function(a) {
        return Option.some(a.concat(b));
    });
};

Stream.prototype.empty = function() {
    return this.chain(function(a) {
        return Option.some(bilby.empty(a));
    });
};

Stream.prototype.foreach = function(f) {
    var env = this;
    return new Stream(function(state) {
        // I only subs for foreach, can I lift this out?
        env.subs.push(function(a) {
            f(a);
            state(a);
        });
    });
};

Stream.prototype.filter = function(f) {
    return this.chain(function(a) {
        return f(a) ? Option.some(a) : Option.none;
    });
};

Stream.prototype.map = function(f) {
    return this.chain(function(a) {
        return Option.some(f(a));
    });
};

Stream.prototype.zip = function(s) {
    var env = this;
    var cur,
        left = List.nil,
        right = List.nil;

    function dispatch() {
        if (and(left.isNonEmpty)(right.isNonEmpty)) {

            cur(left.car)(right.car);

            // This doesn't seem very functional.
            // I'm sure we can use State.js
            left = left.cdr;
            right = right.cdr;
        }
    }

    this.foreach(function(a) {
        left = List.cons.of(a, left);
        dispatch();
    });
    s.foreach(function(a) {
        right = List.cons.of(a, right);
        dispatch();
    });

    return new Stream(function(state) {
        cur = curry(function(a, b) {
            state([a, b]);
        });
    });
};

Stream.prototype.toArray = function() {
    var accum = [];
    this.foreach(function(a) {
        accum.push(a);
    });
    return accum;
};

/**

  ## promise

      Stream.promise(promise).foreach(function (a) {
        console.log(a);
      });
**/
Stream.promise = function(p) {
    return new Stream(function(state) {
        setTimeout(function() {
            p.fork(
                function(data) {
                    state(Attempt.success(data));
                },
                function(error) {<commit_msg>Slight refactor of the stream.js zip method.<commit_after>/**
    ## `Stream(state)`

    * concat(b) - TODO
    * empty() - TODO
    * foreach(f) - TODO
    * filter(f) - TODO
    * map(f) - TODO
    * zip(s) - TODO
**/
function Stream(f) {
    var self = getInstance(this, Stream);
    // Not sure I like this subs array hanging off the stream.
    // Is there any alternatives?
    self.subs = [];

    f(function(a) {
        self.subs.forEach(function(s) {
            s(a);
        });
    });

    return self;
}

Stream.of = function(a, b) {
    var unbinder,
        bounce;

    return new Stream(function(state) {
        unbinder = a(function() {
            bounce = b();
            if (!bounce.isDone) {
                state(bounce.thunk());
            } else {
                state(bounce.result);
                unbinder();
            }
        });
    });
};

Stream.prototype.chain = function(f) {
    var env = this;
    return new Stream(function(state) {
        env.foreach(function(a) {
            f(a).fold(
                function(a) {
                    state(a);
                },
                function(){
                    // Do nothing.
                }
            );
        });
    });
};

Stream.prototype.concat = function(b) {
    return this.chain(function(a) {
        return Option.some(a.concat(b));
    });
};

Stream.prototype.empty = function() {
    return this.chain(function(a) {
        return Option.some(bilby.empty(a));
    });
};

Stream.prototype.foreach = function(f) {
    var env = this;
    return new Stream(function(state) {
        // I only subs for foreach, can I lift this out?
        env.subs.push(function(a) {
            f(a);
            state(a);
        });
    });
};

Stream.prototype.filter = function(f) {
    return this.chain(function(a) {
        return f(a) ? Option.some(a) : Option.none;
    });
};

Stream.prototype.map = function(f) {
    return this.chain(function(a) {
        return Option.some(f(a));
    });
};

Stream.prototype.zip = function(s) {
    var env = this;
    var dispatch,
        left = List.nil,
        right = List.nil;

    var stream = new Stream(function(state) {
        dispatch = function() {
            if (and(left.isNonEmpty)(right.isNonEmpty)) {

                state([left.car, right.car]);

                // This doesn't seem very functional.
                // I'm sure we can use State.js
                left = left.cdr;
                right = right.cdr;
            }
        };
    });

    this.foreach(function(a) {
        left = List.cons.of(a, left);
        dispatch();
    });
    s.foreach(function(a) {
        right = List.cons.of(a, right);
        dispatch();
    });

    return stream;
};

Stream.prototype.toArray = function() {
    var accum = [];
    this.foreach(function(a) {
        accum.push(a);
    });
    return accum;
};

/**

  ## promise

      Stream.promise(promise).foreach(function (a) {
        console.log(a);
      });
**/
Stream.promise = function(p) {
    return new Stream(function(state) {
        setTimeout(function() {
            p.fork(
                function(data) {
                    state(Attempt.success(data));
                },
                function(error) {
{'fraction_non_alphanumeric': 0.1501233045622688, 'fraction_numerical': 0.0, 'mean_word_length': 1.5739785799285997, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '1945071', 'n_tokens_mistral': 2037, 'n_tokens_neox': 1980, 'n_words': 391}