streams: Rework stream merging to use ordered-read-streams

package.json

@@ -32,7 +32,6 @@
     "@types/glob": "^7.1.1",
     "@types/glob-parent": "^5.1.0",
     "@types/is-ci": "^2.0.0",
-    "@types/merge2": "^1.1.4",
     "@types/micromatch": "^4.0.0",
     "@types/minimist": "^1.2.0",
     "@types/mocha": "^5.2.7",
@@ -60,7 +59,7 @@
     "@nodelib/fs.stat": "^2.0.2",
     "@nodelib/fs.walk": "^1.2.3",
     "glob-parent": "^5.1.2",
-    "merge2": "^1.3.0",
+    "ordered-read-streams": "^2.0.0",
     "micromatch": "^4.0.4"
   },
   "scripts": {

src/utils/stream.spec.ts

@@ -3,64 +3,124 @@ import * as stream from 'stream';
 
 import * as util from './stream';
 
+function concat<T extends unknown>(fn: (data: T[]) => void): stream.Writable {
+	const data: T[] = [];
+	return new stream.Writable({
+		objectMode: true,
+		write(chunk: T, _: string, callback: (error?: Error | null | undefined) => void) {
+			data.push(chunk);
+			callback();
+		},
+		final(callback: (error?: Error | null | undefined) => void): void {
+			fn(data);
+			callback();
+		}
+	});
+}
+
 describe('Utils → Stream', () => {
 	describe('.merge', () => {
-		it('should merge two streams into one stream', () => {
-			const first = new stream.PassThrough();
-			const second = new stream.PassThrough();
+		it('should merge two streams into one stream', (done) => {
+			const first = stream.Readable.from(['one']);
+			const second = stream.Readable.from(['two']);
 
-			const expected = 3;
+			const expected = ['one', 'two'];
 
 			const mergedStream = util.merge([first, second]);
 
-			const actual = mergedStream.listenerCount('close');
-
-			assert.strictEqual(actual, expected);
+			stream.pipeline([
+				mergedStream,
+				concat((actual: string[]) => {
+					assert.deepStrictEqual(actual, expected);
+				})
+			], done);
 		});
 
-		it('should propagate errors into merged stream', (done) => {
-			const first = new stream.PassThrough();
-			const second = new stream.PassThrough();
+		it('should propagate first error into merged stream (first)', (done) => {
+			const first = new stream.Readable({
+				read() {
+					this.destroy(new Error('1'));
+				}
+			});
+			const second = stream.Readable.from([]);
 
-			const expected = [1, 2, 3];
+			const expected = '1';
 
 			const mergedStream = util.merge([first, second]);
 
-			const actual: number[] = [];
+			stream.pipeline([
+				mergedStream,
+				concat(() => {
+					assert.fail('Should not reach concat callback');
+				})
+			], (error: Error | null | undefined) => {
+				assert.deepStrictEqual(error?.message, expected);
 
-			mergedStream.on('error', (error: number) => actual.push(error));
+				done();
+			});
+		});
+
+		it('should propagate first error into merged stream (second)', (done) => {
+			const first = stream.Readable.from([]);
+			const second = new stream.Readable({
+				read() {
+					this.destroy(new Error('2'));
+				}
+			});
 
-			mergedStream.once('finish', () => {
-				assert.deepStrictEqual(actual, expected);
+			const expected = '2';
+
+			const mergedStream = util.merge([first, second]);
+
+			stream.pipeline([
+				mergedStream,
+				concat(() => {
+					assert.fail('Should not reach concat callback');
+				})
+			], (error: Error | null | undefined) => {
+				assert.deepStrictEqual(error?.message, expected);
 
 				done();
 			});
-
-			first.emit('error', 1);
-			second.emit('error', 2);
-			mergedStream.emit('error', 3);
 		});
 
-		it('should propagate close event to source streams', (done) => {
-			const first = new stream.PassThrough();
-			const second = new stream.PassThrough();
+		it('should propagate destroy to source streams', (done) => {
+			const first = stream.Readable.from([]);
+			const second = stream.Readable.from([]);
 
 			const mergedStream = util.merge([first, second]);
 
 			const expected = [1, 2];
 
 			const actual: number[] = [];
 
-			first.once('close', () => actual.push(1));
-			second.once('close', () => actual.push(2));
+			let closeCount = 0;
 
-			mergedStream.once('finish', () => {
-				assert.deepStrictEqual(actual, expected);
+			first.once('close', () => {
+				closeCount++;
+				actual.push(1);
+				checkCloses();
+			});
+			second.once('close', () => {
+				closeCount++;
+				actual.push(2);
+				checkCloses();
+			});
 
-				done();
+			mergedStream.once('close', () => {
+				closeCount++;
+				checkCloses();
 			});
 
-			mergedStream.emit('close');
+			mergedStream.destroy();
+
+			function checkCloses(): void {
+				if (closeCount === 3) {
+					assert.deepStrictEqual(actual, expected);
+
+					done();
+				}
+			}
 		});
 	});
 });

src/utils/stream.ts

@@ -1,20 +1,8 @@
 import { Readable } from 'stream';
 
-import * as merge2 from 'merge2';
+// @ts-expect-error
+import * as ordered from 'ordered-read-streams';
 
-export function merge(streams: Readable[]): NodeJS.ReadableStream {
-	const mergedStream = merge2(streams);
-
-	streams.forEach((stream) => {
-		stream.once('error', (error) => mergedStream.emit('error', error));
-	});
-
-	mergedStream.once('close', () => propagateCloseEventToSources(streams));
-	mergedStream.once('end', () => propagateCloseEventToSources(streams));
-
-	return mergedStream;
-}
-
-function propagateCloseEventToSources(streams: Readable[]): void {
-	streams.forEach((stream) => stream.emit('close'));
+export function merge(streams: Readable[]): Readable {
+	return ordered(streams);
 }