Default to an objectMode: true instance but allow creating other instances with other defaults

README.md

@@ -341,3 +341,48 @@ Returns a `Transform` stream which maps incoming data through the async mapper w
 | `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

@@ -9,10 +9,18 @@
     ],
     "contributors": [
         {
-            "name": "Wenzil"
+            "name": "Sami Turcotte",
+            "url": "https://github.com/Wenzil"
         },
         {
-            "name": "Lewis Diamond"
+           "name": "Jerry Kurian",
+           "email": "jerrykurian@protonmail.com",
+           "url": "https://github.com/jkurian"
+        },
+        {
+            "name": "Lewis Diamond",
+            "email": "stromjs@lewisdiamond.com",
+            "url": "https://github.com/lewisdiamond"
         }
     ],
     "license": "MIT",

samples/parallelMap.js

@@ -0,0 +1,15 @@
+const strom = require("stromjs").strom();
+
+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/rate.js

@@ -0,0 +1,21 @@
+const strom = require("../dist/index.js");
+
+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();

src/functions/index.ts

@@ -263,5 +263,10 @@ 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,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,2 +1,29 @@
-export { strom } from "./functions";
+import { 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,10 +1,9 @@
 import test from "ava";
 import { expect } from "chai";
 import { Readable } from "stream";
-import { strom } from "../src";
+import { accumulator, accumulatorBy } 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,8 +1,7 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import { strom } from "../src";
+import { batch, map, fromArray } from "../src";
-const { batch, map, fromArray } = strom({ objectMode: true });
 
 test.cb("batch() batches chunks together", t => {
     t.plan(3);

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

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

tests/defaultOptions.spec.ts

@@ -1,21 +1,30 @@
 import { Readable } from "stream";
 import test from "ava";
-import { strom } from "../src";
+import { batch as _batch, instance as strom } from "../src";
 
-const withDefaultOptions = strom({ objectMode: true });
+const withDefaultOptions = strom({ objectMode: false });
-const withoutOptions = strom();
 
 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 { strom } from "../src";
+import { demux, map, fromArray } 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,8 +2,7 @@ import * as cp from "child_process";
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import { strom } from "../src";
+import { duplex } from "../src";
-const { duplex } = strom();
 
 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 { strom } from "../src";
+import { filter } from "../src";
-const { filter } = strom();
 
 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 { strom } from "../src";
+import { flatMap } from "../src";
-const { flatMap } = strom({ 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 { strom } from "../src";
+import { fromArray } from "../src";
-const { fromArray } = strom();
 
 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 { strom } from "../src";
+import { join } from "../src";
-const { join } = strom();
 
 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 { strom } from "../src";
+import { last } 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,8 +1,7 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import { strom } from "../src";
+import { map } from "../src";
-const { map } = strom();
 
 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 { strom } from "../src";
+import { merge } from "../src";
-const { merge } = strom();
 
 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 { strom } from "../src";
+import { parallelMap } 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,8 +1,7 @@
 import { Readable, finished } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import { strom } from "../src";
+import { parse } from "../src";
-const { parse } = strom();
 
 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 { strom } from "../src";
+import { sleep } from "../src/helpers";
-const { rate } = strom({ 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 { strom } from "../src";
+import { reduce } from "../src";
-const { reduce } = strom({ 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 { strom } from "../src";
+import { replace } 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,8 +1,7 @@
 import { Readable } from "stream";
 import test from "ava";
 import { expect } from "chai";
-import { strom } from "../src";
+import { split } from "../src";
-const { split } = strom();
 
 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 { strom } from "../src";
+import { stringify } from "../src";
-const { stringify } = strom();
 
 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 { strom } from "../src";
+import { unbatch, batch } from "../src";
-const { unbatch, batch } = strom({ 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 { strom } from "../../src";
+import { fromArray, collect } 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()));