<commit_before>define([
    './stream',
    './noop',
    '../functional/each',
    '../functional/contains',
    '../functional/clone'
], function(Stream, noop, each, contains, clone, undefined) {
    'use strict';

    /**
     * @param {Stream[]} streams
     * @param {Array} [buffer]
     * @return {Stream}
     */
    return function when(streams, buffer) {
        buffer = buffer || new Array(streams.length);

        if (contains(streams, Stream.stop)) {
            return noop();
        }

        return new Stream(function(sinkValue, sinkError) {
            each(streams, function(stream, index) {
                stream.on(function(value, next) {
                    buffer[index] = value;
                    streams[index] = next;

                    if (!contains(buffer, undefined)) {
                        sinkValue(
                            clone(buffer),
                            Stream.streamable(next)
                                ? when(streams, buffer)
                                : Stream.stop
                        );
                    }

                    return Stream.stop;
                }, sinkError)
            });
        })
    }
});<commit_msg>When will collect last event<commit_after>define([
    './stream',
    './noop',
    '../functional/each',
    '../functional/contains',
    '../functional/clone'
], function(Stream, noop, each, contains, clone, undefined) {
    'use strict';

    /**
     * @param {Stream[]} streams
     * @param {Array} [buffer]
     * @return {Stream}
     */
    return function when(streams, buffer) {
        buffer = buffer || new Array(streams.length);

        if (contains(streams, Stream.stop)) {
            return noop();
        }

        return new Stream(function(sinkValue, sinkError) {
            each(streams, function(stream, index) {
                stream.on(function(value, next) {
                    buffer[index] = value;
                    streams[index] = next;

                    if (contains(buffer, undefined) && Stream.streamable(next)) {
                        when(streams, buffer)
                    } else {
                        sinkValue(
                            clone(buffer),
                            Stream.streamable(next)
                                ? when(streams, buffer)
                                : Stream.stop
                        );
                    }

                    return Stream.stop;
                }, sinkError)
            });
        })
    }
});
{'fraction_non_alphanumeric': 0.12440758293838862, 'fraction_numerical': 0.0, 'mean_word_length': 1.2006950477845353, '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': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6674696', 'n_tokens_mistral': 651, 'n_tokens_neox': 622, 'n_words': 128}