remove useless test things

package.json

@@ -1,6 +1,6 @@
 {
-    "name": "@jogogo/mhysa",
-    "version": "2.0.0-alpha.4",
+    "name": "mhysa",
+    "version": "2.0.0-alpha.1",
     "description": "Streams and event emitter utils for Node.js",
     "keywords": [
         "promise",
@@ -27,11 +27,8 @@
     "files": [
         "dist"
     ],
-    "publishConfig": {
-        "registry": "https://npm.dev.jogogo.co/"
-    },
     "repository": {
-        "url": "git@github.com:Jogogoplay/mhysa.git",
+        "url": "git@github.com:Wenzil/Mhysa.git",
         "type": "git"
     },
     "scripts": {
@@ -43,7 +40,7 @@
     "dependencies": {},
     "devDependencies": {
         "@types/chai": "^4.1.7",
-        "@types/node": "^12.12.15",
+        "@types/node": "^12.7.2",
         "@types/sinon": "^7.0.13",
         "ava": "^2.4.0",
         "chai": "^4.2.0",
@@ -58,8 +55,7 @@
     },
     "ava": {
         "files": [
-            "tests/*.spec.ts",
-            "tests/utils/*.spec.ts"
+            "tests/*.spec.ts"
         ],
         "sources": [
             "src/**/*.ts"

src/functions/batch.ts

@@ -12,9 +12,7 @@ export function batch(
             clearTimeout(timer);
         }
         timer = null;
-        if (buffer.length > 0) {
-            self.push(buffer);
-        }
+        self.push(buffer);
         buffer = [];
     };
     return new Transform({

src/functions/compose.ts

@@ -1,5 +1,4 @@
-import { AllStreams, isReadable } from "../helpers";
-import { PassThrough, pipeline, TransformOptions, Transform } from "stream";
+import { pipeline, TransformOptions, Transform } from "stream";
 
 export function compose(
     streams: Array<
@@ -22,6 +21,18 @@ enum EventSubscription {
     Self,
 }
 
+type AllStreams =
+    | NodeJS.ReadableStream
+    | NodeJS.ReadWriteStream
+    | NodeJS.WritableStream;
+
+function isReadable(stream: AllStreams): stream is NodeJS.WritableStream {
+    return (
+        (stream as NodeJS.ReadableStream).pipe !== undefined &&
+        (stream as any).readable === true
+    );
+}
+
 export class Compose extends Transform {
     private first: AllStreams;
     private last: AllStreams;
@@ -34,11 +45,11 @@ export class Compose extends Transform {
         options?: TransformOptions,
     ) {
         super(options);
-        this.first = new PassThrough(options);
+        this.first = streams[0];
         this.last = streams[streams.length - 1];
         this.streams = streams;
         pipeline(
-            [this.first, ...streams],
+            streams,
             errorCallback ||
                 ((error: any) => {
                     if (error) {

src/functions/demux.ts

@@ -1,131 +1,99 @@
-import { DuplexOptions, Duplex, Transform } from "stream";
+import { WritableOptions, Writable } from "stream";
 
-import { isReadable } from "../helpers";
+enum EventSubscription {
+    Last = 0,
+    First,
+    All,
+    Self,
+    Unhandled,
+}
+
+const eventsTarget = {
+    close: EventSubscription.Self,
+    data: EventSubscription.All,
+    drain: EventSubscription.Self,
+    end: EventSubscription.Self,
+    error: EventSubscription.Self,
+    finish: EventSubscription.Self,
+    pause: EventSubscription.Self,
+    pipe: EventSubscription.Self,
+    readable: EventSubscription.Self,
+    resume: EventSubscription.Self,
+    unpipe: EventSubscription.Self,
+};
 
 type DemuxStreams = NodeJS.WritableStream | NodeJS.ReadWriteStream;
 
-export interface DemuxOptions extends DuplexOptions {
-    remultiplex?: boolean;
-}
-
 export function demux(
-    pipelineConstructor: (
-        destKey?: string,
-        chunk?: any,
-    ) => DemuxStreams | DemuxStreams[],
+    construct: (destKey?: string) => DemuxStreams,
     demuxBy: string | ((chunk: any) => string),
-    options?: DemuxOptions,
-): Duplex {
-    return new Demux(pipelineConstructor, demuxBy, options);
+    options?: WritableOptions,
+): Writable {
+    return new Demux(construct, demuxBy, options);
 }
 
-class Demux extends Duplex {
+// @TODO handle pipe event ie) Multiplex
+class Demux extends Writable {
     private streamsByKey: {
-        [key: string]: DemuxStreams[];
+        [key: string]: DemuxStreams;
     };
     private demuxer: (chunk: any) => string;
-    private pipelineConstructor: (
-        destKey?: string,
-        chunk?: any,
-    ) => DemuxStreams[];
-    private remultiplex: boolean;
-    private transform: Transform;
+    private construct: (destKey?: string) => DemuxStreams;
     constructor(
-        pipelineConstructor: (
-            destKey?: string,
-            chunk?: any,
-        ) => DemuxStreams | DemuxStreams[],
+        construct: (destKey?: string) => DemuxStreams,
         demuxBy: string | ((chunk: any) => string),
-        options: DemuxOptions = {},
+        options: WritableOptions = {},
     ) {
         super(options);
         this.demuxer =
             typeof demuxBy === "string" ? chunk => chunk[demuxBy] : demuxBy;
-        this.pipelineConstructor = (destKey: string, chunk?: any) => {
-            const pipeline = pipelineConstructor(destKey, chunk);
-            return Array.isArray(pipeline) ? pipeline : [pipeline];
-        };
-        this.remultiplex =
-            options.remultiplex === undefined ? true : options.remultiplex;
+        this.construct = construct;
         this.streamsByKey = {};
-        this.transform = new Transform({
-            ...options,
-            transform: (d, _, cb) => {
-                this.push(d);
-                cb(null);
-            },
-        });
-
-        this.on("unpipe", () => this._flush());
     }
 
-    // tslint:disable-next-line
-    public _read(size: number) {}
-
     public async _write(chunk: any, encoding: any, cb: any) {
         const destKey = this.demuxer(chunk);
         if (this.streamsByKey[destKey] === undefined) {
-            const newPipelines = this.pipelineConstructor(destKey, chunk);
-            this.streamsByKey[destKey] = newPipelines;
-
-            newPipelines.forEach(newPipeline => {
-                if (this.remultiplex && isReadable(newPipeline)) {
-                    (newPipeline as NodeJS.ReadWriteStream).pipe(
-                        this.transform,
-                    );
-                } else if (this.remultiplex) {
-                    console.error(
-                        `Pipeline construct for ${destKey} does not implement readable interface`,
-                    );
-                }
-            });
+            this.streamsByKey[destKey] = await this.construct(destKey);
         }
-        const pipelines = this.streamsByKey[destKey];
-        const pendingDrains: Array<Promise<any>> = [];
+        if (!this.streamsByKey[destKey].write(chunk, encoding)) {
+            this.streamsByKey[destKey].once("drain", () => {
+                cb();
+            });
+        } else {
+            cb();
+        }
+    }
 
-        pipelines.forEach(pipeline => {
-            if (!pipeline.write(chunk, encoding)) {
-                pendingDrains.push(
-                    new Promise(resolve => {
-                        pipeline.once("drain", () => {
-                            resolve();
-                        });
-                    }),
+    public on(event: string, cb: any) {
+        switch (eventsTarget[event]) {
+            case EventSubscription.Self:
+                super.on(event, cb);
+                break;
+            case EventSubscription.All:
+                Object.keys(this.streamsByKey).forEach(key =>
+                    this.streamsByKey[key].on(event, cb),
                 );
-            }
-        });
-        await Promise.all(pendingDrains);
-        cb();
+                break;
+            default:
+                super.on(event, cb);
+        }
+        return this;
     }
 
-    public _flush() {
-        const pipelines: DemuxStreams[] = [].concat.apply(
-            [],
-            Object.values(this.streamsByKey),
-        );
-        const flushPromises: Array<Promise<void>> = [];
-        pipelines.forEach(pipeline => {
-            flushPromises.push(
-                new Promise(resolve => {
-                    pipeline.once("end", () => {
-                        resolve();
-                    });
-                }),
-            );
-        });
-        pipelines.forEach(pipeline => pipeline.end());
-        Promise.all(flushPromises).then(() => {
-            this.push(null);
-            this.emit("end");
-        });
-    }
-
-    public _destroy(error: any, cb: (error?: any) => void) {
-        const pipelines: DemuxStreams[] = [].concat.apply(
-            [],
-            Object.values(this.streamsByKey),
-        );
-        pipelines.forEach(p => (p as any).destroy());
-        cb(error);
+    public once(event: string, cb: any) {
+        switch (eventsTarget[event]) {
+            case EventSubscription.Self:
+                super.once(event, cb);
+                break;
+            case EventSubscription.All:
+                Object.keys(this.streamsByKey).forEach(key =>
+                    this.streamsByKey[key].once(event, cb),
+                );
+                break;
+            default:
+                super.once(event, cb);
+        }
+        return this;
     }
 }

src/helpers.ts

@@ -1,17 +1,3 @@
 export async function sleep(time: number): Promise<{} | null> {
     return time > 0 ? new Promise(resolve => setTimeout(resolve, time)) : null;
 }
-
-export type AllStreams =
-    | NodeJS.ReadableStream
-    | NodeJS.ReadWriteStream
-    | NodeJS.WritableStream;
-
-export function isReadable(
-    stream: AllStreams,
-): stream is NodeJS.WritableStream {
-    return (
-        (stream as NodeJS.ReadableStream).pipe !== undefined &&
-        (stream as any).readable === true
-    );
-}

src/index.ts

@@ -1,6 +1,2 @@
 import mhysa from "./functions";
-import * as _utils from "./utils";
 export default mhysa;
-
-// @TODO fix this with proper import export
-export const utils = { ..._utils };

src/utils/collected.ts

@@ -1,12 +0,0 @@
-import { Transform } from "stream";
-
-export function collected(stream: Transform): any {
-    return new Promise((resolve, reject) => {
-        stream.once("data", d => {
-            resolve(d);
-        });
-        stream.once("error", e => {
-            reject(e);
-        });
-    });
-}

src/utils/index.ts

@@ -1 +0,0 @@
-export { collected } from "./collected";

tests/batch.spec.ts

@@ -2,7 +2,7 @@ import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
 import mhysa from "../src";
-const { batch, map, fromArray } = mhysa({ objectMode: true });
+const { batch } = mhysa({ objectMode: true });
 
 test.cb("batch() batches chunks together", t => {
     t.plan(3);
@@ -57,28 +57,3 @@ test.cb("batch() yields a batch after the timeout", t => {
         source.push(null);
     }, 600 * 2);
 });
-
-test.cb(
-    "batch() yields all input data even when the last element(s) dont make a full batch",
-    t => {
-        const data = [1, 2, 3, 4, 5, 6, 7];
-
-        fromArray([...data])
-            .pipe(batch(3))
-            .pipe(
-                map(d => {
-                    t.deepEqual(
-                        d,
-                        [data.shift(), data.shift(), data.shift()].filter(
-                            x => !!x,
-                        ),
-                    );
-                }),
-            )
-            .on("error", t.fail)
-            .on("finish", () => {
-                t.is(data.length, 0);
-                t.end();
-            });
-    },
-);

tests/compose.spec.ts

@@ -4,7 +4,7 @@ import { sleep } from "../src/helpers";
 import { Readable, Writable } from "stream";
 import mhysa from "../src";
 import { performance } from "perf_hooks";
-const { compose, map, fromArray } = mhysa({ objectMode: true });
+const { compose, map } = mhysa({ objectMode: true });
 
 test.cb("compose() chains two streams together in the correct order", t => {
     t.plan(3);
@@ -98,7 +98,7 @@ test.cb("piping compose() maintains correct order", t => {
 });
 
 test("compose() writable length should be less than highWaterMark when handing writes", async t => {
-    t.plan(2);
+    t.plan(7);
     return new Promise(async (resolve, reject) => {
         interface Chunk {
             key: string;
@@ -114,13 +114,9 @@ test("compose() writable length should be less than highWaterMark when handing w
             return chunk;
         });
 
-        const composed = compose(
-            [first, second],
-            undefined,
-            {
-                highWaterMark: 2,
-            },
-        );
+        const composed = compose([first, second], undefined, {
+            highWaterMark: 2,
+        });
         composed.on("error", err => {
             reject();
         });
@@ -144,12 +140,19 @@ test("compose() writable length should be less than highWaterMark when handing w
             { key: "e", mapped: [] },
         ];
 
-        fromArray(input).pipe(composed);
+        for (const item of input) {
+            const res = composed.write(item);
+            expect(composed._writableState.length).to.be.at.most(2);
+            t.pass();
+            if (!res) {
+                await sleep(10);
+            }
+        }
     });
 });
 
 test("compose() should emit drain event ~rate * highWaterMark ms for every write that causes backpressure", async t => {
-    t.plan(2);
+    t.plan(7);
     const _rate = 100;
     const highWaterMark = 2;
     return new Promise(async (resolve, reject) => {
@@ -168,13 +171,9 @@ test("compose() should emit drain event ~rate * highWaterMark ms for every write
             return chunk;
         });
 
-        const composed = compose(
-            [first, second],
-            undefined,
-            {
-                highWaterMark,
-            },
-        );
+        const composed = compose([first, second], undefined, {
+            highWaterMark,
+        });
         composed.on("error", err => {
             reject();
         });
@@ -182,14 +181,19 @@ test("compose() should emit drain event ~rate * highWaterMark ms for every write
         composed.on("drain", () => {
             t.pass();
             expect(composed._writableState.length).to.be.equal(0);
+            expect(performance.now() - start).to.be.closeTo(
+                _rate * highWaterMark,
+                40,
+            );
         });
 
         composed.on("data", (chunk: Chunk) => {
-            t.deepEqual(chunk.mapped, [1, 2]);
+            pendingReads--;
+            if (pendingReads === 0) {
+                resolve();
+            }
         });
 
-        composed.on("finish", () => resolve());
-
         const input = [
             { key: "a", mapped: [] },
             { key: "b", mapped: [] },
@@ -197,7 +201,19 @@ test("compose() should emit drain event ~rate * highWaterMark ms for every write
             { key: "d", mapped: [] },
             { key: "e", mapped: [] },
         ];
-        fromArray(input).pipe(composed);
+
+        let start = performance.now();
+        let pendingReads = input.length;
+        start = performance.now();
+        for (const item of input) {
+            const res = composed.write(item);
+            expect(composed._writableState.length).to.be.at.most(highWaterMark);
+            t.pass();
+            if (!res) {
+                await sleep(_rate * highWaterMark * 2);
+                start = performance.now();
+            }
+        }
     });
 });
 
@@ -221,13 +237,9 @@ test.cb(
             return chunk;
         });
 
-        const composed = compose(
-            [first, second],
-            undefined,
-            {
-                highWaterMark: 5,
-            },
-        );
+        const composed = compose([first, second], undefined, {
+            highWaterMark: 5,
+        });
 
         composed.on("error", err => {
             t.end(err);
@@ -235,6 +247,10 @@ test.cb(
 
         composed.on("drain", () => {
             expect(composed._writableState.length).to.be.equal(0);
+            expect(performance.now() - start).to.be.closeTo(
+                _rate * input.length,
+                50,
+            );
             t.pass();
         });
 
@@ -255,6 +271,7 @@ test.cb(
         input.forEach(item => {
             composed.write(item);
         });
+        const start = performance.now();
     },
 );
 
@@ -284,13 +301,9 @@ test.cb(
             { highWaterMark: 1 },
         );
 
-        const composed = compose(
-            [first, second],
-            undefined,
-            {
-                highWaterMark: 5,
-            },
-        );
+        const composed = compose([first, second], undefined, {
+            highWaterMark: 5,
+        });
         composed.on("error", err => {
             t.end(err);
         });
@@ -355,13 +368,9 @@ test.cb(
             { highWaterMark: 2 },
         );
 
-        const composed = compose(
-            [first, second],
-            undefined,
-            {
-                highWaterMark: 5,
-            },
-        );
+        const composed = compose([first, second], undefined, {
+            highWaterMark: 5,
+        });
         composed.on("error", err => {
             t.end(err);
         });
@@ -414,13 +423,9 @@ test.cb(
             return chunk;
         });
 
-        const composed = compose(
-            [first, second],
-            undefined,
-            {
-                highWaterMark: 6,
-            },
-        );
+        const composed = compose([first, second], undefined, {
+            highWaterMark: 6,
+        });
 
         composed.on("error", err => {
             t.end(err);
@@ -470,12 +475,9 @@ test.cb("compose() should be 'destroyable'", t => {
         return chunk;
     });
 
-    const composed = compose(
-        [first, second],
-        (err: any) => {
-            t.pass();
-        },
-    );
+    const composed = compose([first, second], (err: any) => {
+        t.pass();
+    });
 
     const fakeSource = new Readable({
         objectMode: true,
@@ -531,13 +533,9 @@ test.cb("compose() `finish` and `end` propagates", t => {
         return chunk;
     });
 
-    const composed = compose(
-        [first, second],
-        undefined,
-        {
-            highWaterMark: 3,
-        },
-    );
+    const composed = compose([first, second], undefined, {
+        highWaterMark: 3,
+    });
 
     const fakeSource = new Readable({
         objectMode: true,

tests/demux.spec.ts

@@ -1,11 +1,11 @@
 import test from "ava";
 import { expect } from "chai";
 import mhysa from "../src";
-import { Writable, Readable } from "stream";
+import { Writable } from "stream";
 const sinon = require("sinon");
 const { sleep } = require("../src/helpers");
 import { performance } from "perf_hooks";
-const { demux, map, fromArray } = mhysa({ objectMode: true });
+const { demux, map } = mhysa();
 
 interface Test {
     key: string;
@@ -31,7 +31,7 @@ test.cb("demux() constructor should be called once per key", t => {
         return dest;
     });
 
-    const demuxed = demux(construct, "key", {});
+    const demuxed = demux(construct, "key", { objectMode: true });
 
     demuxed.on("finish", () => {
         expect(construct.withArgs("a").callCount).to.equal(1);
@@ -41,35 +41,8 @@ test.cb("demux() constructor should be called once per key", t => {
         t.end();
     });
 
-    fromArray(input).pipe(demuxed);
-});
-
-test.cb("demux() item written passed in constructor", t => {
-    t.plan(4);
-    const input = [
-        { key: "a", visited: [] },
-        { key: "b", visited: [] },
-        { key: "c", visited: [] },
-    ];
-    const construct = sinon.spy((destKey: string, item: any) => {
-        expect(item).to.deep.equal({ key: destKey, visited: [] });
-        t.pass();
-        const dest = map((chunk: Test) => {
-            chunk.visited.push(1);
-            return chunk;
-        });
-
-        return dest;
-    });
-
-    const demuxed = demux(construct, "key", {});
-
-    demuxed.on("finish", () => {
-        t.pass();
-        t.end();
-    });
-
-    fromArray(input).pipe(demuxed);
+    input.forEach(event => demuxed.write(event));
+    demuxed.end();
 });
 
 test.cb("demux() should send input through correct pipeline", t => {
@@ -93,7 +66,7 @@ test.cb("demux() should send input through correct pipeline", t => {
         return dest;
     };
 
-    const demuxed = demux(construct, "key", {});
+    const demuxed = demux(construct, "key", { objectMode: true });
 
     demuxed.on("finish", () => {
         pipelineSpies["a"].getCalls().forEach(call => {
@@ -111,7 +84,8 @@ test.cb("demux() should send input through correct pipeline", t => {
         t.end();
     });
 
-    fromArray(input).pipe(demuxed);
+    input.forEach(event => demuxed.write(event));
+    demuxed.end();
 });
 
 test.cb("demux() constructor should be called once per key using keyBy", t => {
@@ -134,7 +108,7 @@ test.cb("demux() constructor should be called once per key using keyBy", t => {
         return dest;
     });
 
-    const demuxed = demux(construct, item => item.key, {});
+    const demuxed = demux(construct, item => item.key, { objectMode: true });
 
     demuxed.on("finish", () => {
         expect(construct.withArgs("a").callCount).to.equal(1);
@@ -144,7 +118,8 @@ test.cb("demux() constructor should be called once per key using keyBy", t => {
         t.end();
     });
 
-    fromArray(input).pipe(demuxed);
+    input.forEach(event => demuxed.write(event));
+    demuxed.end();
 });
 
 test.cb("demux() should send input through correct pipeline using keyBy", t => {
@@ -168,7 +143,7 @@ test.cb("demux() should send input through correct pipeline using keyBy", t => {
         return dest;
     };
 
-    const demuxed = demux(construct, item => item.key, {});
+    const demuxed = demux(construct, item => item.key, { objectMode: true });
 
     demuxed.on("finish", () => {
         pipelineSpies["a"].getCalls().forEach(call => {
@@ -186,10 +161,11 @@ test.cb("demux() should send input through correct pipeline using keyBy", t => {
         t.end();
     });
 
-    fromArray(input).pipe(demuxed);
+    input.forEach(event => demuxed.write(event));
+    demuxed.end();
 });
 
-test("demux() write should return false and emit drain if more than highWaterMark items are buffered", t => {
+test("demux() write should return false after if it has >= highWaterMark items buffered and drain should be emitted", t => {
     return new Promise(async (resolve, reject) => {
         t.plan(7);
         interface Chunk {
@@ -213,7 +189,7 @@ test("demux() write should return false and emit drain if more than highWaterMar
                     await sleep(slowProcessorSpeed);
                     return { ...chunk, mapped: [1] };
                 },
-                { highWaterMark: 1 },
+                { highWaterMark: 1, objectMode: true },
             );
 
             first.on("data", chunk => {
@@ -229,6 +205,7 @@ test("demux() write should return false and emit drain if more than highWaterMar
         };
 
         const _demux = demux(construct, "key", {
+            objectMode: true,
             highWaterMark,
         });
 
@@ -252,7 +229,7 @@ test("demux() write should return false and emit drain if more than highWaterMar
     });
 });
 
-test("demux() should emit one drain event after slowProcessorSpeed * highWaterMark ms when first stream is bottleneck", t => {
+test("demux() should emit one drain event after slowProcessorSpeed * highWaterMark ms", t => {
     return new Promise(async (resolve, reject) => {
         t.plan(7);
         interface Chunk {
@@ -278,7 +255,7 @@ test("demux() should emit one drain event after slowProcessorSpeed * highWaterMa
                     chunk.mapped.push(1);
                     return chunk;
                 },
-                { highWaterMark: 1 },
+                { highWaterMark: 1, objectMode: true },
             );
 
             first.on("data", () => {
@@ -291,6 +268,7 @@ test("demux() should emit one drain event after slowProcessorSpeed * highWaterMa
             return first;
         };
         const _demux = demux(construct, "key", {
+            objectMode: true,
             highWaterMark,
         });
         _demux.on("error", err => {
@@ -318,7 +296,7 @@ test("demux() should emit one drain event after slowProcessorSpeed * highWaterMa
 
 test("demux() should emit one drain event when writing 6 items with highWaterMark of 5", t => {
     return new Promise(async (resolve, reject) => {
-        t.plan(1);
+        t.plan(7);
         interface Chunk {
             key: string;
             mapped: number[];
@@ -340,11 +318,12 @@ test("demux() should emit one drain event when writing 6 items with highWaterMar
                     chunk.mapped.push(2);
                     return chunk;
                 },
-                { highWaterMark: 1 },
+                { highWaterMark: 1, objectMode: true },
             );
 
             first.on("data", () => {
                 pendingReads--;
+                t.pass();
                 if (pendingReads === 0) {
                     resolve();
                 }
@@ -352,6 +331,7 @@ test("demux() should emit one drain event when writing 6 items with highWaterMar
             return first;
         };
         const _demux = demux(construct, "key", {
+            objectMode: true,
             highWaterMark: 5,
         });
 
@@ -375,10 +355,9 @@ test("demux() should emit one drain event when writing 6 items with highWaterMar
     });
 });
 
-test.cb(
-    "demux() should emit drain event when second stream is bottleneck after (highWaterMark - 2) * slowProcessorSpeed ms",
+test.cb.only(
+    "demux() should emit drain event when third stream is bottleneck",
     t => {
-        // ie) first two items are pushed directly into first and second streams (highWaterMark - 2 remain in demux)
         t.plan(8);
         const slowProcessorSpeed = 100;
         const highWaterMark = 5;
@@ -404,7 +383,7 @@ test.cb(
                     chunk.mapped.push(1);
                     return chunk;
                 },
-                { highWaterMark: 1 },
+                { objectMode: true, highWaterMark: 1 },
             );
 
             const second = map(
@@ -413,23 +392,25 @@ test.cb(
                     chunk.mapped.push(2);
                     return chunk;
                 },
-                { highWaterMark: 1 },
+                { objectMode: true, highWaterMark: 1 },
             );
 
             first.pipe(second).pipe(sink);
             return first;
         };
         const _demux = demux(construct, () => "a", {
+            objectMode: true,
             highWaterMark,
         });
         _demux.on("error", err => {
             t.end(err);
         });
 
+        // This event should be received after at least 5 * slowProcessorSpeed (two are read immediately by first and second, 5 remaining in demux before drain event)
         _demux.on("drain", () => {
             expect(_demux._writableState.length).to.be.equal(0);
             expect(performance.now() - start).to.be.greaterThan(
-                slowProcessorSpeed * 3,
+                slowProcessorSpeed * (input.length - 2),
             );
             t.pass();
         });
@@ -446,14 +427,15 @@ test.cb(
         let pendingReads = input.length;
 
         const start = performance.now();
-        fromArray(input).pipe(_demux);
+        input.forEach(item => {
+            _demux.write(item);
+        });
     },
 );
 
 test.cb(
-    "demux() should emit drain event when third stream is bottleneck",
+    "demux() should emit drain event when second stream is bottleneck",
     t => {
-        // @TODO investigate why drain is emitted after slowProcessorSpeed
         t.plan(8);
         const slowProcessorSpeed = 100;
         const highWaterMark = 5;
@@ -464,7 +446,7 @@ test.cb(
         const sink = new Writable({
             objectMode: true,
             write(chunk, encoding, cb) {
-                expect(chunk.mapped).to.deep.equal([1, 2, 3]);
+                expect(chunk.mapped).to.deep.equal([1, 2]);
                 t.pass();
                 pendingReads--;
                 if (pendingReads === 0) {
@@ -479,14 +461,14 @@ test.cb(
                     chunk.mapped.push(1);
                     return chunk;
                 },
-                { highWaterMark: 1 },
+                { objectMode: true, highWaterMark: 1 },
             );
             const second = map(
                 (chunk: Chunk) => {
                     chunk.mapped.push(2);
                     return chunk;
                 },
-                { highWaterMark: 1 },
+                { objectMode: true, highWaterMark: 1 },
             );
 
             const third = map(
@@ -495,7 +477,7 @@ test.cb(
                     chunk.mapped.push(3);
                     return chunk;
                 },
-                { highWaterMark: 1 },
+                { objectMode: true, highWaterMark: 1 },
             );
 
             first
@@ -505,16 +487,18 @@ test.cb(
             return first;
         };
         const _demux = demux(construct, () => "a", {
+            objectMode: true,
             highWaterMark,
         });
         _demux.on("error", err => {
             t.end(err);
         });
 
+        // This event should be received after at least 3 * slowProcessorSpeed (two are read immediately by first and second, 3 remaining in demux before drain event)
         _demux.on("drain", () => {
             expect(_demux._writableState.length).to.be.equal(0);
             expect(performance.now() - start).to.be.greaterThan(
-                slowProcessorSpeed,
+                slowProcessorSpeed * (input.length - 4),
             );
             t.pass();
         });
@@ -531,7 +515,9 @@ test.cb(
         let pendingReads = input.length;
 
         const start = performance.now();
-        fromArray(input).pipe(_demux);
+        input.forEach(item => {
+            _demux.write(item);
+        });
     },
 );
 
@@ -550,37 +536,36 @@ test("demux() should be blocked by slowest pipeline", t => {
                     chunk.mapped.push(1);
                     return chunk;
                 },
-                { highWaterMark: 1 },
+                { objectMode: true, highWaterMark: 1 },
             );
 
+            first.on("data", chunk => {
+                pendingReads--;
+                if (chunk.key === "b") {
+                    expect(performance.now() - start).to.be.greaterThan(
+                        slowProcessorSpeed * totalItems,
+                    );
+                    t.pass();
+                    expect(pendingReads).to.equal(0);
+                    resolve();
+                }
+            });
             return first;
         };
-
         const _demux = demux(construct, "key", {
+            objectMode: true,
             highWaterMark: 1,
         });
-
         _demux.on("error", err => {
             reject(err);
         });
 
-        _demux.on("data", async chunk => {
-            pendingReads--;
-            if (chunk.key === "b") {
-                expect(performance.now() - start).to.be.greaterThan(
-                    slowProcessorSpeed * totalItems,
-                );
-                t.pass();
-                expect(pendingReads).to.equal(0);
-                resolve();
-            }
-        });
-
         const input = [
             { key: "a", mapped: [] },
             { key: "a", mapped: [] },
             { key: "c", mapped: [] },
             { key: "c", mapped: [] },
+            { key: "c", mapped: [] },
             { key: "b", mapped: [] },
         ];
 
@@ -599,266 +584,74 @@ test("demux() should be blocked by slowest pipeline", t => {
     });
 });
 
-test.cb("Demux should remux to sink", t => {
-    t.plan(6);
-    let i = 0;
-    const input = [
-        { key: "a", visited: [] },
-        { key: "b", visited: [] },
-        { key: "a", visited: [] },
-        { key: "c", visited: [] },
-        { key: "a", visited: [] },
-        { key: "b", visited: [] },
-    ];
-    const result = [
-        { key: "a", visited: ["a"] },
-        { key: "b", visited: ["b"] },
-        { key: "a", visited: ["a"] },
-        { key: "c", visited: ["c"] },
-        { key: "a", visited: ["a"] },
-        { key: "b", visited: ["b"] },
-    ];
-    const construct = (destKey: string) => {
-        const dest = map((chunk: any) => {
-            chunk.visited.push(destKey);
-            return chunk;
-        });
-
-        return dest;
-    };
-
-    const sink = map(d => {
-        t.deepEqual(d, result[i]);
-        i++;
-        if (i === input.length) {
-            t.end();
+test("demux() should emit drain event when second stream in pipeline is bottleneck", t => {
+    t.plan(5);
+    const highWaterMark = 3;
+    return new Promise(async (resolve, reject) => {
+        interface Chunk {
+            key: string;
+            mapped: number[];
         }
-    });
-
-    const demuxed = demux(construct, "key", {});
-
-    fromArray(input)
-        .pipe(demuxed)
-        .pipe(sink);
-});
-
-test.cb("Demux should send data events", t => {
-    t.plan(6);
-    let i = 0;
-    const input = [
-        { key: "a", visited: [] },
-        { key: "b", visited: [] },
-        { key: "a", visited: [] },
-        { key: "c", visited: [] },
-        { key: "a", visited: [] },
-        { key: "b", visited: [] },
-    ];
-    const result = [
-        { key: "a", visited: ["a"] },
-        { key: "b", visited: ["b"] },
-        { key: "a", visited: ["a"] },
-        { key: "c", visited: ["c"] },
-        { key: "a", visited: ["a"] },
-        { key: "b", visited: ["b"] },
-    ];
-    const construct = (destKey: string) => {
-        const dest = map((chunk: any) => {
-            chunk.visited.push(destKey);
-            return chunk;
+        const sink = new Writable({
+            objectMode: true,
+            write(chunk, encoding, cb) {
+                expect(chunk.mapped).to.deep.equal([1, 2]);
+                t.pass();
+                cb();
+                if (pendingReads === 0) {
+                    resolve();
+                }
+            },
         });
 
-        return dest;
-    };
+        const construct = (destKey: string) => {
+            const first = map(
+                (chunk: Chunk) => {
+                    expect(first._readableState.length).to.be.at.most(2);
+                    chunk.mapped.push(1);
+                    return chunk;
+                },
+                { objectMode: true, highWaterMark: 2 },
+            );
 
-    const demuxed = demux(construct, "key", {});
+            const second = map(
+                async (chunk: Chunk) => {
+                    await sleep(100);
+                    chunk.mapped.push(2);
+                    expect(second._writableState.length).to.be.equal(1);
+                    pendingReads--;
+                    return chunk;
+                },
+                { objectMode: true, highWaterMark: 1 },
+            );
 
-    fromArray(input).pipe(demuxed);
+            first.pipe(second).pipe(sink);
+            return first;
+        };
 
-    demuxed.on("data", d => {
-        t.deepEqual(d, result[i]);
-        i++;
-        if (i === input.length) {
-            t.end();
-        }
-    });
-});
-
-test.cb("demux() `finish` and `end` propagates", t => {
-    interface Chunk {
-        key: string;
-        mapped: number[];
-    }
-
-    t.plan(9);
-
-    const construct = (destKey: string) => {
-        const dest = map((chunk: any) => {
-            chunk.mapped.push(destKey);
-            return chunk;
+        const _demux = demux(construct, "key", {
+            objectMode: true,
+            highWaterMark,
         });
-        return dest;
-    };
-
-    const _demux = demux(construct, "key", {
-        highWaterMark: 3,
-    });
-
-    const fakeSource = new Readable({
-        objectMode: true,
-        read() {
-            return;
-        },
-    });
-
-    const sink = map((d: any) => {
-        const curr = input.shift();
-        t.is(curr.key, d.key);
-        t.deepEqual(d.mapped, [d.key]);
-    });
-
-    fakeSource.pipe(_demux).pipe(sink);
-
-    fakeSource.on("end", () => {
-        t.pass();
-    });
-    _demux.on("finish", () => {
-        t.pass();
-    });
-    _demux.on("unpipe", () => {
-        t.pass();
-    });
-    _demux.on("end", () => {
-        t.pass();
-        t.end();
-    });
-    sink.on("finish", () => {
-        t.pass();
-    });
-
-    const input = [
-        { key: "a", mapped: [] },
-        { key: "b", mapped: [] },
-        { key: "a", mapped: [] },
-        { key: "a", mapped: [] },
-        { key: "b", mapped: [] },
-    ];
-    fakeSource.push(input[0]);
-    fakeSource.push(input[1]);
-    fakeSource.push(null);
-});
-
-test.cb("demux() `unpipe` propagates", t => {
-    interface Chunk {
-        key: string;
-        mapped: number[];
-    }
-
-    t.plan(7);
-
-    const construct = (destKey: string) => {
-        const dest = map((chunk: any) => {
-            chunk.mapped.push(destKey);
-            return chunk;
+        _demux.on("error", err => {
+            reject();
         });
-        return dest;
-    };
 
-    const _demux = demux(construct, "key", {
-        highWaterMark: 3,
-    });
-
-    const fakeSource = new Readable({
-        objectMode: true,
-        read() {
-            return;
-        },
-    });
-
-    const sink = map((d: any) => {
-        const curr = input.shift();
-        t.is(curr.key, d.key);
-        t.deepEqual(d.mapped, [d.key]);
-    });
-
-    fakeSource.pipe(_demux).pipe(sink);
-
-    _demux.on("unpipe", () => {
-        t.pass();
-    });
-
-    sink.on("unpipe", () => {
-        t.pass();
-    });
-
-    sink.on("finish", () => {
-        t.pass();
-        t.end();
-    });
-
-    const input = [
-        { key: "a", mapped: [] },
-        { key: "b", mapped: [] },
-        { key: "a", mapped: [] },
-        { key: "a", mapped: [] },
-        { key: "b", mapped: [] },
-    ];
-    fakeSource.push(input[0]);
-    fakeSource.push(input[1]);
-    fakeSource.push(null);
-});
-
-test.cb("demux() should be 'destroyable'", t => {
-    t.plan(2);
-    const _sleep = 100;
-    interface Chunk {
-        key: string;
-        mapped: string[];
-    }
-
-    const construct = (destKey: string) => {
-        const first = map(async (chunk: Chunk) => {
-            await sleep(_sleep);
-            chunk.mapped.push(destKey);
-            return chunk;
+        _demux.on("drain", () => {
+            expect(_demux._writableState.length).to.be.equal(0);
+            t.pass();
         });
-        return first;
-    };
 
-    const _demux = demux(construct, "key");
+        const input = [
+            { key: "a", mapped: [] },
+            { key: "a", mapped: [] },
+            { key: "a", mapped: [] },
+            { key: "a", mapped: [] },
+        ];
+        let pendingReads = input.length;
 
-    const fakeSource = new Readable({
-        objectMode: true,
-        read() {
-            return;
-        },
+        input.forEach(item => {
+            _demux.write(item);
+        });
     });
-
-    const fakeSink = new Writable({
-        objectMode: true,
-        write(data, enc, cb) {
-            const cur = input.shift();
-            t.is(cur.key, data.key);
-            t.deepEqual(cur.mapped, ["a"]);
-            if (cur.key === "a") {
-                _demux.destroy();
-            }
-            cb();
-        },
-    });
-
-    _demux.on("close", t.end);
-    fakeSource.pipe(_demux).pipe(fakeSink);
-
-    const input = [
-        { key: "a", mapped: [] },
-        { key: "b", mapped: [] },
-        { key: "c", mapped: [] },
-        { key: "d", mapped: [] },
-        { key: "e", mapped: [] },
-    ];
-    fakeSource.push(input[0]);
-    fakeSource.push(input[1]);
-    fakeSource.push(input[2]);
-    fakeSource.push(input[3]);
-    fakeSource.push(input[4]);
 });

tests/utils/collected.spec.ts

@@ -1,9 +0,0 @@
-import test from "ava";
-import { collected } from "../../src/utils";
-import mhysa from "../../src";
-const { fromArray, collect } = mhysa({ objectMode: true });
-
-test("collected returns a promise for the first data point", async t => {
-    const data = collected(fromArray([1, 2, 3, 4]).pipe(collect()));
-    t.deepEqual(await data, [1, 2, 3, 4]);
-});

yarn.lock

@@ -409,10 +409,10 @@
   resolved "https://registry.yarnpkg.com/@types/node/-/node-12.7.2.tgz#c4e63af5e8823ce9cc3f0b34f7b998c2171f0c44"
   integrity sha512-dyYO+f6ihZEtNPDcWNR1fkoTDf3zAK3lAABDze3mz6POyIercH0lEUawUFXlG8xaQZmm1yEBON/4TsYv/laDYg==
 
-"@types/node@^12.12.15":
-  version "12.12.15"
-  resolved "https://npm.dev.jogogo.co/@types%2fnode/-/node-12.12.15.tgz#8dfb6ce22fedd469128137640a3aa8f17415422f"
-  integrity sha512-Pv+vWicyFd07Hw/SmNnTUguqrHgDfMtjabvD9sQyxeqbpCEg8CmViLBaVPHtNsoBgZECrRf5/pgV6FJIBrGSjw==
+"@types/node@^12.7.2":
+  version "12.12.14"
+  resolved "https://npm.dev.jogogo.co/@types%2fnode/-/node-12.12.14.tgz#1c1d6e3c75dba466e0326948d56e8bd72a1903d2"
+  integrity sha512-u/SJDyXwuihpwjXy7hOOghagLEV1KdAST6syfnOk6QZAMzZuWZqXy5aYYZbh8Jdpd4escVFP0MvftHNDb9pruA==
 
 "@types/sinon@^7.0.13":
   version "7.5.1"
@@ -1935,7 +1935,7 @@ merge2@^1.2.3, merge2@^1.3.0:
   integrity sha512-2j4DAdlBOkiSZIsaXk4mTE3sRS02yBHAtfy127xRV3bQUFqXkjHCHLW6Scv7DwNRbIWNHH8zpnz9zMaKXIdvYw==
 
 mhysa@./:
-  version "2.0.0-alpha.1"
+  version "0.0.1-beta.4"
 
 micromatch@^4.0.2:
   version "4.0.2"