No need to create an array to call concat

* Demux handles arrays by key

* Rename construct

* version

* Minor changes to types

README.md

@@ -1,15 +1,13 @@
-# Mhysa
+# Strom 
 
 **Dependency-free stream utils for Node.js**
 
-<sub>Released under the [MIT](https://github.com/Wenzil/Mhysa/blob/master/LICENSE) license.</sub>
+<sub>Released under the [MIT](https://git.lewis.id/strom/blob/master/LICENSE) license.</sub>
 
 ```sh
-yarn add mhysa
+yarn add strom  // Name TBD
 ```
 
-<sub>Tested with Node.js versions 8+</sub>
-
 ## fromArray(array)
 Convert an array into a `Readable` stream of its elements
 
@@ -18,7 +16,7 @@ Convert an array into a `Readable` stream of its elements
 | `array` | `T[]` | Array of elements to stream |
 
 ```js
-Mhysa.fromArray(["a", "b"])
+strom.fromArray(["a", "b"])
     .pipe(process.stdout);
 // ab is printed out
 ```
@@ -35,8 +33,8 @@ Return a `ReadWrite` stream that maps streamed chunks
 | `options.writableObjectMode` | `boolean` | Whether this stream should behave as a writable stream of objects |
 
 ```js
-Mhysa.fromArray(["a", "b"])
+strom.fromArray(["a", "b"])
-    .pipe(Mhysa.map(s => s.toUpperCase()))
+    .pipe(strom.map(s => s.toUpperCase()))
     .pipe(process.stdout);
 // AB is printed out
 ```
@@ -53,8 +51,8 @@ Return a `ReadWrite` stream that flat maps streamed chunks
 | `options.writableObjectMode` | `boolean` | Whether this stream should behave as a writable stream of objects |
 
 ```js
-Mhysa.fromArray(["a", "AA"])
+strom.fromArray(["a", "AA"])
-    .pipe(Mhysa.flatMap(s => new Array(s.length).fill(s)))
+    .pipe(strom.flatMap(s => new Array(s.length).fill(s)))
     .pipe(process.stdout);
 // aAAAA is printed out
 ```
@@ -70,8 +68,8 @@ Return a `ReadWrite` stream that filters out streamed chunks for which the predi
 | `options.objectMode` | `boolean` | `boolean` | Whether this stream should behave as a stream of objects |
 
 ```js
-Mhysa.fromArray(["a", "b", "c"])
+strom.fromArray(["a", "b", "c"])
-    .pipe(Mhysa.filter(s => s !== "b"))
+    .pipe(strom.filter(s => s !== "b"))
     .pipe(process.stdout);
 // ac is printed out
 ```
@@ -90,9 +88,9 @@ value
 | `options.writableObjectMode` | `boolean` | Whether this stream should behave as a writable stream of objects |
 
 ```js
-Mhysa.fromArray(["a", "b", "cc"])
+strom.fromArray(["a", "b", "cc"])
-    .pipe(Mhysa.reduce((acc, s) => ({ ...acc, [s]: s.length }), {}))
+    .pipe(strom.reduce((acc, s) => ({ ...acc, [s]: s.length }), {}))
-    .pipe(Mhysa.stringify())
+    .pipe(strom.stringify())
     .pipe(process.stdout);
 // {"a":1,"b":1","c":2} is printed out
 ```
@@ -108,9 +106,9 @@ Return a `ReadWrite` stream that splits streamed chunks using the given separato
 | `options.encoding` | `string` | Character encoding to use for decoding chunks. Defaults to utf8
 
 ```js
-Mhysa.fromArray(["a,b", "c,d"])
+strom.fromArray(["a,b", "c,d"])
-    .pipe(Mhysa.split(","))
+    .pipe(strom.split(","))
-    .pipe(Mhysa.join("|"))
+    .pipe(strom.join("|"))
     .pipe(process.stdout);
 // a|bc|d is printed out
 ```
@@ -126,8 +124,8 @@ Return a `ReadWrite` stream that joins streamed chunks using the given separator
 | `options.encoding` | `string` | Character encoding to use for decoding chunks. Defaults to utf8
 
 ```js
-Mhysa.fromArray(["a", "b", "c"])
+strom.fromArray(["a", "b", "c"])
-    .pipe(Mhysa.join(","))
+    .pipe(strom.join(","))
     .pipe(process.stdout);
 // a,b,c is printed out
 ```
@@ -145,8 +143,8 @@ the streamed chunks with the specified replacement string
 | `options.encoding` | `string` | Character encoding to use for decoding chunks. Defaults to utf8
 
 ```js
-Mhysa.fromArray(["a1", "b22", "c333"])
+strom.fromArray(["a1", "b22", "c333"])
-    .pipe(Mhysa.replace(/b\d+/, "B"))
+    .pipe(strom.replace(/b\d+/, "B"))
     .pipe(process.stdout);
 // a1Bc333 is printed out
 ```
@@ -156,8 +154,8 @@ Mhysa.fromArray(["a1", "b22", "c333"])
 Return a `ReadWrite` stream that parses the streamed chunks as JSON
 
 ```js
-Mhysa.fromArray(['{ "a": "b" }'])
+strom.fromArray(['{ "a": "b" }'])
-    .pipe(Mhysa.parse())
+    .pipe(strom.parse())
     .once("data", object => console.log(object));
 // { a: 'b' } is printed out
 ```
@@ -167,8 +165,8 @@ Mhysa.fromArray(['{ "a": "b" }'])
 Return a `ReadWrite` stream that stringifies the streamed chunks to JSON
 
 ```js
-Mhysa.fromArray([{ a: "b" }])
+strom.fromArray([{ a: "b" }])
-    .pipe(Mhysa.stringify())
+    .pipe(strom.stringify())
     .pipe(process.stdout);
 // {"a":"b"} is printed out
 ```
@@ -183,8 +181,8 @@ Return a `ReadWrite` stream that collects streamed chunks into an array or buffe
 | `options.objectMode` | `boolean` | Whether this stream should behave as a stream of objects |
 
 ```js
-Mhysa.fromArray(["a", "b", "c"])
+strom.fromArray(["a", "b", "c"])
-    .pipe(Mhysa.collect({ objectMode: true }))
+    .pipe(strom.collect({ objectMode: true }))
     .once("data", object => console.log(object));
 // [ 'a', 'b', 'c' ] is printed out
 ```
@@ -200,7 +198,7 @@ Return a `Readable` stream of readable streams concatenated together
 ```js
 const source1 = new Readable();
 const source2 = new Readable();
-Mhysa.concat(source1, source2).pipe(process.stdout)
+strom.concat(source1, source2).pipe(process.stdout)
 source1.push("a1 ");
 source2.push("c3 ");
 source1.push("b2 ");
@@ -221,7 +219,7 @@ Return a `Readable` stream of readable streams merged together in chunk arrival
 ```js
 const source1 = new Readable({ read() {} });
 const source2 = new Readable({ read() {} });
-Mhysa.merge(source1, source2).pipe(process.stdout);
+strom.merge(source1, source2).pipe(process.stdout);
 source1.push("a1 ");
 setTimeout(() => source2.push("c3 "), 10);
 setTimeout(() => source1.push("b2 "), 20);
@@ -243,8 +241,8 @@ cause the given readable stream to yield chunks
 
 ```js
 const catProcess = require("child_process").exec("grep -o ab");
-Mhysa.fromArray(["a", "b", "c"])
+strom.fromArray(["a", "b", "c"])
-    .pipe(Mhysa.duplex(catProcess.stdin, catProcess.stdout))
+    .pipe(strom.duplex(catProcess.stdin, catProcess.stdout))
     .pipe(process.stdout);
 // ab is printed out
 ```
@@ -259,8 +257,8 @@ Return a `Duplex` stream from a child process' stdin and stdout
 
 ```js
 const catProcess = require("child_process").exec("grep -o ab");
-Mhysa.fromArray(["a", "b", "c"])
+strom.fromArray(["a", "b", "c"])
-    .pipe(Mhysa.child(catProcess))
+    .pipe(strom.child(catProcess))
     .pipe(process.stdout);
 // ab is printed out
 ```
@@ -276,8 +274,8 @@ ended
 
 ```js
 let f = async () => {
-    const source = Mhysa.fromArray(["a", "b", "c"]);
+    const source = strom.fromArray(["a", "b", "c"]);
-    console.log(await Mhysa.last(source));
+    console.log(await strom.last(source));
 };
 f();
 // c is printed out

package.json

@@ -1,17 +1,16 @@
 {
-    "name": "mhysa",
+    "name": "strom",
-    "version": "2.0.0-alpha.1",
+    "version": "2.0.0",
-    "description": "Streams and event emitter utils for Node.js",
+    "description": "Streams utils for Node.js",
     "keywords": [
         "promise",
         "stream",
-        "event emitter",
         "utils"
     ],
-    "author": {
-        "name": "Wenzil"
-    },
     "contributors": [
+        {
+            "name": "Wenzil"
+        },
         {
             "name": "jerry",
             "email": "jerry@jogogo.co"
@@ -28,19 +27,20 @@
         "dist"
     ],
     "repository": {
-        "url": "git@github.com:Wenzil/Mhysa.git",
+        "url": "git@git.lewis.id:ldiamond/strom.git",
         "type": "git"
     },
     "scripts": {
         "test": "ava",
         "lint": "tslint -p tsconfig.json",
         "validate:tslint": "tslint-config-prettier-check ./tslint.json",
-        "prepublishOnly": "yarn lint && yarn test && yarn tsc -d"
+        "prepublishOnly": "yarn lint && yarn test && yarn tsc -d",
+        "prepare": "tsc"
     },
     "dependencies": {},
     "devDependencies": {
         "@types/chai": "^4.1.7",
-        "@types/node": "^12.7.2",
+        "@types/node": "^12.12.15",
         "@types/sinon": "^7.0.13",
         "ava": "^2.4.0",
         "chai": "^4.2.0",
@@ -55,7 +55,8 @@
     },
     "ava": {
         "files": [
-            "tests/*.spec.ts"
+            "tests/*.spec.ts",
+            "tests/utils/*.spec.ts"
         ],
         "sources": [
             "src/**/*.ts"

samples/async.js

@@ -1,8 +1,8 @@
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 
-Mhysa.fromArray(["a", "b", "c"])
+strom.fromArray(["a", "b", "c"])
-    .pipe(Mhysa.map(s => Promise.resolve(s + s)))
+    .pipe(strom.map(s => Promise.resolve(s + s)))
-    .pipe(Mhysa.flatMap(s => Promise.resolve([s, s.toUpperCase()])))
+    .pipe(strom.flatMap(s => Promise.resolve([s, s.toUpperCase()])))
-    .pipe(Mhysa.filter(s => Promise.resolve(s !== "bb")))
+    .pipe(strom.filter(s => Promise.resolve(s !== "bb")))
-    .pipe(Mhysa.join(","))
+    .pipe(strom.join(","))
     .pipe(process.stdout);

samples/child.js

@@ -1,6 +1,6 @@
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 const catProcess = require("child_process").exec("grep -o ab");
 
-Mhysa.fromArray(["a", "b", "c"])
+strom.fromArray(["a", "b", "c"])
-    .pipe(Mhysa.child(catProcess))
+    .pipe(strom.child(catProcess))
     .pipe(process.stdout);

samples/collect.js

@@ -1,5 +1,5 @@
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 
-Mhysa.fromArray(["a", "b", "c"])
+strom.fromArray(["a", "b", "c"])
-    .pipe(Mhysa.collect({ objectMode: true }))
+    .pipe(strom.collect({ objectMode: true }))
     .on("data", object => console.log(object));

samples/concat.js

@@ -1,9 +1,9 @@
 const { Readable } = require("stream");
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 
 const source1 = new Readable();
 const source2 = new Readable();
-Mhysa.concat(source1, source2).pipe(process.stdout);
+strom.concat(source1, source2).pipe(process.stdout);
 source1.push("a1 ");
 source2.push("c3 ");
 source1.push("b2 ");

samples/duplex.js

@@ -1,6 +1,6 @@
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 const catProcess = require("child_process").exec("grep -o ab");
 
-Mhysa.fromArray(["a", "b", "c"])
+strom.fromArray(["a", "b", "c"])
-    .pipe(Mhysa.duplex(catProcess.stdin, catProcess.stdout))
+    .pipe(strom.duplex(catProcess.stdin, catProcess.stdout))
     .pipe(process.stdout);

samples/filter.js

@@ -1,6 +1,6 @@
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 
-Mhysa.fromArray(["a", "b", "c"])
+strom.fromArray(["a", "b", "c"])
-    .pipe(Mhysa.filter(s => s !== "b"))
+    .pipe(strom.filter(s => s !== "b"))
-    .pipe(Mhysa.join(","))
+    .pipe(strom.join(","))
     .pipe(process.stdout);

samples/flatMap.js

@@ -1,6 +1,6 @@
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 
-Mhysa.fromArray(["a", "AA"])
+strom.fromArray(["a", "AA"])
-    .pipe(Mhysa.flatMap(s => new Array(s.length).fill(s)))
+    .pipe(strom.flatMap(s => new Array(s.length).fill(s)))
-    .pipe(Mhysa.join(","))
+    .pipe(strom.join(","))
     .pipe(process.stdout);

samples/join.js

@@ -1,5 +1,5 @@
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 
-Mhysa.fromArray(["a", "b", "c"])
+strom.fromArray(["a", "b", "c"])
-    .pipe(Mhysa.join(","))
+    .pipe(strom.join(","))
     .pipe(process.stdout);

samples/last.js

@@ -1,7 +1,7 @@
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 
 let f = async () => {
-    const source = Mhysa.fromArray(["a", "b", "c"]);
+    const source = strom.fromArray(["a", "b", "c"]);
-    console.log(await Mhysa.last(source));
+    console.log(await strom.last(source));
 };
 f();

samples/map.js

@@ -1,6 +1,6 @@
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 
-Mhysa.fromArray(["a", "b"])
+strom.fromArray(["a", "b"])
-    .pipe(Mhysa.map(s => s.toUpperCase()))
+    .pipe(strom.map(s => s.toUpperCase()))
-    .pipe(Mhysa.join(","))
+    .pipe(strom.join(","))
     .pipe(process.stdout);

samples/merge.js

@@ -1,9 +1,9 @@
 const { Readable } = require("stream");
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 
 const source1 = new Readable({ read() {} });
 const source2 = new Readable({ read() {} });
-Mhysa.merge(source1, source2).pipe(process.stdout);
+strom.merge(source1, source2).pipe(process.stdout);
 source1.push("a1 ");
 setTimeout(() => source2.push("c3 "), 10);
 setTimeout(() => source1.push("b2 "), 20);

samples/parse.js

@@ -1,5 +1,5 @@
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 
-Mhysa.fromArray(['{ "a": "b" }'])
+strom.fromArray(['{ "a": "b" }'])
-    .pipe(Mhysa.parse())
+    .pipe(strom.parse())
     .on("data", object => console.log(object));

samples/reduce.js

@@ -1,6 +1,6 @@
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 
-Mhysa.fromArray(["a", "b", "cc"])
+strom.fromArray(["a", "b", "cc"])
-    .pipe(Mhysa.reduce((acc, s) => ({ ...acc, [s]: s.length }), {}))
+    .pipe(strom.reduce((acc, s) => ({ ...acc, [s]: s.length }), {}))
-    .pipe(Mhysa.stringify())
+    .pipe(strom.stringify())
     .pipe(process.stdout);

samples/replace.js

@@ -1,5 +1,5 @@
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 
-Mhysa.fromArray(["a1", "b22", "c333"])
+strom.fromArray(["a1", "b22", "c333"])
-    .pipe(Mhysa.replace(/b\d+/, "B"))
+    .pipe(strom.replace(/b\d+/, "B"))
     .pipe(process.stdout);

samples/split.js

@@ -1,6 +1,6 @@
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 
-Mhysa.fromArray(["a,b", "c,d"])
+strom.fromArray(["a,b", "c,d"])
-    .pipe(Mhysa.split(","))
+    .pipe(strom.split(","))
-    .pipe(Mhysa.join("|"))
+    .pipe(strom.join("|"))
     .pipe(process.stdout);

samples/stringify.js

@@ -1,5 +1,5 @@
-const Mhysa = require("mhysa");
+const strom = require("strom").strom();
 
-Mhysa.fromArray([{ a: "b" }])
+strom.fromArray([{ a: "b" }])
-    .pipe(Mhysa.stringify())
+    .pipe(strom.stringify())
     .pipe(process.stdout);

src/functions/batch.ts

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

src/functions/compose.ts

@@ -1,4 +1,5 @@
-import { pipeline, TransformOptions, Transform } from "stream";
+import { AllStreams, isReadable } from "../helpers";
+import { PassThrough, pipeline, TransformOptions, Transform } from "stream";
 
 export function compose(
     streams: Array<
@@ -21,18 +22,6 @@ 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;
@@ -45,11 +34,11 @@ export class Compose extends Transform {
         options?: TransformOptions,
     ) {
         super(options);
-        this.first = streams[0];
+        this.first = new PassThrough(options);
         this.last = streams[streams.length - 1];
         this.streams = streams;
         pipeline(
-            streams,
+            [this.first, ...streams],
             errorCallback ||
                 ((error: any) => {
                     if (error) {

src/functions/demux.ts

@@ -1,99 +1,131 @@
-import { WritableOptions, Writable } from "stream";
+import { DuplexOptions, Duplex, Transform } from "stream";
 
-enum EventSubscription {
+import { isReadable } from "../helpers";
-    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 function demux(
+export interface DemuxOptions extends DuplexOptions {
-    construct: (destKey?: string) => DemuxStreams,
+    remultiplex?: boolean;
-    demuxBy: string | ((chunk: any) => string),
-    options?: WritableOptions,
-): Writable {
-    return new Demux(construct, demuxBy, options);
 }
 
-// @TODO handle pipe event ie) Multiplex
+export function demux(
-class Demux extends Writable {
+    pipelineConstructor: (
+        destKey?: string,
+        chunk?: any,
+    ) => DemuxStreams | DemuxStreams[],
+    demuxBy: string | ((chunk: any) => string),
+    options?: DemuxOptions,
+): Duplex {
+    return new Demux(pipelineConstructor, demuxBy, options);
+}
+
+class Demux extends Duplex {
     private streamsByKey: {
-        [key: string]: DemuxStreams;
+        [key: string]: DemuxStreams[];
     };
     private demuxer: (chunk: any) => string;
-    private construct: (destKey?: string) => DemuxStreams;
+    private pipelineConstructor: (
+        destKey?: string,
+        chunk?: any,
+    ) => DemuxStreams[];
+    private remultiplex: boolean;
+    private transform: Transform;
     constructor(
-        construct: (destKey?: string) => DemuxStreams,
+        pipelineConstructor: (
+            destKey?: string,
+            chunk?: any,
+        ) => DemuxStreams | DemuxStreams[],
         demuxBy: string | ((chunk: any) => string),
-        options: WritableOptions = {},
+        options: DemuxOptions = {},
     ) {
         super(options);
         this.demuxer =
             typeof demuxBy === "string" ? chunk => chunk[demuxBy] : demuxBy;
-        this.construct = construct;
+        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.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) {
-            this.streamsByKey[destKey] = await this.construct(destKey);
+            const newPipelines = this.pipelineConstructor(destKey, chunk);
-        }
+            this.streamsByKey[destKey] = newPipelines;
-        if (!this.streamsByKey[destKey].write(chunk, encoding)) {
+
-            this.streamsByKey[destKey].once("drain", () => {
+            newPipelines.forEach(newPipeline => {
-                cb();
+                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`,
+                    );
+                }
             });
