From 87c44de7999625b52f63ac8da3581fa5780acd35 Mon Sep 17 00:00:00 2001 From: Lewis Diamond Date: Mon, 3 Aug 2020 18:54:40 -0400 Subject: [PATCH] Default to an objectMode: true instance but allow creating other instances with other defaults --- README.md | 45 +++++++++++++++++++++++ package.json | 12 +++++-- samples/parallelMap.js | 15 ++++++++ samples/rate.js | 21 +++++++++++ src/functions/index.ts | 5 +++ src/functions/rate.ts | 29 ++++++++++++--- src/index.ts | 29 ++++++++++++++- tests/accumulator.spec.ts | 3 +- tests/batch.spec.ts | 3 +- tests/child.spec.ts | 3 +- tests/collect.spec.ts | 3 +- tests/compose.spec.ts | 3 +- tests/concat.spec.ts | 3 +- tests/defaultOptions.spec.ts | 25 ++++++++----- tests/demux.spec.ts | 3 +- tests/duplex.spec.ts | 3 +- tests/filter.spec.ts | 3 +- tests/flatMap.spec.ts | 3 +- tests/fromArray.spec.ts | 3 +- tests/join.spec.ts | 3 +- tests/last.spec.ts | 3 +- tests/map.spec.ts | 3 +- tests/merge.spec.ts | 3 +- tests/parallelMap.spec.ts | 3 +- tests/parse.spec.ts | 3 +- tests/rate.spec.ts | 67 +++++++++++++++++++++++++++-------- tests/reduce.spec.ts | 3 +- tests/replace.spec.ts | 3 +- tests/split.spec.ts | 3 +- tests/stringify.spec.ts | 3 +- tests/unbatch.spec.ts | 3 +- tests/utils/collected.spec.ts | 3 +- 32 files changed, 241 insertions(+), 76 deletions(-) create mode 100644 samples/parallelMap.js create mode 100644 samples/rate.js diff --git a/README.md b/README.md index 2896a2c..a9361d4 100644 --- a/README.md +++ b/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(); +``` diff --git a/package.json b/package.json index 2eea983..2e76033 100644 --- a/package.json +++ b/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", diff --git a/samples/parallelMap.js b/samples/parallelMap.js new file mode 100644 index 0000000..8d8d94c --- /dev/null +++ b/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); diff --git a/samples/rate.js b/samples/rate.js new file mode 100644 index 0000000..78b255c --- /dev/null +++ b/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(); diff --git a/src/functions/index.ts b/src/functions/index.ts index 81c065b..d8ec7ac 100644 --- a/src/functions/index.ts +++ b/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, }; } diff --git a/src/functions/rate.ts b/src/functions/rate.ts index 8f0f734..1ed4321 100644 --- a/src/functions/rate.ts +++ b/src/functions/rate.ts @@ -2,20 +2,41 @@ 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) { - await sleep(deltaMS); + if (behavior === Behavior.DROP) { + callback(undefined); + return; + } else { + await sleep(deltaMS); + } } total += 1; callback(undefined, data); diff --git a/src/index.ts b/src/index.ts index d6124ca..32ff0da 100644 --- a/src/index.ts +++ b/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(); diff --git a/tests/accumulator.spec.ts b/tests/accumulator.spec.ts index 467bf17..a95c76b 100644 --- a/tests/accumulator.spec.ts +++ b/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); diff --git a/tests/batch.spec.ts b/tests/batch.spec.ts index 39e4a00..829b6b4 100644 --- a/tests/batch.spec.ts +++ b/tests/batch.spec.ts @@ -1,8 +1,7 @@ import { Readable } from "stream"; import test from "ava"; import { expect } from "chai"; -import { strom } from "../src"; -const { batch, map, fromArray } = strom({ objectMode: true }); +import { batch, map, fromArray } from "../src"; test.cb("batch() batches chunks together", t => { t.plan(3); diff --git a/tests/child.spec.ts b/tests/child.spec.ts index 6e8fa26..7730790 100644 --- a/tests/child.spec.ts +++ b/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"; -const { child } = strom(); +import { child } from "../src"; test.cb( "child() allows easily writing to child process stdin and reading from its stdout", diff --git a/tests/collect.spec.ts b/tests/collect.spec.ts index 5af912d..bae4757 100644 --- a/tests/collect.spec.ts +++ b/tests/collect.spec.ts @@ -1,8 +1,7 @@ import { Readable } from "stream"; import test from "ava"; import { expect } from "chai"; -import { strom } from "../src"; -const { collect } = strom(); +import { collect } from "../src"; test.cb( "collect() collects streamed elements into an array (object, flowing mode)", diff --git a/tests/compose.spec.ts b/tests/compose.spec.ts index d6c8363..da198ac 100644 --- a/tests/compose.spec.ts +++ b/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); diff --git a/tests/concat.spec.ts b/tests/concat.spec.ts index 8f4bf7b..079eff7 100644 --- a/tests/concat.spec.ts +++ b/tests/concat.spec.ts @@ -1,8 +1,7 @@ import { Readable } from "stream"; import test from "ava"; import { expect } from "chai"; -import { strom } from "../src"; -const { concat, collect } = strom(); +import { concat, collect } from "../src"; test.cb( "concat() concatenates multiple readable streams (object, flowing mode)", diff --git a/tests/defaultOptions.spec.ts b/tests/defaultOptions.spec.ts index a762929..2775822 100644 --- a/tests/defaultOptions.spec.ts +++ b/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 withoutOptions = strom(); +const withDefaultOptions = strom({ objectMode: false }); test("strom instances can have default options", t => { let batch = withDefaultOptions.batch(); - t.true(batch._readableState.objectMode); - t.true(batch._writableState.objectMode); + t.false(batch._readableState.objectMode); + t.false(batch._writableState.objectMode); batch = withDefaultOptions.batch(3); - t.true(batch._readableState.objectMode); - t.true(batch._writableState.objectMode); + t.false(batch._readableState.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); }); diff --git a/tests/demux.spec.ts b/tests/demux.spec.ts index b3ca157..93c8a8c 100644 --- a/tests/demux.spec.ts +++ b/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; diff --git a/tests/duplex.spec.ts b/tests/duplex.spec.ts index b477482..e5fafd7 100644 --- a/tests/duplex.spec.ts +++ b/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"; -const { duplex } = strom(); +import { duplex } from "../src"; test.cb( "duplex() combines a writable and readable stream into a ReadWrite stream", diff --git a/tests/filter.spec.ts b/tests/filter.spec.ts index eaf6128..badfda7 100644 --- a/tests/filter.spec.ts +++ b/tests/filter.spec.ts @@ -1,8 +1,7 @@ import test from "ava"; import { expect } from "chai"; import { Readable } from "stream"; -import { strom } from "../src"; -const { filter } = strom(); +import { filter } from "../src"; test.cb("filter() filters elements synchronously", t => { t.plan(2); diff --git a/tests/flatMap.spec.ts b/tests/flatMap.spec.ts index 1697b15..84cebfb 100644 --- a/tests/flatMap.spec.ts +++ b/tests/flatMap.spec.ts @@ -1,8 +1,7 @@ import { Readable } from "stream"; import test from "ava"; import { expect } from "chai"; -import { strom } from "../src"; -const { flatMap } = strom({ objectMode: true }); +import { flatMap } from "../src"; test.cb("flatMap() maps elements synchronously", t => { t.plan(6); diff --git a/tests/fromArray.spec.ts b/tests/fromArray.spec.ts index 84d9f0b..3e4c93e 100644 --- a/tests/fromArray.spec.ts +++ b/tests/fromArray.spec.ts @@ -1,7 +1,6 @@ import test from "ava"; import { expect } from "chai"; -import { strom } from "../src"; -const { fromArray } = strom(); +import { fromArray } from "../src"; test.cb("fromArray() streams array elements in flowing mode", t => { t.plan(3); diff --git a/tests/join.spec.ts b/tests/join.spec.ts index 403f80f..6b0be52 100644 --- a/tests/join.spec.ts +++ b/tests/join.spec.ts @@ -1,8 +1,7 @@ import { Readable } from "stream"; import test from "ava"; import { expect } from "chai"; -import { strom } from "../src"; -const { join } = strom(); +import { join } from "../src"; test.cb("join() joins chunks using the specified separator", t => { t.plan(9); diff --git a/tests/last.spec.ts b/tests/last.spec.ts index a5ae2d2..033c9d8 100644 --- a/tests/last.spec.ts +++ b/tests/last.spec.ts @@ -1,8 +1,7 @@ import { Readable } from "stream"; import test from "ava"; import { expect } from "chai"; -import { strom } from "../src"; -const { last } = strom(); +import { last } from "../src"; test("last() resolves to the last chunk streamed by the given readable stream", async t => { const source = new Readable({ objectMode: true }); diff --git a/tests/map.spec.ts b/tests/map.spec.ts index 8898e54..a88da17 100644 --- a/tests/map.spec.ts +++ b/tests/map.spec.ts @@ -1,8 +1,7 @@ import { Readable } from "stream"; import test from "ava"; import { expect } from "chai"; -import { strom } from "../src"; -const { map } = strom(); +import { map } from "../src"; test.cb("map() maps elements synchronously", t => { t.plan(3); diff --git a/tests/merge.spec.ts b/tests/merge.spec.ts index 6d50605..dbbfd79 100644 --- a/tests/merge.spec.ts +++ b/tests/merge.spec.ts @@ -1,8 +1,7 @@ import { Readable } from "stream"; import test from "ava"; import { expect } from "chai"; -import { strom } from "../src"; -const { merge } = strom(); +import { merge } from "../src"; test.cb( "merge() merges multiple readable streams in chunk arrival order", diff --git a/tests/parallelMap.spec.ts b/tests/parallelMap.spec.ts index 9214464..dff719a 100644 --- a/tests/parallelMap.spec.ts +++ b/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); diff --git a/tests/parse.spec.ts b/tests/parse.spec.ts index 4eed5fa..269125e 100644 --- a/tests/parse.spec.ts +++ b/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"; -const { parse } = strom(); +import { parse } from "../src"; test.cb("parse() parses the streamed elements as JSON", t => { t.plan(3); diff --git a/tests/rate.spec.ts b/tests/rate.spec.ts index 31fe5fe..8bf20fd 100644 --- a/tests/rate.spec.ts +++ b/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 { strom } from "../src"; -const { rate } = strom({ objectMode: true }); +import { rate } from "../src"; +import { sleep } from "../src/helpers"; 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]); - expect(currentRate).lessThan(targetRate); + t.is(element, expectedElements[i]); + 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]); - expect(currentRate).lessThan(targetRate); + t.is(element, expectedElements[i]); + 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]); - expect(currentRate).lessThan(targetRate); + t.is(element, expectedElements[i]); + 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); +}); diff --git a/tests/reduce.spec.ts b/tests/reduce.spec.ts index 3cbef0a..8d504db 100644 --- a/tests/reduce.spec.ts +++ b/tests/reduce.spec.ts @@ -1,8 +1,7 @@ import { Readable } from "stream"; import test from "ava"; import { expect } from "chai"; -import { strom } from "../src"; -const { reduce } = strom({ objectMode: true }); +import { reduce } from "../src"; test.cb("reduce() reduces elements synchronously", t => { t.plan(1); diff --git a/tests/replace.spec.ts b/tests/replace.spec.ts index ffcba5b..5829f8e 100644 --- a/tests/replace.spec.ts +++ b/tests/replace.spec.ts @@ -1,8 +1,7 @@ import { Readable } from "stream"; import test from "ava"; import { expect } from "chai"; -import { strom } from "../src"; -const { replace } = strom(); +import { replace } from "../src"; test.cb( "replace() replaces occurrences of the given string in the streamed elements with the specified " + diff --git a/tests/split.spec.ts b/tests/split.spec.ts index ea77a1b..1819e2b 100644 --- a/tests/split.spec.ts +++ b/tests/split.spec.ts @@ -1,8 +1,7 @@ import { Readable } from "stream"; import test from "ava"; import { expect } from "chai"; -import { strom } from "../src"; -const { split } = strom(); +import { split } from "../src"; test.cb("split() splits chunks using the default separator (\\n)", t => { t.plan(5); diff --git a/tests/stringify.spec.ts b/tests/stringify.spec.ts index 8695654..7452e99 100644 --- a/tests/stringify.spec.ts +++ b/tests/stringify.spec.ts @@ -1,8 +1,7 @@ import { Readable } from "stream"; import test from "ava"; import { expect } from "chai"; -import { strom } from "../src"; -const { stringify } = strom(); +import { stringify } from "../src"; test.cb("stringify() stringifies the streamed elements as JSON", t => { t.plan(4); diff --git a/tests/unbatch.spec.ts b/tests/unbatch.spec.ts index 93c2c80..d48b1b9 100644 --- a/tests/unbatch.spec.ts +++ b/tests/unbatch.spec.ts @@ -1,8 +1,7 @@ import { Readable } from "stream"; import test from "ava"; import { expect } from "chai"; -import { strom } from "../src"; -const { unbatch, batch } = strom({ objectMode: true }); +import { unbatch, batch } from "../src"; test.cb("unbatch() unbatches", t => { t.plan(3); diff --git a/tests/utils/collected.spec.ts b/tests/utils/collected.spec.ts index 3aa1676..fe36926 100644 --- a/tests/utils/collected.spec.ts +++ b/tests/utils/collected.spec.ts @@ -1,7 +1,6 @@ import test from "ava"; import { collected } from "../../src/utils"; -import { strom } from "../../src"; -const { fromArray, collect } = strom({ objectMode: true }); +import { fromArray, collect } from "../../src"; test("collected returns a promise for the first data point", async t => { const data = collected(fromArray([1, 2, 3, 4]).pipe(collect()));