set repo to github

* Demux handles arrays by key

* Rename construct

* version

* Minor changes to types

README.md

@@ -1,14 +1,15 @@
-# 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](LICENSE) license.</sub>
 
 ```sh
-yarn add mhysa
+yarn add stromjs
+```
+```sh
+npm add stromjs
 ```
-
-<sub>Tested with Node.js versions 8+</sub>
 
 ## fromArray(array)
 Convert an array into a `Readable` stream of its elements
@@ -18,14 +19,14 @@ 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
 ```
 
 
 ## map(mapper, options)
-Return a `ReadWrite` stream that maps streamed chunks
+Returns a `ReadWrite` stream that maps streamed chunks
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -35,15 +36,15 @@ 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
 ```
 
 
 ## flatMap(mapper, options)
-Return a `ReadWrite` stream that flat maps streamed chunks
+Returns a `ReadWrite` stream that flat maps streamed chunks
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -53,15 +54,15 @@ 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
 ```
 
 
 ## filter(predicate, options)
-Return a `ReadWrite` stream that filters out streamed chunks for which the predicate does not hold
+Returns a `ReadWrite` stream that filters out streamed chunks for which the predicate does not hold
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -70,15 +71,15 @@ 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
 ```
 
 
 ## reduce(iteratee, initialValue, options)
-Return a `ReadWrite` stream that reduces streamed chunks down to a single value and yield that
+Returns a `ReadWrite` stream that reduces streamed chunks down to a single value and yield that
 value
 
 | Param | Type | Description |
@@ -90,16 +91,16 @@ 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
 ```
 
 
 ## split(separator)
-Return a `ReadWrite` stream that splits streamed chunks using the given separator
+Returns a `ReadWrite` stream that splits streamed chunks using the given separator
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -108,16 +109,16 @@ 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
 ```
 
 
 ## join(separator)
-Return a `ReadWrite` stream that joins streamed chunks using the given separator
+Returns a `ReadWrite` stream that joins streamed chunks using the given separator
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -126,15 +127,15 @@ 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
 ```
 
 
 ## replace(searchValue, replaceValue)
-Return a `ReadWrite` stream that replaces occurrences of the given string or regular expression  in
+Returns a `ReadWrite` stream that replaces occurrences of the given string or regular expression  in
 the streamed chunks with the specified replacement string
 
 | Param | Type | Description |
@@ -145,37 +146,37 @@ 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
 ```
 
 
 ## parse()
-Return a `ReadWrite` stream that parses the streamed chunks as JSON
+Returns 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
 ```
 
 
 ## stringify()
-Return a `ReadWrite` stream that stringifies the streamed chunks to JSON
+Returns 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
 ```
 
 
 ## collect(options)
-Return a `ReadWrite` stream that collects streamed chunks into an array or buffer
+Returns a `ReadWrite` stream that collects streamed chunks into an array or buffer
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -183,15 +184,15 @@ 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
 ```
 
 
 ## concat(streams)
-Return a `Readable` stream of readable streams concatenated together
+Returns a `Readable` stream of readable streams concatenated together
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -200,7 +201,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 ");
@@ -212,7 +213,7 @@ source2.push(null);
 
 
 ## merge(streams)
-Return a `Readable` stream of readable streams merged together in chunk arrival order
+Returns a `Readable` stream of readable streams merged together in chunk arrival order
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -221,7 +222,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);
@@ -233,7 +234,7 @@ setTimeout(() => source2.push(null), 50);
 
 
 ## duplex(writable, readable)
-Return a `Duplex` stream from a writable stream that is assumed to somehow, when written to,
+Returns a `Duplex` stream from a writable stream that is assumed to somehow, when written to,
 cause the given readable stream to yield chunks
 
 | Param | Type | Description |
@@ -243,15 +244,15 @@ 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
 ```
 
 
 ## child(childProcess)
-Return a `Duplex` stream from a child process' stdin and stdout
+Returns a `Duplex` stream from a child process' stdin and stdout
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -259,15 +260,15 @@ 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
 ```
 
 
 ## last(readable)
-Return a `Promise` resolving to the last streamed chunk of the given readable stream, after it has 
+Returns a `Promise` resolving to the last streamed chunk of the given readable stream, after it has 
 ended
 
 | Param | Type | Description |
@@ -276,9 +277,112 @@ 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
 ```
