<commit_before>'use strict';
var stream = require('stream');
var Readable = stream.Readable;
var PassThrough = stream.PassThrough;

/**
 * Given an ordered list of streams and Buffers, streamCat will pipe each stream or buffer to
 * the writable stream in order.
 */
function streamCat(streams, writeable, endCallback) {

    var concatenate = streams.reduceRight(function(next, stream) {
        return function() { handleStream(stream, writeable, next); };
    }, endCallback);

	process.nextTick(concatenate);
}

function handleStream(inputStream, outputStream, next) {
	if (inputStream instanceof Buffer) {
		outputStream.write(inputStream);
		process.nextTick(next);
	} else if (inputStream instanceof Promise || (inputStream.then && inputStream.catch)) {
		inputStream.then(function(content) {
			return handleStream(content, outputStream, next);
		}).catch(handleError);
	} else if (inputStream instanceof Readable || (inputStream.on && inputStream.read)) {
		inputStream.on('data', function(chunk, enc) {
			outputStream.write(chunk, enc);
		});

		inputStream.on('end', function() {
			next();
		});

		inputStream.on('error', handleError);
	} else {
		process.nextTick(function() {
			handleError(new Error("Invalid stream component '" + (typeof inputStream) + "', must be: stream.Readable, Buffer, or Promise"));
		});
	}

	function handleError(error) {
		process.nextTick(function() {
			outputStream.emit('error', error);
			outputStream.end();
		});
	}
}


module.exports = function(streams) {
	var passThrough = new PassThrough();
	passThrough.__streamCatBufferedError = null;

	var originalPassthroughOn = passThrough.on.bind(passThrough);

	// Complex: Node's EventEmitter will fire and forget events - if there is
	// nothing listening then the event is forgotten.  In the case of 'error'
	// events, if there are no listeners then an exception is thrown inside
	// the emitter.  To avoid this and to give some time to attach error
	// handlers to the final composed stream, if an error occurs _before_ an
	// error handler has been attached it gets buffered.  Then, when the
	// appropriate event handler is attached, the event is fired and the
	// appropriate handler can pick the error up.
	originalPassthroughOn('error', function(error) {
		this.__streamCatBufferedError = error;
	}.bind(passThrough));

	passThrough.on = function(ev, handler) {
		originalPassthroughOn(ev, handler);

		if (ev === 'error') {
			if (this.__streamCatBufferedError !== null) {
				this.emit(ev, this.__streamCatBufferedError);
			}

			passThrough.on = originalPassthroughOn;
		}
	}.bind(passThrough);

	streamCat(streams, passThrough, function() { passThrough.end(); });
	return passThrough;
};
<commit_msg>Cleaner solution to buffering errors<commit_after>'use strict';
var stream = require('stream');
var Readable = stream.Readable;
var PassThrough = stream.PassThrough;

/**
 * Given an ordered list of streams and Buffers, streamCat will pipe each stream or buffer to
 * the writable stream in order.
 */
function streamCat(streams, writeable, endCallback) {

    var concatenate = streams.reduceRight(function(next, stream) {
        return function() { handleStream(stream, writeable, next); };
    }, endCallback);

	process.nextTick(concatenate);
}

function handleStream(inputStream, outputStream, next) {
	if (inputStream instanceof Buffer) {
		outputStream.write(inputStream);
		process.nextTick(next);
	} else if (inputStream instanceof Promise || (inputStream.then && inputStream.catch)) {
		inputStream.then(function(content) {
			return handleStream(content, outputStream, next);
		}).catch(handleError);
	} else if (inputStream instanceof Readable || (inputStream.on && inputStream.read)) {
		inputStream.on('data', function(chunk, enc) {
			outputStream.write(chunk, enc);
		});

		inputStream.on('end', function() {
			next();
		});

		inputStream.on('error', handleError);
	} else {
		process.nextTick(function() {
			handleError(new Error("Invalid stream component '" + (typeof inputStream) + "', must be: stream.Readable, Buffer, or Promise"));
		});
	}

	function handleError(error) {
		process.nextTick(function() {
			outputStream.emit('error', error);
			outputStream.end();
		});
	}
}


module.exports = function(streams) {
	var passThrough = new PassThrough();
	var streamCatBufferedError = null;

	var originalPassthroughOn = passThrough.on.bind(passThrough);

	// Complex: Node's EventEmitter will fire and forget events - if there is
	// nothing listening then the event is forgotten.  In the case of 'error'
	// events, if there are no listeners then an exception is thrown inside
	// the emitter.  To avoid this and to give some time to attach error
	// handlers to the final composed stream, if an error occurs _before_ an
	// error handler has been attached it gets buffered.  Then, when the
	// appropriate event handler is attached, the event is fired and the
	// appropriate handler can pick the error up.
	//
	function handleError(error) {
		streamCatBufferedError = error;
	}
	originalPassthroughOn('error', handleError);

	passThrough.on = function(ev, handler) {
		originalPassthroughOn(ev, handler);

		if (ev === 'error') {
			if (streamCatBufferedError !== null) {
				this.removeListener('error', handleError);
				this.emit(ev, streamCatBufferedError);
			}

			passThrough.on = originalPassthroughOn;
		}
	}.bind(passThrough);

	streamCat(streams, passThrough, function() { passThrough.end(); });
	return passThrough;
};

{'fraction_non_alphanumeric': 0.11988304093567251, 'fraction_numerical': 0.0, 'mean_word_length': 4.500502512562814, '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': '689260', 'n_tokens_mistral': 1715, 'n_tokens_neox': 1569, 'n_words': 476}