-        } else {
-            cb();
         }
+        const pipelines = this.streamsByKey[destKey];
+        const pendingDrains: Array<Promise<any>> = [];
+
+        pipelines.forEach(pipeline => {
+            if (!pipeline.write(chunk, encoding)) {
+                pendingDrains.push(
+                    new Promise(resolve => {
+                        pipeline.once("drain", () => {
+                            resolve();
+                        });
+                    }),
+                );
+            }
+        });
+        await Promise.all(pendingDrains);
+        cb();
     }
 
-    public on(event: string, cb: any) {
+    public _flush() {
-        switch (eventsTarget[event]) {
+        const pipelines: DemuxStreams[] = Array.prototype.concat.apply(
-            case EventSubscription.Self:
+            [],
-                super.on(event, cb);
+            Object.values(this.streamsByKey),
-                break;
+        );
-            case EventSubscription.All:
+        const flushPromises: Array<Promise<void>> = [];
-                Object.keys(this.streamsByKey).forEach(key =>
+        pipelines.forEach(pipeline => {
-                    this.streamsByKey[key].on(event, cb),
+            flushPromises.push(
-                );
+                new Promise(resolve => {
-                break;
+                    pipeline.once("end", () => {
-            default:
+                        resolve();
-                super.on(event, cb);
+                    });
-        }
+                }),
-        return this;
+            );
+        });
+        pipelines.forEach(pipeline => pipeline.end());
+        Promise.all(flushPromises).then(() => {
+            this.push(null);
+            this.emit("end");
+        });
     }
 
-    public once(event: string, cb: any) {
+    public _destroy(error: any, cb: (error?: any) => void) {
-        switch (eventsTarget[event]) {
+        const pipelines: DemuxStreams[] = [].concat.apply(
-            case EventSubscription.Self:
+            [],
-                super.once(event, cb);
+            Object.values(this.streamsByKey),
-                break;
+        );
-            case EventSubscription.All:
+        pipelines.forEach(p => (p as any).destroy());
-                Object.keys(this.streamsByKey).forEach(key =>
+        cb(error);
-                    this.streamsByKey[key].once(event, cb),
-                );
-                break;
-            default:
-                super.once(event, cb);
-        }
-        return this;
     }
 }

src/functions/index.ts

@@ -28,7 +28,7 @@ import { unbatch } from "./unbatch";
 import { compose } from "./compose";
 import { demux } from "./demux";
 
-export default function mhysa(defaultOptions?: TransformOptions) {
+export function strom(defaultOptions: TransformOptions = { objectMode: true }) {
     function withDefaultOptions<T extends any[], R>(
         n: number,
         fn: (...args: T) => R,

src/helpers.ts

@@ -1,3 +1,17 @@
 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,2 +1,2 @@
-import mhysa from "./functions";
+export { strom } from "./functions";
-export default mhysa;
+export * from "./utils";

src/utils/collected.ts

@@ -0,0 +1,12 @@
+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

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

tests/accumulator.spec.ts

@@ -1,10 +1,10 @@
 import test from "ava";
 import { expect } from "chai";
 import { Readable } from "stream";
-import mhysa from "../src";
+import { strom } from "../src";
 import { FlushStrategy } from "../src/functions/accumulator";
 import { performance } from "perf_hooks";
-const { accumulator, accumulatorBy } = mhysa({ objectMode: true });
+const { accumulator, accumulatorBy } = strom({ objectMode: true });
 
 test.cb("accumulator() rolling", t => {
     t.plan(3);
@@ -14,8 +14,14 @@ test.cb("accumulator() rolling", t => {
         key: string;
     }
     const source = new Readable({ objectMode: true });
-    const firstFlush = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }];
+    const firstFlush = [
-    const secondFlush = [{ ts: 2, key: "d" }, { ts: 3, key: "e" }];
+        { ts: 0, key: "a" },
+        { ts: 1, key: "b" },
+    ];
+    const secondFlush = [
+        { ts: 2, key: "d" },
+        { ts: 3, key: "e" },
+    ];
     const thirdFlush = [{ ts: 4, key: "f" }];
     const flushes = [firstFlush, secondFlush, thirdFlush];
 
@@ -88,7 +94,10 @@ test.cb(
             "nonExistingKey",
             { objectMode: true },
         );
-        const input = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }];
+        const input = [
+            { ts: 0, key: "a" },
+            { ts: 1, key: "b" },
+        ];
 
         source
             .pipe(accumulatorStream)
@@ -182,7 +191,10 @@ test.cb("accumulator() sliding", t => {
         { ts: 4, key: "d" },
     ];
     const firstFlush = [{ ts: 0, key: "a" }];
-    const secondFlush = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }];
+    const secondFlush = [
+        { ts: 0, key: "a" },
+        { ts: 1, key: "b" },
+    ];
     const thirdFlush = [
         { ts: 0, key: "a" },
         { ts: 1, key: "b" },
@@ -232,7 +244,10 @@ test.cb("accumulator() sliding with key", t => {
         { ts: 6, key: "g" },
     ];
     const firstFlush = [{ ts: 0, key: "a" }];
-    const secondFlush = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }];
+    const secondFlush = [
+        { ts: 0, key: "a" },
+        { ts: 1, key: "b" },
+    ];
     const thirdFlush = [
         { ts: 0, key: "a" },
         { ts: 1, key: "b" },
@@ -243,8 +258,14 @@ test.cb("accumulator() sliding with key", t => {
         { ts: 2, key: "c" },
         { ts: 3, key: "d" },
     ];
-    const fifthFlush = [{ ts: 3, key: "d" }, { ts: 5, key: "f" }];
+    const fifthFlush = [
-    const sixthFlush = [{ ts: 5, key: "f" }, { ts: 6, key: "g" }];
+        { ts: 3, key: "d" },
+        { ts: 5, key: "f" },
+    ];
+    const sixthFlush = [
+        { ts: 5, key: "f" },
+        { ts: 6, key: "g" },
+    ];
 
     const flushes = [
         firstFlush,
@@ -286,7 +307,10 @@ test.cb(
             "nonExistingKey",
             { objectMode: true },
         );
-        const input = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }];
+        const input = [
+            { ts: 0, key: "a" },
+            { ts: 1, key: "b" },
+        ];
 
         source
             .pipe(accumulatorStream)
@@ -334,10 +358,22 @@ test.cb(
             { ts: 6, key: "g" },
         ];
         const firstFlush = [{ ts: 0, key: "a" }];
-        const secondFlush = [{ ts: 0, key: "a" }, { ts: 2, key: "c" }];
+        const secondFlush = [
-        const thirdFlush = [{ ts: 2, key: "c" }, { ts: 3, key: "d" }];
+            { ts: 0, key: "a" },
-        const fourthFlush = [{ ts: 3, key: "d" }, { ts: 5, key: "f" }];
+            { ts: 2, key: "c" },
-        const fifthFlush = [{ ts: 5, key: "f" }, { ts: 6, key: "g" }];
+        ];
+        const thirdFlush = [
+            { ts: 2, key: "c" },
+            { ts: 3, key: "d" },
+        ];
+        const fourthFlush = [
+            { ts: 3, key: "d" },
+            { ts: 5, key: "f" },
+        ];
+        const fifthFlush = [
+            { ts: 5, key: "f" },
+            { ts: 6, key: "g" },
+        ];
 
         const flushes = [
             firstFlush,
@@ -469,7 +505,10 @@ test.cb("accumulatorBy() sliding", t => {
         { ts: 6, key: "g" },
     ];
     const firstFlush = [{ ts: 0, key: "a" }];
-    const secondFlush = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }];
+    const secondFlush = [
+        { ts: 0, key: "a" },
+        { ts: 1, key: "b" },
+    ];
     const thirdFlush = [
         { ts: 0, key: "a" },
         { ts: 1, key: "b" },
@@ -480,8 +519,14 @@ test.cb("accumulatorBy() sliding", t => {
         { ts: 2, key: "c" },
         { ts: 3, key: "d" },
     ];
-    const fifthFlush = [{ ts: 3, key: "d" }, { ts: 5, key: "f" }];
+    const fifthFlush = [
-    const sixthFlush = [{ ts: 5, key: "f" }, { ts: 6, key: "g" }];
+        { ts: 3, key: "d" },
+        { ts: 5, key: "f" },
+    ];
+    const sixthFlush = [
+        { ts: 5, key: "f" },
+        { ts: 6, key: "g" },
+    ];
 
     const flushes = [
         firstFlush,

tests/batch.spec.ts

@@ -1,8 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { batch } = mhysa({ objectMode: true });
+const { batch, map, fromArray } = strom({ objectMode: true });
 
 test.cb("batch() batches chunks together", t => {
     t.plan(3);
@@ -57,3 +57,28 @@ 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/child.spec.ts

@@ -2,8 +2,8 @@ import * as cp from "child_process";
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { child } = mhysa();
+const { child } = strom();
 
 test.cb(
     "child() allows easily writing to child process stdin and reading from its stdout",

tests/collect.spec.ts

@@ -1,8 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { collect } = mhysa();
+const { collect } = strom();
 
 test.cb(
     "collect() collects streamed elements into an array (object, flowing mode)",
@@ -59,7 +59,7 @@ test.cb(
         const source = new Readable({ objectMode: false });
 
         source
-            .pipe(collect())
+            .pipe(collect({ objectMode: false }))
             .on("data", collected => {
                 expect(collected).to.deep.equal(Buffer.from("abc"));
                 t.pass();

tests/compose.spec.ts

@@ -2,9 +2,9 @@ import * as test from "ava";
 import { expect } from "chai";
 import { sleep } from "../src/helpers";
 import { Readable, Writable } from "stream";
-import mhysa from "../src";
+import { strom } from "../src";
 import { performance } from "perf_hooks";
-const { compose, map } = mhysa({ objectMode: true });
+const { compose, map, fromArray } = strom({ 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(7);
+    t.plan(2);
     return new Promise(async (resolve, reject) => {
         interface Chunk {
             key: string;
@@ -140,19 +140,12 @@ test("compose() writable length should be less than highWaterMark when handing w
             { key: "e", mapped: [] },
         ];
 
-        for (const item of input) {
+        fromArray(input).pipe(composed);
-            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(7);
+    t.plan(2);
     const _rate = 100;
     const highWaterMark = 2;
     return new Promise(async (resolve, reject) => {
@@ -181,19 +174,14 @@ 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) => {
-            pendingReads--;
+            t.deepEqual(chunk.mapped, [1, 2]);
-            if (pendingReads === 0) {
-                resolve();
-            }
         });
 
+        composed.on("finish", () => resolve());
+
         const input = [
             { key: "a", mapped: [] },
             { key: "b", mapped: [] },
@@ -201,19 +189,7 @@ 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();
-            }
-        }
     });
 });
 
@@ -247,10 +223,6 @@ 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();
         });
 
@@ -271,7 +243,6 @@ test.cb(
         input.forEach(item => {
             composed.write(item);
         });
-        const start = performance.now();
     },
 );
 

tests/concat.spec.ts

@@ -1,8 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { concat, collect } = mhysa();
+const { concat, collect } = strom();
 
 test.cb(
     "concat() concatenates multiple readable streams (object, flowing mode)",
@@ -172,7 +172,7 @@ test.cb(
 test.cb("concat() concatenates empty list of readable streams", t => {
     t.plan(0);
     concat()
-        .pipe(collect())
+        .pipe(collect({ objectMode: false }))
         .on("data", _ => {
             t.fail();
         })

tests/defaultOptions.spec.ts

@@ -1,11 +1,11 @@
 import { Readable } from "stream";
 import test from "ava";
-import mhysa from "../src";
+import { strom } from "../src";
 
-const withDefaultOptions = mhysa({ objectMode: true });
+const withDefaultOptions = strom({ objectMode: true });
-const withoutOptions = mhysa();
+const withoutOptions = strom();
 
-test("Mhysa instances can have default options", t => {
+test("strom instances can have default options", t => {
     let batch = withDefaultOptions.batch();
     t.true(batch._readableState.objectMode);
     t.true(batch._writableState.objectMode);

tests/demux.spec.ts

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

@@ -2,8 +2,8 @@ import * as cp from "child_process";
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { duplex } = mhysa();
+const { duplex } = strom();
 
 test.cb(
     "duplex() combines a writable and readable stream into a ReadWrite stream",

tests/filter.spec.ts

@@ -1,8 +1,8 @@
 import test from "ava";
 import { expect } from "chai";
 import { Readable } from "stream";
-import mhysa from "../src";
+import { strom } from "../src";
-const { filter } = mhysa();
+const { filter } = strom();
 
 test.cb("filter() filters elements synchronously", t => {
     t.plan(2);

tests/flatMap.spec.ts

@@ -1,8 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { flatMap } = mhysa({ objectMode: true });
+const { flatMap } = strom({ objectMode: true });
 
 test.cb("flatMap() maps elements synchronously", t => {
     t.plan(6);

tests/fromArray.spec.ts

@@ -1,7 +1,7 @@
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { fromArray } = mhysa();
+const { fromArray } = strom();
 
 test.cb("fromArray() streams array elements in flowing mode", t => {
     t.plan(3);

tests/join.spec.ts

@@ -1,8 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { join } = mhysa();
+const { join } = strom();
 
 test.cb("join() joins chunks using the specified separator", t => {
     t.plan(9);

tests/last.spec.ts

@@ -1,8 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { last } = mhysa();
+const { last } = strom();
 
 test("last() resolves to the last chunk streamed by the given readable stream", async t => {
     const source = new Readable({ objectMode: true });

tests/map.spec.ts

@@ -1,8 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { map } = mhysa();
+const { map } = strom();
 
 test.cb("map() maps elements synchronously", t => {
     t.plan(3);

tests/merge.spec.ts

@@ -1,8 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { merge } = mhysa();
+const { merge } = strom();
 
 test.cb(
     "merge() merges multiple readable streams in chunk arrival order",

tests/parallelMap.spec.ts

@@ -2,9 +2,9 @@ import { Readable } from "stream";
 import { performance } from "perf_hooks";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
 import { sleep } from "../src/helpers";
-const { parallelMap } = mhysa({ objectMode: true });
+const { parallelMap } = strom({ objectMode: true });
 
 test.cb("parallelMap() parallel mapping", t => {
     t.plan(6);

tests/parse.spec.ts

@@ -1,8 +1,8 @@
 import { Readable, finished } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { parse } = mhysa();
+const { parse } = strom();
 
 test.cb("parse() parses the streamed elements as JSON", t => {
     t.plan(3);

tests/rate.spec.ts

@@ -2,8 +2,8 @@ import { Readable } from "stream";
 import { performance } from "perf_hooks";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { rate } = mhysa({ objectMode: true });
+const { rate } = strom({ objectMode: true });
 
 test.cb("rate() sends data at a rate of 150", t => {
     t.plan(5);

tests/reduce.spec.ts

@@ -1,8 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { reduce } = mhysa({ objectMode: true });
+const { reduce } = strom({ objectMode: true });
 
 test.cb("reduce() reduces elements synchronously", t => {
     t.plan(1);

tests/replace.spec.ts

@@ -1,8 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { replace } = mhysa();
+const { replace } = strom();
 
 test.cb(
     "replace() replaces occurrences of the given string in the streamed elements with the specified " +

tests/split.spec.ts

@@ -1,8 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { split } = mhysa();
+const { split } = strom();
 
 test.cb("split() splits chunks using the default separator (\\n)", t => {
     t.plan(5);

tests/stringify.spec.ts

@@ -1,8 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { stringify } = mhysa();
+const { stringify } = strom();
 
 test.cb("stringify() stringifies the streamed elements as JSON", t => {
     t.plan(4);

tests/unbatch.spec.ts

@@ -1,8 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { strom } from "../src";
-const { unbatch, batch } = mhysa({ objectMode: true });
+const { unbatch, batch } = strom({ objectMode: true });
 
 test.cb("unbatch() unbatches", t => {
     t.plan(3);

tests/utils/collected.spec.ts

@@ -0,0 +1,9 @@
+import test from "ava";
+import { collected } from "../../src/utils";
+import { strom } from "../../src";
+const { fromArray, collect } = strom({ 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.7.2":
+"@types/node@^12.12.15":
-  version "12.12.14"
+  version "12.12.15"
-  resolved "https://npm.dev.jogogo.co/@types%2fnode/-/node-12.12.14.tgz#1c1d6e3c75dba466e0326948d56e8bd72a1903d2"
+  resolved "https://npm.dev.jogogo.co/@types%2fnode/-/node-12.12.15.tgz#8dfb6ce22fedd469128137640a3aa8f17415422f"
-  integrity sha512-u/SJDyXwuihpwjXy7hOOghagLEV1KdAST6syfnOk6QZAMzZuWZqXy5aYYZbh8Jdpd4escVFP0MvftHNDb9pruA==
+  integrity sha512-Pv+vWicyFd07Hw/SmNnTUguqrHgDfMtjabvD9sQyxeqbpCEg8CmViLBaVPHtNsoBgZECrRf5/pgV6FJIBrGSjw==
 
 "@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 "0.0.1-beta.4"
+  version "2.0.0-alpha.1"
 
 micromatch@^4.0.2:
   version "4.0.2"