+
+## accumulator(flushStrategy, iteratee, options)
+TO BE DOCUMENTED
+
+## batch(batchSize, maxBatchAge, options)
+Returns a `Transform` stream which produces all incoming data in batches of size `batchSize`.
+
+| Param | Type | Description |
+| --- | --- | --- |
+| `batchSize` | `number` | Size of the batches to be produced |
+| `maxBatchAge` | `number`  | Maximum number of milliseconds a message will be queued for. E.g. a batch will be produced before reaching `batchSize` if the first message queued is `maxBatchAge` ms old or more |
+| `options` | `TransformOptions` | Options passed down to the Transform object |
+
+```js
+strom.fromArray(["a", "b", "c", "d"])
+    .pipe(strom.batch(3, 500))
+    .pipe(process.stdout);
+// ["a","b","c"]
+// ["d"] //After 500ms
+```
+
+## compose(streams, errorCb, options)
+
+Returns a `Transform` stream which consists of all `streams` but behaves as a single stream. The returned stream can be piped into and from transparently.
+
+| Param | Type | Description |
+| --- | --- | --- |
+| `streams` | `Array` | Streams to be composed |
+| `errorCb` | `(err: Error) => void`  | Function called when an error occurs in any of the streams |
+| `options` | `TransformOptions` | Options passed down to the Transform object |
+
+```js
+const composed = strom.compose([
+    strom.split(),
+    strom.map(data => data.trim()),
+    strom.filter(str => !!str),
+    strom.parse(),
+    strom.flatMap(data => data),
+    strom.stringify(),
+]);
+
+const data = ["[1,2,3] \n  [4,5,6] ", "\n [7,8,9] \n\n"];
+
+strom.fromArray(data).pipe(composed).pipe(process.stdout);
+// 123456789
+```
+
+## demux(pipelineConstructor, demuxBy, options)
+TO BE DOCUMENTED
+
+## parallelMap(mapper, parallel, sleepTime, options)
+Returns a `Transform` stream which maps incoming data through the async mapper with the given parallelism. 
+
+| Param | Type | Description | Default |
+| --- | --- | --- | --- |
+| `mapper` | `async (chunk: T, encoding: string) => R` | Mapper function, mapping each (chunk, encoding) to a new chunk (non-async will not be parallelized) | -- |
+| `parallel` | `number`  | Number of concurrent executions of the mapper allowed | 10 |
+| `sleepTime` | `number` | Number of milliseconds to wait before testing if more messages can be processed | 1 |
+
+```js
+function sleep(time) {
+    return time > 0 ? new Promise(resolve => setTimeout(resolve, time)) : null;
+}
+
+strom
+    .fromArray([1, 2, 3, 4, 6, 8])
+    .pipe(
+        strom.parallelMap(async d => {
+            await sleep(10000 - d * 1000);
+            return `${d}`;
+        }, 3),
+    )
+    .pipe(process.stdout);
+    
+// 321864
+```
+
+## rate()
+
+
+```js
+const strom = require("stromjs").strom();
+
+function sleep(time) {
+    return time > 0 ? new Promise(resolve => setTimeout(resolve, time)) : null;
+}
+
+const rate = strom.rate(2, 1, { behavior: 1 });
+rate.pipe(strom.map(x => console.log(x)));
+async function produce() {
+    rate.write(1);
+    await sleep(500);
+    rate.write(2);
+    await sleep(500);
+    rate.write(3);
+    rate.write(4);
+    rate.write(5);
+    await sleep(500);
+    rate.write(6);
+}
+
+produce();
+```

package.json

