Stream Copier

Collect a Stream into a Buffer, providing a fully readable stream to any consumers, even when they start reading late.

License, donations

GPL-3.0. If you want to support my work, you can:

Example

see example.js in the repo:

"use strict";

const {streamCopier} = require("stream-copier");

const { Readable, Writable } = require('stream');

function namedWritable(name) {
	return Writable({
		write: function(chunk, enc, next) {
			console.log(name, chunk.toString());
			next();
		}
	});
}

let c = 96;
const letters = new Readable();
letters._read = function () {
	if (c >= 'z'.charCodeAt(0)) {
		c = 96;
		return letters.push(null);
	}

	setTimeout(function () {
		letters.push(String.fromCharCode(++c));
	}, 10);
};

const splitter = streamCopier(letters);

splitter.getStream().pipe(namedWritable("first"));

setTimeout(() => {
	// second consumer starts late, still gets all the data
	splitter.getStream().pipe(namedWritable("late"));
}, 100);

splitter.on("end", () => {
	// and even when the stream has finished
	splitter.getStream().pipe(namedWritable("end"));
});

/* 
  You should see the following result:	
  block of first <letter>
  block of late <letter> as it starts streaming and receives all the missed chunks at once
  first <letter> and late <letter> interleaving as they are both getting the newer chunks
  block of end <letter> as end starts streaming when everything is done, receiving all the chunks at once 
*/

API

streamCopier(inputStream, [options], [callback])

• inputStream: Readable Stream
• options: Optional object with
  • maxLength: Maximum length of the internal buffer collection, in bytes
• callback: Callback function called when the inputStream has end-ed.

Returns an EventEmitter with the end and error events, getStream() function that returns a new copy of the split stream to pipe elsewhere, length() function to get the size in bytes (so far).

Error types

streamCopier.BufferFull

Thrown when the collected buffer size is about to get greater than maxLength (Infinity by default).

Changelog

v0.0.3 (🏳️‍🌈 June 21st, 2021)

• bugfix: actually keep track of length
• feature: expose length() function to query stored length in bytes

Changelog

v0.0.2 (🏳️‍🌈 June 1st, 2021)

add error event where BufferFull is emitted, or send error in callback

v0.0.1 (🏳️‍🌈 June 1st, 2021)

initial publication