README.md

@@ -2,13 +2,10 @@
 
 **Dependency-free stream utils for Node.js**
 
-<sub>Released under the [MIT](LICENSE) license.</sub>
+<sub>Released under the [MIT](https://git.lewis.id/strom/blob/master/LICENSE) license.</sub>
 
 ```sh
-yarn add stromjs
+yarn add strom  // Name TBD
-```
-```sh
-npm add stromjs
 ```
 
 ## fromArray(array)
@@ -26,7 +23,7 @@ strom.fromArray(["a", "b"])
 
 
 ## map(mapper, options)
-Returns a `ReadWrite` stream that maps streamed chunks
+Return a `ReadWrite` stream that maps streamed chunks
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -44,7 +41,7 @@ strom.fromArray(["a", "b"])
 
 
 ## flatMap(mapper, options)
-Returns a `ReadWrite` stream that flat maps streamed chunks
+Return a `ReadWrite` stream that flat maps streamed chunks
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -62,7 +59,7 @@ strom.fromArray(["a", "AA"])
 
 
 ## filter(predicate, options)
-Returns a `ReadWrite` stream that filters out streamed chunks for which the predicate does not hold
+Return a `ReadWrite` stream that filters out streamed chunks for which the predicate does not hold
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -79,7 +76,7 @@ strom.fromArray(["a", "b", "c"])
 
 
 ## reduce(iteratee, initialValue, options)
-Returns a `ReadWrite` stream that reduces streamed chunks down to a single value and yield that
+Return a `ReadWrite` stream that reduces streamed chunks down to a single value and yield that
 value
 
 | Param | Type | Description |
@@ -100,7 +97,7 @@ strom.fromArray(["a", "b", "cc"])
 
 
 ## split(separator)
-Returns a `ReadWrite` stream that splits streamed chunks using the given separator
+Return a `ReadWrite` stream that splits streamed chunks using the given separator
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -118,7 +115,7 @@ strom.fromArray(["a,b", "c,d"])
 
 
 ## join(separator)
-Returns a `ReadWrite` stream that joins streamed chunks using the given separator
+Return a `ReadWrite` stream that joins streamed chunks using the given separator
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -135,7 +132,7 @@ strom.fromArray(["a", "b", "c"])
 
 
 ## replace(searchValue, replaceValue)
-Returns a `ReadWrite` stream that replaces occurrences of the given string or regular expression  in
+Return 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 |
@@ -154,7 +151,7 @@ strom.fromArray(["a1", "b22", "c333"])
 
 
 ## parse()
-Returns a `ReadWrite` stream that parses the streamed chunks as JSON
+Return a `ReadWrite` stream that parses the streamed chunks as JSON
 
 ```js
 strom.fromArray(['{ "a": "b" }'])
@@ -165,7 +162,7 @@ strom.fromArray(['{ "a": "b" }'])
 
 
 ## stringify()
-Returns a `ReadWrite` stream that stringifies the streamed chunks to JSON
+Return a `ReadWrite` stream that stringifies the streamed chunks to JSON
 
 ```js
 strom.fromArray([{ a: "b" }])
@@ -176,7 +173,7 @@ strom.fromArray([{ a: "b" }])
 
 
 ## collect(options)
-Returns a `ReadWrite` stream that collects streamed chunks into an array or buffer
+Return a `ReadWrite` stream that collects streamed chunks into an array or buffer
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -192,7 +189,7 @@ strom.fromArray(["a", "b", "c"])
 
 
 ## concat(streams)
-Returns a `Readable` stream of readable streams concatenated together
+Return a `Readable` stream of readable streams concatenated together
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -213,7 +210,7 @@ source2.push(null);
 
 
 ## merge(streams)
-Returns a `Readable` stream of readable streams merged together in chunk arrival order
+Return a `Readable` stream of readable streams merged together in chunk arrival order
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -234,7 +231,7 @@ setTimeout(() => source2.push(null), 50);
 
 
 ## duplex(writable, readable)
-Returns a `Duplex` stream from a writable stream that is assumed to somehow, when written to,
+Return 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 |
@@ -252,7 +249,7 @@ strom.fromArray(["a", "b", "c"])
 
 
 ## child(childProcess)
-Returns a `Duplex` stream from a child process' stdin and stdout
+Return a `Duplex` stream from a child process' stdin and stdout
 
 | Param | Type | Description |
 | --- | --- | --- |
@@ -268,7 +265,7 @@ strom.fromArray(["a", "b", "c"])
 
 
 ## last(readable)
-Returns a `Promise` resolving to the last streamed chunk of the given readable stream, after it has 
+Return a `Promise` resolving to the last streamed chunk of the given readable stream, after it has 
 ended
 
 | Param | Type | Description |
@@ -283,106 +280,3 @@ let f = async () => {
 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,7 +1,7 @@
 {
-    "name": "stromjs",
+    "name": "strom",
-    "version": "0.5.1",
+    "version": "2.0.0",
-    "description": "Dependency-free streams utils for Node.js",
+    "description": "Streams utils for Node.js",
     "keywords": [
         "promise",
         "stream",
@@ -9,18 +9,15 @@
     ],
     "contributors": [
         {
-            "name": "Sami Turcotte",
+            "name": "Wenzil"
-            "url": "https://github.com/Wenzil"
         },
         {
-           "name": "Jerry Kurian",
+            "name": "jerry",
-           "email": "jerrykurian@protonmail.com",
+            "email": "jerry@jogogo.co"
-           "url": "https://github.com/jkurian"
         },
         {
-            "name": "Lewis Diamond",
+            "name": "lewis",
-            "email": "stromjs@lewisdiamond.com",
+            "email": "lewis@jogogo.co"
-            "url": "https://github.com/lewisdiamond"
         }
     ],
     "license": "MIT",
@@ -30,7 +27,7 @@
         "dist"
     ],
     "repository": {
-        "url": "git@git.lewis.id:strom/strom.git",
+        "url": "git@git.lewis.id:ldiamond/strom.git",
         "type": "git"
     },
     "scripts": {
@@ -47,7 +44,7 @@
         "@types/sinon": "^7.0.13",
         "ava": "^2.4.0",
         "chai": "^4.2.0",
-        "stromjs": "./",
+        "mhysa": "./",
         "prettier": "^1.14.3",
         "sinon": "^7.4.2",
         "ts-node": "^8.3.0",

samples/async.js

@@ -1,7 +1,6 @@
-const strom = require("stromjs");
+const strom = require("strom").strom();
 
-strom
+strom.fromArray(["a", "b", "c"])
-    .fromArray(["a", "b", "c"])
     .pipe(strom.map(s => Promise.resolve(s + s)))
     .pipe(strom.flatMap(s => Promise.resolve([s, s.toUpperCase()])))
     .pipe(strom.filter(s => Promise.resolve(s !== "bb")))

samples/child.js

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

samples/collect.js

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

samples/concat.js

@@ -1,5 +1,5 @@
 const { Readable } = require("stream");
-const strom = require("stromjs");
+const strom = require("strom").strom();
 
 const source1 = new Readable();
 const source2 = new Readable();

samples/duplex.js

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

samples/filter.js

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

samples/flatMap.js

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

samples/join.js

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

samples/last.js

@@ -1,4 +1,4 @@
-const strom = require("stromjs");
+const strom = require("strom").strom();
 
 let f = async () => {
     const source = strom.fromArray(["a", "b", "c"]);

samples/map.js

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

samples/merge.js

@@ -1,5 +1,5 @@
 const { Readable } = require("stream");
-const strom = require("stromjs");
+const strom = require("strom").strom();
 
 const source1 = new Readable({ read() {} });
 const source2 = new Readable({ read() {} });

samples/parallelMap.js

@@ -1,15 +0,0 @@
-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,6 +1,5 @@
-const strom = require("stromjs");
+const strom = require("strom").strom();
 
-strom
+strom.fromArray(['{ "a": "b" }'])
-    .fromArray(['{ "a": "b" }'])
     .pipe(strom.parse())
     .on("data", object => console.log(object));

samples/rate.js

@@ -1,21 +0,0 @@
-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,7 +1,6 @@
-const strom = require("stromjs");
+const strom = require("strom").strom();
 
-strom
+strom.fromArray(["a", "b", "cc"])
-    .fromArray(["a", "b", "cc"])
     .pipe(strom.reduce((acc, s) => ({ ...acc, [s]: s.length }), {}))
     .pipe(strom.stringify())
     .pipe(process.stdout);

samples/replace.js

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

samples/split.js

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

samples/stringify.js

@@ -1,6 +1,5 @@
-const strom = require("stromjs");
+const strom = require("strom").strom();
 
-strom
+strom.fromArray([{ a: "b" }])
-    .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 = 0,
+    maxBatchAge: number = 500,
     options: TransformOptions = {},
 ): Transform {
     let buffer: any[] = [];
@@ -23,7 +23,7 @@ export function batch(
             buffer.push(chunk);
             if (buffer.length === batchSize) {
                 sendChunk(this);
-            } else if (maxBatchAge) {
+            } else {
                 if (timer === null) {
                     timer = setInterval(() => {
                         sendChunk(this);

src/functions/index.ts

@@ -263,10 +263,5 @@ export function strom(defaultOptions: TransformOptions = { objectMode: true }) {
          * @param options Writable stream options
          */
         demux: withDefaultOptions(2, demux),
-
-        /**
-         * Create a new strom instance overriding the defaults
-         */
-        instance: strom,
     };
 }

src/functions/rate.ts

@@ -2,41 +2,20 @@ 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 & RateOptions,
+    options?: TransformOptions,
 ): Transform {
     const deltaMS = ((1 / targetRate) * 1000) / period; // Skip a full period
     let total = 0;
-    const window = options?.window || Infinity;
+    const start = performance.now();
-    const behavior = options?.behavior || Behavior.BUFFER;
-    let start = performance.now();
     return new Transform({
         ...options,
         async transform(data, encoding, callback) {
-            const now = performance.now();
+            const currentRate = (total / (performance.now() - start)) * 1000;
-            if (now - start >= window) {
-                start = now - window;
-            }
-            const currentRate = (total / (now - start)) * 1000;
             if (targetRate && currentRate > targetRate) {
-                if (behavior === Behavior.DROP) {
+                await sleep(deltaMS);
-                    callback(undefined);
-                    return;
-                } else {
-                    await sleep(deltaMS);
-                }
             }
             total += 1;
             callback(undefined, data);

src/index.ts

@@ -1,29 +1,2 @@
-import { strom } from "./functions";
+export { strom } from "./functions";
 export * from "./utils";
-export const {
-    fromArray,
-    map,
-    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,9 +1,10 @@
 import test from "ava";
 import { expect } from "chai";
 import { Readable } from "stream";
-import { accumulator, accumulatorBy } from "../src";
+import { strom } from "../src";
 import { FlushStrategy } from "../src/functions/accumulator";
 import { performance } from "perf_hooks";
+const { accumulator, accumulatorBy } = strom({ objectMode: true });
 
 test.cb("accumulator() rolling", t => {
     t.plan(3);

tests/batch.spec.ts

@@ -1,7 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import { batch, map, fromArray } from "../src";
+import { strom } from "../src";
+const { batch, map, fromArray } = strom({ objectMode: true });
 
 test.cb("batch() batches chunks together", t => {
     t.plan(3);
@@ -38,7 +39,7 @@ test.cb("batch() yields a batch after the timeout", t => {
     const expectedElements = [["a", "b"], ["c"], ["d"]];
     let i = 0;
     source
-        .pipe(batch(3, 500))
+        .pipe(batch(3))
         .on("data", (element: string[]) => {
             t.deepEqual(element, expectedElements[i]);
             i++;

tests/child.spec.ts

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

tests/collect.spec.ts

@@ -1,7 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import { collect } from "../src";
+import { strom } from "../src";
+const { collect } = strom();
 
 test.cb(
     "collect() collects streamed elements into an array (object, flowing mode)",

tests/compose.spec.ts

@@ -2,8 +2,9 @@ import * as test from "ava";
 import { expect } from "chai";
 import { sleep } from "../src/helpers";
 import { Readable, Writable } from "stream";
-import { compose, map, fromArray } from "../src";
+import { strom } from "../src";
 import { performance } from "perf_hooks";
+const { compose, map, fromArray } = strom({ objectMode: true });
 
 test.cb("compose() chains two streams together in the correct order", t => {
     t.plan(3);

tests/concat.spec.ts

@@ -1,7 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import { concat, collect } from "../src";
+import { strom } from "../src";
+const { concat, collect } = strom();
 
 test.cb(
     "concat() concatenates multiple readable streams (object, flowing mode)",

tests/defaultOptions.spec.ts

@@ -1,30 +1,21 @@
 import { Readable } from "stream";
 import test from "ava";
-import { batch as _batch, instance as strom } from "../src";
+import { strom } from "../src";
 
-const withDefaultOptions = strom({ objectMode: false });
+const withDefaultOptions = strom({ objectMode: true });
+const withoutOptions = strom();
 
 test("strom instances can have default options", t => {
     let batch = withDefaultOptions.batch();
-    t.false(batch._readableState.objectMode);
+    t.true(batch._readableState.objectMode);
-    t.false(batch._writableState.objectMode);
+    t.true(batch._writableState.objectMode);
     batch = withDefaultOptions.batch(3);
-    t.false(batch._readableState.objectMode);
+    t.true(batch._readableState.objectMode);
-    t.false(batch._writableState.objectMode);
+    t.true(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,10 +1,11 @@
 import test from "ava";
 import { expect } from "chai";
-import { demux, map, fromArray } from "../src";
+import { strom } from "../src";
 import { Writable, Readable } from "stream";
 import * as sinon from "sinon";
 import { sleep } from "../src/helpers";
 import { performance } from "perf_hooks";
+const { demux, map, fromArray } = strom({ objectMode: true });
 
 interface Test {
     key: string;

tests/duplex.spec.ts

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

tests/filter.spec.ts

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

tests/flatMap.spec.ts

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

tests/fromArray.spec.ts

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

tests/join.spec.ts

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

tests/last.spec.ts

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

tests/merge.spec.ts

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

tests/parallelMap.spec.ts

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

tests/parse.spec.ts

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

tests/rate.spec.ts

@@ -1,11 +1,12 @@
 import { Readable } from "stream";
 import { performance } from "perf_hooks";
 import test from "ava";
-import { rate } from "../src";
+import { expect } from "chai";
-import { sleep } from "../src/helpers";
+import { strom } from "../src";
+const { rate } = strom({ objectMode: true });
 
 test.cb("rate() sends data at a rate of 150", t => {
-    t.plan(15);
+    t.plan(5);
     const targetRate = 150;
     const source = new Readable({ objectMode: true });
     const expectedElements = ["a", "b", "c", "d", "e"];
@@ -14,10 +15,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;
-            t.is(element, expectedElements[i]);
+            expect(element).to.deep.equal(expectedElements[i]);
-            t.true(currentRate <= targetRate);
+            expect(currentRate).lessThan(targetRate);
             t.pass();
             i++;
         })
@@ -33,7 +34,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(15);
+    t.plan(5);
     const targetRate = 50;
     const source = new Readable({ objectMode: true });
     const expectedElements = ["a", "b", "c", "d", "e"];
@@ -42,10 +43,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;
-            t.is(element, expectedElements[i]);
+            expect(element).to.deep.equal(expectedElements[i]);
-            t.true(currentRate <= targetRate);
+            expect(currentRate).lessThan(targetRate);
             t.pass();
             i++;
         })
@@ -61,7 +62,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(15);
+    t.plan(5);
     const targetRate = 1;
     const source = new Readable({ objectMode: true });
     const expectedElements = ["a", "b", "c", "d", "e"];
@@ -70,10 +71,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;
-            t.is(element, expectedElements[i]);
+            expect(element).to.deep.equal(expectedElements[i]);
-            t.true(currentRate <= targetRate);
+            expect(currentRate).lessThan(targetRate);
             t.pass();
             i++;
         })
@@ -87,41 +88,3 @@ 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,7 +1,8 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import { reduce } from "../src";
+import { strom } from "../src";
+const { reduce } = strom({ objectMode: true });
 
 test.cb("reduce() reduces elements synchronously", t => {
     t.plan(1);

tests/replace.spec.ts

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

tests/split.spec.ts

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

tests/stringify.spec.ts

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

tests/unbatch.spec.ts

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

tests/utils/collected.spec.ts

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

yarn.lock

@@ -1934,6 +1934,9 @@ 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"
@@ -2716,9 +2719,6 @@ 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"