@@ -1,24 +1,26 @@
 {
-    "name": "@jogogo/mhysa",
+    "name": "stromjs",
-    "version": "2.0.0-alpha.3",
+    "version": "0.5.1",
-    "description": "Streams and event emitter utils for Node.js",
+    "description": "Dependency-free streams utils for Node.js",
     "keywords": [
         "promise",
         "stream",
-        "event emitter",
         "utils"
     ],
-    "author": {
-        "name": "Wenzil"
-    },
     "contributors": [
         {
-            "name": "jerry",
+            "name": "Sami Turcotte",
-            "email": "jerry@jogogo.co"
+            "url": "https://github.com/Wenzil"
         },
         {
-            "name": "lewis",
+           "name": "Jerry Kurian",
-            "email": "lewis@jogogo.co"
+           "email": "jerrykurian@protonmail.com",
+           "url": "https://github.com/jkurian"
+        },
+        {
+            "name": "Lewis Diamond",
+            "email": "stromjs@lewisdiamond.com",
+            "url": "https://github.com/lewisdiamond"
         }
     ],
     "license": "MIT",
@@ -27,18 +29,16 @@
     "files": [
         "dist"
     ],
-    "publishConfig": {
-        "registry": "https://npm.dev.jogogo.co/"
-    },
     "repository": {
-        "url": "git@github.com:Jogogoplay/mhysa.git",
+        "url": "https://github.com/lewisdiamond/stromjs",
         "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": {
@@ -47,7 +47,7 @@
         "@types/sinon": "^7.0.13",
         "ava": "^2.4.0",
         "chai": "^4.2.0",
-        "mhysa": "./",
+        "stromjs": "./",
         "prettier": "^1.14.3",
         "sinon": "^7.4.2",
         "ts-node": "^8.3.0",

samples/async.js

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

samples/child.js

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

samples/collect.js

@@ -1,5 +1,6 @@
-const Mhysa = require("mhysa");
+const strom = require("stromjs");
 
-Mhysa.fromArray(["a", "b", "c"])
+strom
-    .pipe(Mhysa.collect({ objectMode: true }))
+    .fromArray(["a", "b", "c"])
+    .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("stromjs");
 
 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,7 @@
-const Mhysa = require("mhysa");
+const strom = require("stromjs");
 const catProcess = require("child_process").exec("grep -o ab");
 
-Mhysa.fromArray(["a", "b", "c"])
+strom
-    .pipe(Mhysa.duplex(catProcess.stdin, catProcess.stdout))
+    .fromArray(["a", "b", "c"])
+    .pipe(strom.duplex(catProcess.stdin, catProcess.stdout))
     .pipe(process.stdout);

samples/filter.js

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

samples/flatMap.js

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

samples/join.js

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

samples/last.js

@@ -1,7 +1,7 @@
-const Mhysa = require("mhysa");
+const strom = require("stromjs");
 
 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,7 @@
-const Mhysa = require("mhysa");
+const strom = require("stromjs");
 
-Mhysa.fromArray(["a", "b"])
+strom
-    .pipe(Mhysa.map(s => s.toUpperCase()))
+    .fromArray(["a", "b"])
-    .pipe(Mhysa.join(","))
+    .pipe(strom.map(s => s.toUpperCase()))
+    .pipe(strom.join(","))
     .pipe(process.stdout);

samples/merge.js

@@ -1,9 +1,9 @@
 const { Readable } = require("stream");
-const Mhysa = require("mhysa");
+const strom = require("stromjs");
 
 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/parallelMap.js

@@ -0,0 +1,15 @@
+const strom = require("stromjs");
+
+function sleep(time) {
+    return time > 0 ? new Promise(resolve => setTimeout(resolve, time)) : null;
+}
+
+strom
+    .fromArray([1, 2, 3, 4, 6, 8])
+    .pipe(
+        strom.parallelMap(async d => {
+            await sleep(10000 - d * 1000);
+            return `${d}`;
+        }, 3),
+    )
+    .pipe(process.stdout);

samples/parse.js

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

samples/rate.js

@@ -0,0 +1,21 @@
+const strom = require("stromjs");
+
+function sleep(time) {
+    return time > 0 ? new Promise(resolve => setTimeout(resolve, time)) : null;
+}
+
+const rate = strom.rate(2, 1, { behavior: 1 });
+rate.pipe(strom.map(x => console.log(x)));
+async function produce() {
+    rate.write(1);
+    await sleep(500);
+    rate.write(2);
+    await sleep(500);
+    rate.write(3);
+    rate.write(4);
+    rate.write(5);
+    await sleep(500);
+    rate.write(6);
+}
+
+produce();

samples/reduce.js

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

samples/replace.js

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

samples/split.js

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

samples/stringify.js

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

src/functions/batch.ts

@@ -2,7 +2,7 @@ import { Transform, TransformOptions } from "stream";
 
 export function batch(
     batchSize: number = 1000,
-    maxBatchAge: number = 500,
+    maxBatchAge: number = 0,
     options: TransformOptions = {},
 ): Transform {
     let buffer: any[] = [];
@@ -23,7 +23,7 @@ export function batch(
             buffer.push(chunk);
             if (buffer.length === batchSize) {
                 sendChunk(this);
-            } else {
+            } else if (maxBatchAge) {
                 if (timer === null) {
                     timer = setInterval(() => {
                         sendChunk(this);

src/functions/demux.ts

@@ -6,36 +6,45 @@ type DemuxStreams = NodeJS.WritableStream | NodeJS.ReadWriteStream;
 
 export interface DemuxOptions extends DuplexOptions {
     remultiplex?: boolean;
-    purgeIdleInterval?: number;
-    maxIdleTime?: number;
 }
 
 export function demux(
-    construct: (destKey?: string, chunk?: any) => DemuxStreams,
+    pipelineConstructor: (
+        destKey?: string,
+        chunk?: any,
+    ) => DemuxStreams | DemuxStreams[],
     demuxBy: string | ((chunk: any) => string),
     options?: DemuxOptions,
 ): Duplex {
-    return new Demux(construct, demuxBy, options);
+    return new Demux(pipelineConstructor, demuxBy, options);
 }
 
 class Demux extends Duplex {
     private streamsByKey: {
-        [key: string]: { stream: DemuxStreams; lastWrite: number };
+        [key: string]: DemuxStreams[];
     };
     private demuxer: (chunk: any) => string;
-    private construct: (destKey?: string, chunk?: any) => DemuxStreams;
+    private pipelineConstructor: (
+        destKey?: string,
+        chunk?: any,
+    ) => DemuxStreams[];
     private remultiplex: boolean;
     private transform: Transform;
-    private maxIdleTime: number;
     constructor(
-        construct: (destKey?: string) => DemuxStreams,
+        pipelineConstructor: (
+            destKey?: string,
+            chunk?: any,
+        ) => DemuxStreams | DemuxStreams[],
         demuxBy: string | ((chunk: any) => string),
         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 = {};
@@ -46,74 +55,77 @@ class Demux extends Duplex {
                 cb(null);
             },
         });
-        this.maxIdleTime = options.maxIdleTime || 600000;
-        const purgeIdleInterval = options.purgeIdleInterval || 600000;
-        setInterval(() => {
-            this._destroyIdle();
-        }, purgeIdleInterval);
 
         this.on("unpipe", () => this._flush());
     }
 
-    private _destroyIdle() {
-        for (let key in this.streamsByKey) {
-            const curTime = Date.now();
-            const pipeline = this.streamsByKey[key];
-
-            if (curTime - pipeline.lastWrite > this.maxIdleTime) {
-                delete this.streamsByKey[key];
-            }
-        }
-    }
     // 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 newPipeline = await this.construct(destKey, chunk);
+            const newPipelines = this.pipelineConstructor(destKey, chunk);
-            this.streamsByKey[destKey] = {
+            this.streamsByKey[destKey] = newPipelines;
-                stream: newPipeline,
+
-                lastWrite: Date.now(),
+            newPipelines.forEach(newPipeline => {
-            };
                 if (this.remultiplex && isReadable(newPipeline)) {
-                (newPipeline as NodeJS.ReadWriteStream).pipe(this.transform);
+                    (newPipeline as NodeJS.ReadWriteStream).pipe(
+                        this.transform,
+                    );
                 } else if (this.remultiplex) {
                     console.error(
                         `Pipeline construct for ${destKey} does not implement readable interface`,
                     );
                 }
-        } else {
-            this.streamsByKey[destKey].lastWrite = Date.now();
-        }
-
-        if (!this.streamsByKey[destKey].stream.write(chunk, encoding)) {
-            this.streamsByKey[destKey].stream.once("drain", () => {
-                cb();
             });
-        } 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 _flush() {
-        const pipelines = Object.values(this.streamsByKey);
+        const pipelines: DemuxStreams[] = Array.prototype.concat.apply(
-        let totalEnded = 0;
+            [],
-        pipelines.forEach(({ stream }) => {
+            Object.values(this.streamsByKey),
-            stream.once("end", () => {
+        );
-                totalEnded++;
+        const flushPromises: Array<Promise<void>> = [];
-                if (pipelines.length === totalEnded) {
+        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");
-                }
         });
-        });
-        pipelines.forEach(({ stream }) => stream.end());
     }
 
     public _destroy(error: any, cb: (error?: any) => void) {
-        const pipelines = Object.values(this.streamsByKey);
+        const pipelines: DemuxStreams[] = [].concat.apply(
-        pipelines.forEach(({ stream }) => (stream as any).destroy());
+            [],
+            Object.values(this.streamsByKey),
+        );
+        pipelines.forEach(p => (p as any).destroy());
         cb(error);
     }
 }

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,
@@ -256,15 +256,17 @@ export default function mhysa(defaultOptions?: TransformOptions) {
 
         /**
          * Composes multiple streams together. Writing occurs on first stream, piping occurs from last stream.
-         * @param {Function} construct Constructor for new output source. Should return a Writable or ReadWrite stream.
+         * @param construct Constructor for new output source. Should return a Writable or ReadWrite stream.
-         * @param {String | Function} demuxBy
+         * @param demuxBy
-         * @param {string} demuxBy.key? Key to fetch value from source chunks to demultiplex source.
+         * @param demuxBy.key? Key to fetch value from source chunks to demultiplex source.
-         * @param {Function} demuxBy.keyBy? Function to fetch value from source chunks to demultiplex source.
+         * @param demuxBy.keyBy? Function to fetch value from source chunks to demultiplex source.
-         * @param {Object} options Demux stream options
+         * @param options Writable stream options
-         * @param {boolean} options.remultiplex? If demux should be remultiplexed into a single destination
-         * @param {number} options.purgeIdleInterval? Interval at which a purge for idle pipelines will occur
-         * @param {number} options.maxIdleTime? Min time a demuxed pipeline must be idle for to be purged
          */
         demux: withDefaultOptions(2, demux),
+
+        /**
+         * Create a new strom instance overriding the defaults
+         */
+        instance: strom,
     };
 }

src/functions/rate.ts

@@ -2,21 +2,42 @@ import { Transform, TransformOptions } from "stream";
 import { performance } from "perf_hooks";
 import { sleep } from "../helpers";
 
+export enum Behavior {
+    BUFFER = 0,
+    DROP = 1,
+}
+
+export interface RateOptions {
+    window?: number;
+    behavior?: Behavior;
+}
+
 export function rate(
     targetRate: number = 50,
     period: number = 1,
-    options?: TransformOptions,
+    options?: TransformOptions & RateOptions,
 ): Transform {
     const deltaMS = ((1 / targetRate) * 1000) / period; // Skip a full period
     let total = 0;
-    const start = performance.now();
+    const window = options?.window || Infinity;
+    const behavior = options?.behavior || Behavior.BUFFER;
+    let start = performance.now();
     return new Transform({
         ...options,
         async transform(data, encoding, callback) {
-            const currentRate = (total / (performance.now() - start)) * 1000;
+            const now = performance.now();
+            if (now - start >= window) {
+                start = now - window;
+            }
+            const currentRate = (total / (now - start)) * 1000;
             if (targetRate && currentRate > targetRate) {
+                if (behavior === Behavior.DROP) {
+                    callback(undefined);
+                    return;
+                } else {
                     await sleep(deltaMS);
                 }
+            }
             total += 1;
             callback(undefined, data);
         },

src/index.ts

@@ -1,6 +1,29 @@
-import mhysa from "./functions";
+import { strom } from "./functions";
-import * as _utils from "./utils";
+export * from "./utils";
-export default mhysa;
+export const {
-
+    fromArray,
-// @TODO fix this with proper import export
+    map,
-export const utils = { ..._utils };
+    flatMap,
+    filter,
+    reduce,
+    split,
+    join,
+    replace,
+    parse,
+    stringify,
+    collect,
+    concat,
+    merge,
+    duplex,
+    child,
+    last,
+    batch,
+    unbatch,
+    rate,
+    parallelMap,
+    accumulator,
+    accumulatorBy,
+    compose,
+    demux,
+    instance,
+} = strom();

tests/accumulator.spec.ts

@@ -1,10 +1,9 @@
 import test from "ava";
 import { expect } from "chai";
 import { Readable } from "stream";
-import mhysa from "../src";
+import { accumulator, accumulatorBy } from "../src";
 import { FlushStrategy } from "../src/functions/accumulator";
 import { performance } from "perf_hooks";
-const { accumulator, accumulatorBy } = mhysa({ objectMode: true });
 
 test.cb("accumulator() rolling", t => {
     t.plan(3);
@@ -14,8 +13,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 +93,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 +190,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 +243,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 +257,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 +306,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 +357,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 +504,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 +518,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,7 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { batch, map, fromArray } from "../src";
-const { batch, map, fromArray } = mhysa({ objectMode: true });
 
 test.cb("batch() batches chunks together", t => {
     t.plan(3);
@@ -39,7 +38,7 @@ test.cb("batch() yields a batch after the timeout", t => {
     const expectedElements = [["a", "b"], ["c"], ["d"]];
     let i = 0;
     source
-        .pipe(batch(3))
+        .pipe(batch(3, 500))
         .on("data", (element: string[]) => {
             t.deepEqual(element, expectedElements[i]);
             i++;

tests/child.spec.ts

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

tests/collect.spec.ts

@@ -1,8 +1,7 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { collect } from "../src";
-const { collect } = mhysa();
 
 test.cb(
     "collect() collects streamed elements into an array (object, flowing mode)",
@@ -59,7 +58,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,8 @@ import * as test from "ava";
 import { expect } from "chai";
 import { sleep } from "../src/helpers";
 import { Readable, Writable } from "stream";
-import mhysa from "../src";
+import { compose, map, fromArray } from "../src";
 import { performance } from "perf_hooks";
-const { compose, map, fromArray } = mhysa({ objectMode: true });
 
 test.cb("compose() chains two streams together in the correct order", t => {
     t.plan(3);
@@ -114,13 +113,9 @@ test("compose() writable length should be less than highWaterMark when handing w
             return chunk;
         });
 
-        const composed = compose(
+        const composed = compose([first, second], undefined, {
-            [first, second],
-            undefined,
-            {
             highWaterMark: 2,
-            },
+        });
-        );
         composed.on("error", err => {
             reject();
         });
@@ -168,13 +163,9 @@ test("compose() should emit drain event ~rate * highWaterMark ms for every write
             return chunk;
         });
 
-        const composed = compose(
+        const composed = compose([first, second], undefined, {
-            [first, second],
-            undefined,
-            {
             highWaterMark,
-            },
+        });
-        );
         composed.on("error", err => {
             reject();
         });
@@ -221,13 +212,9 @@ test.cb(
             return chunk;
         });
 
-        const composed = compose(
+        const composed = compose([first, second], undefined, {
-            [first, second],
-            undefined,
-            {
             highWaterMark: 5,
-            },
+        });
-        );
 
         composed.on("error", err => {
             t.end(err);
@@ -284,13 +271,9 @@ test.cb(
             { highWaterMark: 1 },
         );
 
-        const composed = compose(
+        const composed = compose([first, second], undefined, {
-            [first, second],
-            undefined,
-            {
             highWaterMark: 5,
-            },
+        });
-        );
         composed.on("error", err => {
             t.end(err);
         });
@@ -355,13 +338,9 @@ test.cb(
             { highWaterMark: 2 },
         );
 
-        const composed = compose(
+        const composed = compose([first, second], undefined, {
-            [first, second],
-            undefined,
-            {
             highWaterMark: 5,
-            },
+        });
-        );
         composed.on("error", err => {
             t.end(err);
         });
@@ -414,13 +393,9 @@ test.cb(
             return chunk;
         });
 
-        const composed = compose(
+        const composed = compose([first, second], undefined, {
-            [first, second],
-            undefined,
-            {
             highWaterMark: 6,
-            },
+        });
-        );
 
         composed.on("error", err => {
             t.end(err);
@@ -470,12 +445,9 @@ test.cb("compose() should be 'destroyable'", t => {
         return chunk;
     });
 
-    const composed = compose(
+    const composed = compose([first, second], (err: any) => {
-        [first, second],
-        (err: any) => {
         t.pass();
-        },
+    });
-    );
 
     const fakeSource = new Readable({
         objectMode: true,
@@ -531,13 +503,9 @@ test.cb("compose() `finish` and `end` propagates", t => {
         return chunk;
     });
 
-    const composed = compose(
+    const composed = compose([first, second], undefined, {
-        [first, second],
-        undefined,
-        {
         highWaterMark: 3,
-        },
+    });
-    );
 
     const fakeSource = new Readable({
         objectMode: true,

tests/concat.spec.ts

@@ -1,8 +1,7 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { concat, collect } from "../src";
-const { concat, collect } = mhysa();
 
 test.cb(
     "concat() concatenates multiple readable streams (object, flowing mode)",
@@ -172,7 +171,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,21 +1,30 @@
 import { Readable } from "stream";
 import test from "ava";
-import mhysa from "../src";
+import { batch as _batch, instance as strom } from "../src";
 
-const withDefaultOptions = mhysa({ objectMode: true });
+const withDefaultOptions = strom({ objectMode: false });
-const withoutOptions = mhysa();
 
-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.false(batch._readableState.objectMode);
-    t.true(batch._writableState.objectMode);
+    t.false(batch._writableState.objectMode);
     batch = withDefaultOptions.batch(3);
-    t.true(batch._readableState.objectMode);
+    t.false(batch._readableState.objectMode);
-    t.true(batch._writableState.objectMode);
+    t.false(batch._writableState.objectMode);
     batch = withDefaultOptions.batch(3, 1);
+    t.false(batch._readableState.objectMode);
+    t.false(batch._writableState.objectMode);
+    batch = withDefaultOptions.batch(3, 1, { objectMode: true });
     t.true(batch._readableState.objectMode);
     t.true(batch._writableState.objectMode);
-    batch = withDefaultOptions.batch(3, 1, { objectMode: false });
+
+    batch = _batch(3);
+    t.true(batch._readableState.objectMode);
+    t.true(batch._writableState.objectMode);
+    batch = _batch(3, 1);
+    t.true(batch._readableState.objectMode);
+    t.true(batch._writableState.objectMode);
+    batch = _batch(3, 1, { objectMode: false });
     t.false(batch._readableState.objectMode);
     t.false(batch._writableState.objectMode);
 });

tests/demux.spec.ts

@@ -1,11 +1,10 @@
 import test from "ava";
 import { expect } from "chai";
-import mhysa from "../src";
+import { demux, map, fromArray } from "../src";
 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, fromArray } = mhysa({ objectMode: true });
 
 interface Test {
     key: string;
@@ -862,39 +861,3 @@ test.cb("demux() should be 'destroyable'", t => {
     fakeSource.push(input[3]);
     fakeSource.push(input[4]);
 });
-
-test.cb("Should delete idle pipelines", t => {
-    t.plan(5);
-    const input = [
-        { key: "a", visited: [] },
-        { key: "b", visited: [] },
-        { key: "b", visited: [] },
-        { key: "a", visited: [] },
-        { key: "b", visited: [] },
-        { key: "c", visited: [] },
-    ];
-    const construct = sinon.spy((destKey: string) => {
-        const dest = map((chunk: Test) => {
-            chunk.visited.push(1);
-            return chunk;
-        });
-        t.pass();
-
-        return dest;
-    });
-
-    const demuxed = demux(construct, "key", {
-        maxIdleTime: 110,
-        purgeIdleInterval: 110,
-    });
-
-    demuxed.on("data", data => {
-        if (data.key === "c") t.end();
-    });
-
-    for (let i = 0; i < input.length; i++) {
-        setTimeout(() => {
-            demuxed.write(input[i]);
-        }, i * 100);
-    }
-});

tests/duplex.spec.ts

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

tests/filter.spec.ts

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

tests/flatMap.spec.ts

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

tests/fromArray.spec.ts

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

tests/join.spec.ts

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

tests/last.spec.ts

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

tests/merge.spec.ts

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

tests/parallelMap.spec.ts

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

tests/parse.spec.ts

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

tests/rate.spec.ts

@@ -1,12 +1,11 @@
 import { Readable } from "stream";
 import { performance } from "perf_hooks";
 import test from "ava";
-import { expect } from "chai";
+import { rate } from "../src";
-import mhysa from "../src";
+import { sleep } from "../src/helpers";
-const { rate } = mhysa({ objectMode: true });
 
 test.cb("rate() sends data at a rate of 150", t => {
-    t.plan(5);
+    t.plan(15);
     const targetRate = 150;
     const source = new Readable({ objectMode: true });
     const expectedElements = ["a", "b", "c", "d", "e"];
@@ -15,10 +14,10 @@ test.cb("rate() sends data at a rate of 150", t => {
 
     source
         .pipe(rate(targetRate))
-        .on("data", (element: string[]) => {
+        .on("data", (element: string) => {
             const currentRate = (i / (performance.now() - start)) * 1000;
-            expect(element).to.deep.equal(expectedElements[i]);
+            t.is(element, expectedElements[i]);
-            expect(currentRate).lessThan(targetRate);
+            t.true(currentRate <= targetRate);
             t.pass();
             i++;
         })
@@ -34,7 +33,7 @@ test.cb("rate() sends data at a rate of 150", t => {
 });
 
 test.cb("rate() sends data at a rate of 50", t => {
-    t.plan(5);
+    t.plan(15);
     const targetRate = 50;
     const source = new Readable({ objectMode: true });
     const expectedElements = ["a", "b", "c", "d", "e"];
@@ -43,10 +42,10 @@ test.cb("rate() sends data at a rate of 50", t => {
 
     source
         .pipe(rate(targetRate))
-        .on("data", (element: string[]) => {
+        .on("data", (element: string) => {
             const currentRate = (i / (performance.now() - start)) * 1000;
-            expect(element).to.deep.equal(expectedElements[i]);
+            t.is(element, expectedElements[i]);
-            expect(currentRate).lessThan(targetRate);
+            t.true(currentRate <= targetRate);
             t.pass();
             i++;
         })
@@ -62,7 +61,7 @@ test.cb("rate() sends data at a rate of 50", t => {
 });
 
 test.cb("rate() sends data at a rate of 1", t => {
-    t.plan(5);
+    t.plan(15);
     const targetRate = 1;
     const source = new Readable({ objectMode: true });
     const expectedElements = ["a", "b", "c", "d", "e"];
@@ -71,10 +70,10 @@ test.cb("rate() sends data at a rate of 1", t => {
 
     source
         .pipe(rate(targetRate))
-        .on("data", (element: string[]) => {
+        .on("data", (element: string) => {
             const currentRate = (i / (performance.now() - start)) * 1000;
-            expect(element).to.deep.equal(expectedElements[i]);
+            t.is(element, expectedElements[i]);
-            expect(currentRate).lessThan(targetRate);
+            t.true(currentRate <= targetRate);
             t.pass();
             i++;
         })
@@ -88,3 +87,41 @@ test.cb("rate() sends data at a rate of 1", t => {
     source.push("e");
     source.push(null);
 });
+
+test("rate() sends data at a rate of 1 and drops extra messages", async t => {
+    t.plan(9);
+    const targetRate = 1;
+    const source = new Readable({
+        objectMode: true,
+        read: () => {
+            return;
+        },
+    });
+    const expectedElements = ["a", "b", "e"];
+    const start = performance.now();
+    let i = 0;
+
+    let plan = 0;
+    source
+        .pipe(rate(targetRate, 1, { behavior: 1 }))
+        .on("data", (element: string) => {
+            const currentRate = (i / (performance.now() - start)) * 1000;
+            t.is(element, expectedElements[i]);
+            t.true(currentRate <= targetRate);
+            plan++;
+            t.pass();
+            i++;
+        })
+        .on("error", t.fail)
+        .on("end", t.fail);
+
+    source.push("a");
+    await sleep(1000);
+    source.push("b");
+    source.push("c");
+    source.push("d");
+    await sleep(1000);
+    source.push("e");
+    await sleep(1000);
+    source.push(null);
+});

tests/reduce.spec.ts

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

tests/replace.spec.ts

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

tests/split.spec.ts

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

tests/stringify.spec.ts

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

tests/unbatch.spec.ts

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

tests/utils/collected.spec.ts

@@ -1,7 +1,6 @@
 import test from "ava";
 import { collected } from "../../src/utils";
-import mhysa from "../../src";
+import { fromArray, collect } 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()));

yarn.lock

@@ -1934,9 +1934,6 @@ merge2@^1.2.3, merge2@^1.3.0:
   resolved "https://npm.dev.jogogo.co/merge2/-/merge2-1.3.0.tgz#5b366ee83b2f1582c48f87e47cf1a9352103ca81"
   integrity sha512-2j4DAdlBOkiSZIsaXk4mTE3sRS02yBHAtfy127xRV3bQUFqXkjHCHLW6Scv7DwNRbIWNHH8zpnz9zMaKXIdvYw==
 
-mhysa@./:
-  version "2.0.0-alpha.1"
-
 micromatch@^4.0.2:
   version "4.0.2"
   resolved "https://npm.dev.jogogo.co/micromatch/-/micromatch-4.0.2.tgz#4fcb0999bf9fbc2fcbdd212f6d629b9a56c39259"
@@ -2719,6 +2716,9 @@ strip-json-comments@~2.0.1:
   resolved "https://registry.yarnpkg.com/strip-json-comments/-/strip-json-comments-2.0.1.tgz#3c531942e908c2697c0ec344858c286c7ca0a60a"
   integrity sha1-PFMZQukIwml8DsNEhYwobHygpgo=
 
+stromjs@./:
+  version "0.5.1"
+
 supertap@^1.0.0:
   version "1.0.0"
   resolved "https://registry.yarnpkg.com/supertap/-/supertap-1.0.0.tgz#bd9751c7fafd68c68cf8222a29892206a119fa9e"