From 4aac05c9c04e3be8da8ee8096c71f3cbed63cf31 Mon Sep 17 00:00:00 2001 From: Lewis Diamond Date: Fri, 31 Jul 2020 23:29:28 -0400 Subject: [PATCH] Adding some partial documentation. Fixing batching timeout --- README.md | 95 ++++++++++++++++++++++++++++++++++-------- src/functions/batch.ts | 4 +- tests/batch.spec.ts | 2 +- 3 files changed, 81 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index 0fa244b..2896a2c 100644 --- a/README.md +++ b/README.md @@ -2,10 +2,13 @@ **Dependency-free stream utils for Node.js** -Released under the [MIT](https://git.lewis.id/strom/blob/master/LICENSE) license. +Released under the [MIT](LICENSE) license. ```sh -yarn add strom // Name TBD +yarn add stromjs +``` +```sh +npm add stromjs ``` ## fromArray(array) @@ -23,7 +26,7 @@ strom.fromArray(["a", "b"]) ## map(mapper, options) -Return a `ReadWrite` stream that maps streamed chunks +Returns a `ReadWrite` stream that maps streamed chunks | Param | Type | Description | | --- | --- | --- | @@ -41,7 +44,7 @@ strom.fromArray(["a", "b"]) ## flatMap(mapper, options) -Return a `ReadWrite` stream that flat maps streamed chunks +Returns a `ReadWrite` stream that flat maps streamed chunks | Param | Type | Description | | --- | --- | --- | @@ -59,7 +62,7 @@ strom.fromArray(["a", "AA"]) ## 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 | | --- | --- | --- | @@ -76,7 +79,7 @@ strom.fromArray(["a", "b", "c"]) ## 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 | @@ -97,7 +100,7 @@ strom.fromArray(["a", "b", "cc"]) ## 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 | | --- | --- | --- | @@ -115,7 +118,7 @@ strom.fromArray(["a,b", "c,d"]) ## 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 | | --- | --- | --- | @@ -132,7 +135,7 @@ strom.fromArray(["a", "b", "c"]) ## 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 | @@ -151,7 +154,7 @@ strom.fromArray(["a1", "b22", "c333"]) ## parse() -Return a `ReadWrite` stream that parses the streamed chunks as JSON +Returns a `ReadWrite` stream that parses the streamed chunks as JSON ```js strom.fromArray(['{ "a": "b" }']) @@ -162,7 +165,7 @@ strom.fromArray(['{ "a": "b" }']) ## stringify() -Return a `ReadWrite` stream that stringifies the streamed chunks to JSON +Returns a `ReadWrite` stream that stringifies the streamed chunks to JSON ```js strom.fromArray([{ a: "b" }]) @@ -173,7 +176,7 @@ strom.fromArray([{ a: "b" }]) ## 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 | | --- | --- | --- | @@ -189,7 +192,7 @@ strom.fromArray(["a", "b", "c"]) ## concat(streams) -Return a `Readable` stream of readable streams concatenated together +Returns a `Readable` stream of readable streams concatenated together | Param | Type | Description | | --- | --- | --- | @@ -210,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 | | --- | --- | --- | @@ -231,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 | @@ -249,7 +252,7 @@ strom.fromArray(["a", "b", "c"]) ## 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 | | --- | --- | --- | @@ -265,7 +268,7 @@ strom.fromArray(["a", "b", "c"]) ## 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 | @@ -280,3 +283,61 @@ 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 | diff --git a/src/functions/batch.ts b/src/functions/batch.ts index 1909c16..8b8a7a7 100644 --- a/src/functions/batch.ts +++ b/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); diff --git a/tests/batch.spec.ts b/tests/batch.spec.ts index 7e059fc..39e4a00 100644 --- a/tests/batch.spec.ts +++ b/tests/batch.spec.ts @@ -39,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)) + .pipe(batch(3, 500)) .on("data", (element: string[]) => { t.deepEqual(element, expectedElements[i]); i++;