diff --git a/README.md b/README.md
index d3b2f45..540de49 100644
--- a/README.md
+++ b/README.md
@@ -2,11 +2,14 @@
**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 stromjs
```
+```sh
+npm add stromjs
+```
## fromArray(array)
Convert an array into a `Readable` stream of its elements
@@ -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,106 @@ 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();
+```
diff --git a/package.json b/package.json
index 2eea983..061d33f 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
{
"name": "stromjs",
- "version": "2.0.0-alpha.1",
+ "version": "0.5.1",
"description": "Dependency-free streams utils for Node.js",
"keywords": [
"promise",
@@ -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/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/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 7e059fc..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);
@@ -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++;
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()));