Compare commits

...

17 Commits

Author SHA1 Message Date
Lewis Diamond
e005b45207 set repo to github 2021-02-05 20:13:45 -05:00
Lewis Diamond
2fb7942d9c Update package to use itself in dev, update samples 2020-08-05 19:08:13 -04:00
Lewis Diamond
fc5b96f725 Merge branch 'dev' into master 2020-08-05 10:42:33 -04:00
Lewis Diamond
491ec146a6 Update readme 2020-08-05 10:41:23 -04:00
Lewis Diamond
38bb853181 Set version to 0.5.0, diverge from Mhysa's versioning 2020-08-03 18:56:44 -04:00
Lewis Diamond
87c44de799 Default to an objectMode: true instance but allow creating other instances with other defaults 2020-08-03 18:54:40 -04:00
Lewis Diamond
7f1309f45f update pkg name 2020-08-01 22:09:00 -04:00
Lewis Diamond
4aac05c9c0 Adding some partial documentation. Fixing batching timeout 2020-07-31 23:32:26 -04:00
Lewis Diamond
caaabf4427 Make the package name stromjs because strom is taken 2020-07-31 22:17:57 -04:00
Lewis Diamond
1c02cb5aea No need to create an array to call concat 2020-07-12 18:52:21 -04:00
Lewis Diamond
12cbddf7e0 Update README 2020-07-09 12:51:08 -04:00
Lewis Diamond
2e5a6dbcc8 update samples 2020-07-04 11:06:41 -04:00
Lewis Diamond
e6fb55eb1a ObjectMode is now true by default 2020-07-04 10:58:08 -04:00
Lewis Diamond
74ce415118 Update tests make export a bit nicer 2020-07-04 10:43:52 -04:00
Lewis Diamond
1675b18d5b add prepare script 2020-07-04 00:15:11 -04:00
Lewis Diamond
ca6f36d19a Quick rename 2020-07-04 00:08:31 -04:00
Jerry Kurian
db783805ad
Arrays of pipelines by key (#7)
* Demux handles arrays by key

* Rename construct

* version

* Minor changes to types
2020-05-12 09:44:11 -04:00
51 changed files with 589 additions and 315 deletions

206
README.md
View File

@ -1,14 +1,15 @@
# Mhysa # Strom
**Dependency-free stream utils for Node.js** **Dependency-free stream utils for Node.js**
<sub>Released under the [MIT](https://github.com/Wenzil/Mhysa/blob/master/LICENSE) license.</sub> <sub>Released under the [MIT](LICENSE) license.</sub>
```sh ```sh
yarn add mhysa yarn add stromjs
```
```sh
npm add stromjs
``` ```
<sub>Tested with Node.js versions 8+</sub>
## fromArray(array) ## fromArray(array)
Convert an array into a `Readable` stream of its elements Convert an array into a `Readable` stream of its elements
@ -18,14 +19,14 @@ Convert an array into a `Readable` stream of its elements
| `array` | `T[]` | Array of elements to stream | | `array` | `T[]` | Array of elements to stream |
```js ```js
Mhysa.fromArray(["a", "b"]) strom.fromArray(["a", "b"])
.pipe(process.stdout); .pipe(process.stdout);
// ab is printed out // ab is printed out
``` ```
## map(mapper, options) ## map(mapper, options)
Return a `ReadWrite` stream that maps streamed chunks Returns a `ReadWrite` stream that maps streamed chunks
| Param | Type | Description | | Param | Type | Description |
| --- | --- | --- | | --- | --- | --- |
@ -35,15 +36,15 @@ Return a `ReadWrite` stream that maps streamed chunks
| `options.writableObjectMode` | `boolean` | Whether this stream should behave as a writable stream of objects | | `options.writableObjectMode` | `boolean` | Whether this stream should behave as a writable stream of objects |
```js ```js
Mhysa.fromArray(["a", "b"]) strom.fromArray(["a", "b"])
.pipe(Mhysa.map(s => s.toUpperCase())) .pipe(strom.map(s => s.toUpperCase()))
.pipe(process.stdout); .pipe(process.stdout);
// AB is printed out // AB is printed out
``` ```
## flatMap(mapper, options) ## flatMap(mapper, options)
Return a `ReadWrite` stream that flat maps streamed chunks Returns a `ReadWrite` stream that flat maps streamed chunks
| Param | Type | Description | | Param | Type | Description |
| --- | --- | --- | | --- | --- | --- |
@ -53,15 +54,15 @@ Return a `ReadWrite` stream that flat maps streamed chunks
| `options.writableObjectMode` | `boolean` | Whether this stream should behave as a writable stream of objects | | `options.writableObjectMode` | `boolean` | Whether this stream should behave as a writable stream of objects |
```js ```js
Mhysa.fromArray(["a", "AA"]) strom.fromArray(["a", "AA"])
.pipe(Mhysa.flatMap(s => new Array(s.length).fill(s))) .pipe(strom.flatMap(s => new Array(s.length).fill(s)))
.pipe(process.stdout); .pipe(process.stdout);
// aAAAA is printed out // aAAAA is printed out
``` ```
## filter(predicate, options) ## 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 | | Param | Type | Description |
| --- | --- | --- | | --- | --- | --- |
@ -70,15 +71,15 @@ Return a `ReadWrite` stream that filters out streamed chunks for which the predi
| `options.objectMode` | `boolean` | `boolean` | Whether this stream should behave as a stream of objects | | `options.objectMode` | `boolean` | `boolean` | Whether this stream should behave as a stream of objects |
```js ```js
Mhysa.fromArray(["a", "b", "c"]) strom.fromArray(["a", "b", "c"])
.pipe(Mhysa.filter(s => s !== "b")) .pipe(strom.filter(s => s !== "b"))
.pipe(process.stdout); .pipe(process.stdout);
// ac is printed out // ac is printed out
``` ```
## reduce(iteratee, initialValue, options) ## 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 value
| Param | Type | Description | | Param | Type | Description |
@ -90,16 +91,16 @@ value
| `options.writableObjectMode` | `boolean` | Whether this stream should behave as a writable stream of objects | | `options.writableObjectMode` | `boolean` | Whether this stream should behave as a writable stream of objects |
```js ```js
Mhysa.fromArray(["a", "b", "cc"]) strom.fromArray(["a", "b", "cc"])
.pipe(Mhysa.reduce((acc, s) => ({ ...acc, [s]: s.length }), {})) .pipe(strom.reduce((acc, s) => ({ ...acc, [s]: s.length }), {}))
.pipe(Mhysa.stringify()) .pipe(strom.stringify())
.pipe(process.stdout); .pipe(process.stdout);
// {"a":1,"b":1","c":2} is printed out // {"a":1,"b":1","c":2} is printed out
``` ```
## split(separator) ## 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 | | Param | Type | Description |
| --- | --- | --- | | --- | --- | --- |
@ -108,16 +109,16 @@ Return a `ReadWrite` stream that splits streamed chunks using the given separato
| `options.encoding` | `string` | Character encoding to use for decoding chunks. Defaults to utf8 | `options.encoding` | `string` | Character encoding to use for decoding chunks. Defaults to utf8
```js ```js
Mhysa.fromArray(["a,b", "c,d"]) strom.fromArray(["a,b", "c,d"])
.pipe(Mhysa.split(",")) .pipe(strom.split(","))
.pipe(Mhysa.join("|")) .pipe(strom.join("|"))
.pipe(process.stdout); .pipe(process.stdout);
// a|bc|d is printed out // a|bc|d is printed out
``` ```
## join(separator) ## 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 | | Param | Type | Description |
| --- | --- | --- | | --- | --- | --- |
@ -126,15 +127,15 @@ Return a `ReadWrite` stream that joins streamed chunks using the given separator
| `options.encoding` | `string` | Character encoding to use for decoding chunks. Defaults to utf8 | `options.encoding` | `string` | Character encoding to use for decoding chunks. Defaults to utf8
```js ```js
Mhysa.fromArray(["a", "b", "c"]) strom.fromArray(["a", "b", "c"])
.pipe(Mhysa.join(",")) .pipe(strom.join(","))
.pipe(process.stdout); .pipe(process.stdout);
// a,b,c is printed out // a,b,c is printed out
``` ```
## replace(searchValue, replaceValue) ## 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 the streamed chunks with the specified replacement string
| Param | Type | Description | | Param | Type | Description |
@ -145,37 +146,37 @@ the streamed chunks with the specified replacement string
| `options.encoding` | `string` | Character encoding to use for decoding chunks. Defaults to utf8 | `options.encoding` | `string` | Character encoding to use for decoding chunks. Defaults to utf8
```js ```js
Mhysa.fromArray(["a1", "b22", "c333"]) strom.fromArray(["a1", "b22", "c333"])
.pipe(Mhysa.replace(/b\d+/, "B")) .pipe(strom.replace(/b\d+/, "B"))
.pipe(process.stdout); .pipe(process.stdout);
// a1Bc333 is printed out // a1Bc333 is printed out
``` ```
## parse() ## parse()
Return a `ReadWrite` stream that parses the streamed chunks as JSON Returns a `ReadWrite` stream that parses the streamed chunks as JSON
```js ```js
Mhysa.fromArray(['{ "a": "b" }']) strom.fromArray(['{ "a": "b" }'])
.pipe(Mhysa.parse()) .pipe(strom.parse())
.once("data", object => console.log(object)); .once("data", object => console.log(object));
// { a: 'b' } is printed out // { a: 'b' } is printed out
``` ```
## stringify() ## stringify()
Return a `ReadWrite` stream that stringifies the streamed chunks to JSON Returns a `ReadWrite` stream that stringifies the streamed chunks to JSON
```js ```js
Mhysa.fromArray([{ a: "b" }]) strom.fromArray([{ a: "b" }])
.pipe(Mhysa.stringify()) .pipe(strom.stringify())
.pipe(process.stdout); .pipe(process.stdout);
// {"a":"b"} is printed out // {"a":"b"} is printed out
``` ```
## collect(options) ## 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 | | Param | Type | Description |
| --- | --- | --- | | --- | --- | --- |
@ -183,15 +184,15 @@ Return a `ReadWrite` stream that collects streamed chunks into an array or buffe
| `options.objectMode` | `boolean` | Whether this stream should behave as a stream of objects | | `options.objectMode` | `boolean` | Whether this stream should behave as a stream of objects |
```js ```js
Mhysa.fromArray(["a", "b", "c"]) strom.fromArray(["a", "b", "c"])
.pipe(Mhysa.collect({ objectMode: true })) .pipe(strom.collect({ objectMode: true }))
.once("data", object => console.log(object)); .once("data", object => console.log(object));
// [ 'a', 'b', 'c' ] is printed out // [ 'a', 'b', 'c' ] is printed out
``` ```
## concat(streams) ## concat(streams)
Return a `Readable` stream of readable streams concatenated together Returns a `Readable` stream of readable streams concatenated together
| Param | Type | Description | | Param | Type | Description |
| --- | --- | --- | | --- | --- | --- |
@ -200,7 +201,7 @@ Return a `Readable` stream of readable streams concatenated together
```js ```js
const source1 = new Readable(); const source1 = new Readable();
const source2 = new Readable(); const source2 = new Readable();
Mhysa.concat(source1, source2).pipe(process.stdout) strom.concat(source1, source2).pipe(process.stdout)
source1.push("a1 "); source1.push("a1 ");
source2.push("c3 "); source2.push("c3 ");
source1.push("b2 "); source1.push("b2 ");
@ -212,7 +213,7 @@ source2.push(null);
## merge(streams) ## 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 | | Param | Type | Description |
| --- | --- | --- | | --- | --- | --- |
@ -221,7 +222,7 @@ Return a `Readable` stream of readable streams merged together in chunk arrival
```js ```js
const source1 = new Readable({ read() {} }); const source1 = new Readable({ read() {} });
const source2 = new Readable({ read() {} }); const source2 = new Readable({ read() {} });
Mhysa.merge(source1, source2).pipe(process.stdout); strom.merge(source1, source2).pipe(process.stdout);
source1.push("a1 "); source1.push("a1 ");
setTimeout(() => source2.push("c3 "), 10); setTimeout(() => source2.push("c3 "), 10);
setTimeout(() => source1.push("b2 "), 20); setTimeout(() => source1.push("b2 "), 20);
@ -233,7 +234,7 @@ setTimeout(() => source2.push(null), 50);
## duplex(writable, readable) ## 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 cause the given readable stream to yield chunks
| Param | Type | Description | | Param | Type | Description |
@ -243,15 +244,15 @@ cause the given readable stream to yield chunks
```js ```js
const catProcess = require("child_process").exec("grep -o ab"); const catProcess = require("child_process").exec("grep -o ab");
Mhysa.fromArray(["a", "b", "c"]) strom.fromArray(["a", "b", "c"])
.pipe(Mhysa.duplex(catProcess.stdin, catProcess.stdout)) .pipe(strom.duplex(catProcess.stdin, catProcess.stdout))
.pipe(process.stdout); .pipe(process.stdout);
// ab is printed out // ab is printed out
``` ```
## child(childProcess) ## 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 | | Param | Type | Description |
| --- | --- | --- | | --- | --- | --- |
@ -259,15 +260,15 @@ Return a `Duplex` stream from a child process' stdin and stdout
```js ```js
const catProcess = require("child_process").exec("grep -o ab"); const catProcess = require("child_process").exec("grep -o ab");
Mhysa.fromArray(["a", "b", "c"]) strom.fromArray(["a", "b", "c"])
.pipe(Mhysa.child(catProcess)) .pipe(strom.child(catProcess))
.pipe(process.stdout); .pipe(process.stdout);
// ab is printed out // ab is printed out
``` ```
## last(readable) ## 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 ended
| Param | Type | Description | | Param | Type | Description |
@ -276,9 +277,112 @@ ended
```js ```js
let f = async () => { let f = async () => {
const source = Mhysa.fromArray(["a", "b", "c"]); const source = strom.fromArray(["a", "b", "c"]);
console.log(await Mhysa.last(source)); console.log(await strom.last(source));
}; };
f(); f();
// c is printed out // 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();
```

View File

@ -1,24 +1,26 @@
{ {
"name": "@jogogo/mhysa", "name": "stromjs",
"version": "2.0.0-alpha.3", "version": "0.5.1",
"description": "Streams and event emitter utils for Node.js", "description": "Dependency-free streams utils for Node.js",
"keywords": [ "keywords": [
"promise", "promise",
"stream", "stream",
"event emitter",
"utils" "utils"
], ],
"author": {
"name": "Wenzil"
},
"contributors": [ "contributors": [
{ {
"name": "jerry", "name": "Sami Turcotte",
"email": "jerry@jogogo.co" "url": "https://github.com/Wenzil"
}, },
{ {
"name": "lewis", "name": "Jerry Kurian",
"email": "lewis@jogogo.co" "email": "jerrykurian@protonmail.com",
"url": "https://github.com/jkurian"
},
{
"name": "Lewis Diamond",
"email": "stromjs@lewisdiamond.com",
"url": "https://github.com/lewisdiamond"
} }
], ],
"license": "MIT", "license": "MIT",
@ -27,18 +29,16 @@
"files": [ "files": [
"dist" "dist"
], ],
"publishConfig": {
"registry": "https://npm.dev.jogogo.co/"
},
"repository": { "repository": {
"url": "git@github.com:Jogogoplay/mhysa.git", "url": "https://github.com/lewisdiamond/stromjs",
"type": "git" "type": "git"
}, },
"scripts": { "scripts": {
"test": "ava", "test": "ava",
"lint": "tslint -p tsconfig.json", "lint": "tslint -p tsconfig.json",
"validate:tslint": "tslint-config-prettier-check ./tslint.json", "validate:tslint": "tslint-config-prettier-check ./tslint.json",
"prepublishOnly": "yarn lint && yarn test && yarn tsc -d" "prepublishOnly": "yarn lint && yarn test && yarn tsc -d",
"prepare": "tsc"
}, },
"dependencies": {}, "dependencies": {},
"devDependencies": { "devDependencies": {
@ -47,7 +47,7 @@
"@types/sinon": "^7.0.13", "@types/sinon": "^7.0.13",
"ava": "^2.4.0", "ava": "^2.4.0",
"chai": "^4.2.0", "chai": "^4.2.0",
"mhysa": "./", "stromjs": "./",
"prettier": "^1.14.3", "prettier": "^1.14.3",
"sinon": "^7.4.2", "sinon": "^7.4.2",
"ts-node": "^8.3.0", "ts-node": "^8.3.0",

View File

@ -1,8 +1,9 @@
const Mhysa = require("mhysa"); const strom = require("stromjs");
Mhysa.fromArray(["a", "b", "c"]) strom
.pipe(Mhysa.map(s => Promise.resolve(s + s))) .fromArray(["a", "b", "c"])
.pipe(Mhysa.flatMap(s => Promise.resolve([s, s.toUpperCase()]))) .pipe(strom.map(s => Promise.resolve(s + s)))
.pipe(Mhysa.filter(s => Promise.resolve(s !== "bb"))) .pipe(strom.flatMap(s => Promise.resolve([s, s.toUpperCase()])))
.pipe(Mhysa.join(",")) .pipe(strom.filter(s => Promise.resolve(s !== "bb")))
.pipe(strom.join(","))
.pipe(process.stdout); .pipe(process.stdout);

View File

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

View File

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

View File

@ -1,9 +1,9 @@
const { Readable } = require("stream"); const { Readable } = require("stream");
const Mhysa = require("mhysa"); const strom = require("stromjs");
const source1 = new Readable(); const source1 = new Readable();
const source2 = new Readable(); const source2 = new Readable();
Mhysa.concat(source1, source2).pipe(process.stdout); strom.concat(source1, source2).pipe(process.stdout);
source1.push("a1 "); source1.push("a1 ");
source2.push("c3 "); source2.push("c3 ");
source1.push("b2 "); source1.push("b2 ");

View File

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

View File

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

View File

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

View File

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

View File

@ -1,7 +1,7 @@
const Mhysa = require("mhysa"); const strom = require("stromjs");
let f = async () => { let f = async () => {
const source = Mhysa.fromArray(["a", "b", "c"]); const source = strom.fromArray(["a", "b", "c"]);
console.log(await Mhysa.last(source)); console.log(await strom.last(source));
}; };
f(); f();

View File

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

View File

@ -1,9 +1,9 @@
const { Readable } = require("stream"); const { Readable } = require("stream");
const Mhysa = require("mhysa"); const strom = require("stromjs");
const source1 = new Readable({ read() {} }); const source1 = new Readable({ read() {} });
const source2 = new Readable({ read() {} }); const source2 = new Readable({ read() {} });
Mhysa.merge(source1, source2).pipe(process.stdout); strom.merge(source1, source2).pipe(process.stdout);
source1.push("a1 "); source1.push("a1 ");
setTimeout(() => source2.push("c3 "), 10); setTimeout(() => source2.push("c3 "), 10);
setTimeout(() => source1.push("b2 "), 20); setTimeout(() => source1.push("b2 "), 20);

15
samples/parallelMap.js Normal file
View File

@ -0,0 +1,15 @@
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);

View File

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

21
samples/rate.js Normal file
View File

@ -0,0 +1,21 @@
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();

View File

@ -1,6 +1,7 @@
const Mhysa = require("mhysa"); const strom = require("stromjs");
Mhysa.fromArray(["a", "b", "cc"]) strom
.pipe(Mhysa.reduce((acc, s) => ({ ...acc, [s]: s.length }), {})) .fromArray(["a", "b", "cc"])
.pipe(Mhysa.stringify()) .pipe(strom.reduce((acc, s) => ({ ...acc, [s]: s.length }), {}))
.pipe(strom.stringify())
.pipe(process.stdout); .pipe(process.stdout);

View File

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

View File

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

View File

@ -1,5 +1,6 @@
const Mhysa = require("mhysa"); const strom = require("stromjs");
Mhysa.fromArray([{ a: "b" }]) strom
.pipe(Mhysa.stringify()) .fromArray([{ a: "b" }])
.pipe(strom.stringify())
.pipe(process.stdout); .pipe(process.stdout);

View File

@ -2,7 +2,7 @@ import { Transform, TransformOptions } from "stream";
export function batch( export function batch(
batchSize: number = 1000, batchSize: number = 1000,
maxBatchAge: number = 500, maxBatchAge: number = 0,
options: TransformOptions = {}, options: TransformOptions = {},
): Transform { ): Transform {
let buffer: any[] = []; let buffer: any[] = [];
@ -23,7 +23,7 @@ export function batch(
buffer.push(chunk); buffer.push(chunk);
if (buffer.length === batchSize) { if (buffer.length === batchSize) {
sendChunk(this); sendChunk(this);
} else { } else if (maxBatchAge) {
if (timer === null) { if (timer === null) {
timer = setInterval(() => { timer = setInterval(() => {
sendChunk(this); sendChunk(this);

View File

@ -9,30 +9,42 @@ export interface DemuxOptions extends DuplexOptions {
} }
export function demux( export function demux(
construct: (destKey?: string, chunk?: any) => DemuxStreams, pipelineConstructor: (
destKey?: string,
chunk?: any,
) => DemuxStreams | DemuxStreams[],
demuxBy: string | ((chunk: any) => string), demuxBy: string | ((chunk: any) => string),
options?: DemuxOptions, options?: DemuxOptions,
): Duplex { ): Duplex {
return new Demux(construct, demuxBy, options); return new Demux(pipelineConstructor, demuxBy, options);
} }
class Demux extends Duplex { class Demux extends Duplex {
private streamsByKey: { private streamsByKey: {
[key: string]: DemuxStreams; [key: string]: DemuxStreams[];
}; };
private demuxer: (chunk: any) => string; private demuxer: (chunk: any) => string;
private construct: (destKey?: string, chunk?: any) => DemuxStreams; private pipelineConstructor: (
destKey?: string,
chunk?: any,
) => DemuxStreams[];
private remultiplex: boolean; private remultiplex: boolean;
private transform: Transform; private transform: Transform;
constructor( constructor(
construct: (destKey?: string) => DemuxStreams, pipelineConstructor: (
destKey?: string,
chunk?: any,
) => DemuxStreams | DemuxStreams[],
demuxBy: string | ((chunk: any) => string), demuxBy: string | ((chunk: any) => string),
options: DemuxOptions = {}, options: DemuxOptions = {},
) { ) {
super(options); super(options);
this.demuxer = this.demuxer =
typeof demuxBy === "string" ? chunk => chunk[demuxBy] : demuxBy; typeof demuxBy === "string" ? chunk => chunk[demuxBy] : demuxBy;
this.construct = construct; this.pipelineConstructor = (destKey: string, chunk?: any) => {
const pipeline = pipelineConstructor(destKey, chunk);
return Array.isArray(pipeline) ? pipeline : [pipeline];
};
this.remultiplex = this.remultiplex =
options.remultiplex === undefined ? true : options.remultiplex; options.remultiplex === undefined ? true : options.remultiplex;
this.streamsByKey = {}; this.streamsByKey = {};
@ -53,43 +65,66 @@ class Demux extends Duplex {
public async _write(chunk: any, encoding: any, cb: any) { public async _write(chunk: any, encoding: any, cb: any) {
const destKey = this.demuxer(chunk); const destKey = this.demuxer(chunk);
if (this.streamsByKey[destKey] === undefined) { if (this.streamsByKey[destKey] === undefined) {
const newPipeline = await this.construct(destKey, chunk); const newPipelines = this.pipelineConstructor(destKey, chunk);
this.streamsByKey[destKey] = newPipeline; this.streamsByKey[destKey] = newPipelines;
newPipelines.forEach(newPipeline => {
if (this.remultiplex && isReadable(newPipeline)) { if (this.remultiplex && isReadable(newPipeline)) {
(newPipeline as NodeJS.ReadWriteStream).pipe(this.transform); (newPipeline as NodeJS.ReadWriteStream).pipe(
this.transform,
);
} else if (this.remultiplex) { } else if (this.remultiplex) {
console.error( console.error(
`Pipeline construct for ${destKey} does not implement readable interface`, `Pipeline construct for ${destKey} does not implement readable interface`,
); );
} }
}
if (!this.streamsByKey[destKey].write(chunk, encoding)) {
this.streamsByKey[destKey].once("drain", () => {
cb();
}); });
} else {
cb();
} }
const pipelines = this.streamsByKey[destKey];
const pendingDrains: Array<Promise<any>> = [];
pipelines.forEach(pipeline => {
if (!pipeline.write(chunk, encoding)) {
pendingDrains.push(
new Promise(resolve => {
pipeline.once("drain", () => {
resolve();
});
}),
);
}
});
await Promise.all(pendingDrains);
cb();
} }
public _flush() { public _flush() {
const pipelines = Object.values(this.streamsByKey); const pipelines: DemuxStreams[] = Array.prototype.concat.apply(
let totalEnded = 0; [],
Object.values(this.streamsByKey),
);
const flushPromises: Array<Promise<void>> = [];
pipelines.forEach(pipeline => { pipelines.forEach(pipeline => {
flushPromises.push(
new Promise(resolve => {
pipeline.once("end", () => { pipeline.once("end", () => {
totalEnded++; resolve();
if (pipelines.length === totalEnded) {
this.push(null);
this.emit("end");
}
}); });
}),
);
}); });
pipelines.forEach(pipeline => pipeline.end()); pipelines.forEach(pipeline => pipeline.end());
Promise.all(flushPromises).then(() => {
this.push(null);
this.emit("end");
});
} }
public _destroy(error: any, cb: (error?: any) => void) { public _destroy(error: any, cb: (error?: any) => void) {
const pipelines = Object.values(this.streamsByKey); const pipelines: DemuxStreams[] = [].concat.apply(
[],
Object.values(this.streamsByKey),
);
pipelines.forEach(p => (p as any).destroy()); pipelines.forEach(p => (p as any).destroy());
cb(error); cb(error);
} }

View File

@ -28,7 +28,7 @@ import { unbatch } from "./unbatch";
import { compose } from "./compose"; import { compose } from "./compose";
import { demux } from "./demux"; import { demux } from "./demux";
export default function mhysa(defaultOptions?: TransformOptions) { export function strom(defaultOptions: TransformOptions = { objectMode: true }) {
function withDefaultOptions<T extends any[], R>( function withDefaultOptions<T extends any[], R>(
n: number, n: number,
fn: (...args: T) => R, fn: (...args: T) => R,
@ -263,5 +263,10 @@ export default function mhysa(defaultOptions?: TransformOptions) {
* @param options Writable stream options * @param options Writable stream options
*/ */
demux: withDefaultOptions(2, demux), demux: withDefaultOptions(2, demux),
/**
* Create a new strom instance overriding the defaults
*/
instance: strom,
}; };
} }

View File

@ -2,21 +2,42 @@ import { Transform, TransformOptions } from "stream";
import { performance } from "perf_hooks"; import { performance } from "perf_hooks";
import { sleep } from "../helpers"; import { sleep } from "../helpers";
export enum Behavior {
BUFFER = 0,
DROP = 1,
}
export interface RateOptions {
window?: number;
behavior?: Behavior;
}
export function rate( export function rate(
targetRate: number = 50, targetRate: number = 50,
period: number = 1, period: number = 1,
options?: TransformOptions, options?: TransformOptions & RateOptions,
): Transform { ): Transform {
const deltaMS = ((1 / targetRate) * 1000) / period; // Skip a full period const deltaMS = ((1 / targetRate) * 1000) / period; // Skip a full period
let total = 0; 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({ return new Transform({
...options, ...options,
async transform(data, encoding, callback) { 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 (targetRate && currentRate > targetRate) {
if (behavior === Behavior.DROP) {
callback(undefined);
return;
} else {
await sleep(deltaMS); await sleep(deltaMS);
} }
}
total += 1; total += 1;
callback(undefined, data); callback(undefined, data);
}, },

View File

@ -1,6 +1,29 @@
import mhysa from "./functions"; import { strom } from "./functions";
import * as _utils from "./utils"; export * from "./utils";
export default mhysa; export const {
fromArray,
// @TODO fix this with proper import export map,
export const utils = { ..._utils }; flatMap,
filter,
reduce,
split,
join,
replace,
parse,
stringify,
collect,
concat,
merge,
duplex,
child,
last,
batch,
unbatch,
rate,
parallelMap,
accumulator,
accumulatorBy,
compose,
demux,
instance,
} = strom();

View File

@ -1,10 +1,9 @@
import test from "ava"; import test from "ava";
import { expect } from "chai"; import { expect } from "chai";
import { Readable } from "stream"; import { Readable } from "stream";
import mhysa from "../src"; import { accumulator, accumulatorBy } from "../src";
import { FlushStrategy } from "../src/functions/accumulator"; import { FlushStrategy } from "../src/functions/accumulator";
import { performance } from "perf_hooks"; import { performance } from "perf_hooks";
const { accumulator, accumulatorBy } = mhysa({ objectMode: true });
test.cb("accumulator() rolling", t => { test.cb("accumulator() rolling", t => {
t.plan(3); t.plan(3);
@ -14,8 +13,14 @@ test.cb("accumulator() rolling", t => {
key: string; key: string;
} }
const source = new Readable({ objectMode: true }); const source = new Readable({ objectMode: true });
const firstFlush = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }]; const firstFlush = [
const secondFlush = [{ ts: 2, key: "d" }, { ts: 3, key: "e" }]; { ts: 0, key: "a" },
{ ts: 1, key: "b" },
];
const secondFlush = [
{ ts: 2, key: "d" },
{ ts: 3, key: "e" },
];
const thirdFlush = [{ ts: 4, key: "f" }]; const thirdFlush = [{ ts: 4, key: "f" }];
const flushes = [firstFlush, secondFlush, thirdFlush]; const flushes = [firstFlush, secondFlush, thirdFlush];
@ -88,7 +93,10 @@ test.cb(
"nonExistingKey", "nonExistingKey",
{ objectMode: true }, { objectMode: true },
); );
const input = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }]; const input = [
{ ts: 0, key: "a" },
{ ts: 1, key: "b" },
];
source source
.pipe(accumulatorStream) .pipe(accumulatorStream)
@ -182,7 +190,10 @@ test.cb("accumulator() sliding", t => {
{ ts: 4, key: "d" }, { ts: 4, key: "d" },
]; ];
const firstFlush = [{ ts: 0, key: "a" }]; const firstFlush = [{ ts: 0, key: "a" }];
const secondFlush = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }]; const secondFlush = [
{ ts: 0, key: "a" },
{ ts: 1, key: "b" },
];
const thirdFlush = [ const thirdFlush = [
{ ts: 0, key: "a" }, { ts: 0, key: "a" },
{ ts: 1, key: "b" }, { ts: 1, key: "b" },
@ -232,7 +243,10 @@ test.cb("accumulator() sliding with key", t => {
{ ts: 6, key: "g" }, { ts: 6, key: "g" },
]; ];
const firstFlush = [{ ts: 0, key: "a" }]; const firstFlush = [{ ts: 0, key: "a" }];
const secondFlush = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }]; const secondFlush = [
{ ts: 0, key: "a" },
{ ts: 1, key: "b" },
];
const thirdFlush = [ const thirdFlush = [
{ ts: 0, key: "a" }, { ts: 0, key: "a" },
{ ts: 1, key: "b" }, { ts: 1, key: "b" },
@ -243,8 +257,14 @@ test.cb("accumulator() sliding with key", t => {
{ ts: 2, key: "c" }, { ts: 2, key: "c" },
{ ts: 3, key: "d" }, { ts: 3, key: "d" },
]; ];
const fifthFlush = [{ ts: 3, key: "d" }, { ts: 5, key: "f" }]; const fifthFlush = [
const sixthFlush = [{ ts: 5, key: "f" }, { ts: 6, key: "g" }]; { ts: 3, key: "d" },
{ ts: 5, key: "f" },
];
const sixthFlush = [
{ ts: 5, key: "f" },
{ ts: 6, key: "g" },
];
const flushes = [ const flushes = [
firstFlush, firstFlush,
@ -286,7 +306,10 @@ test.cb(
"nonExistingKey", "nonExistingKey",
{ objectMode: true }, { objectMode: true },
); );
const input = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }]; const input = [
{ ts: 0, key: "a" },
{ ts: 1, key: "b" },
];
source source
.pipe(accumulatorStream) .pipe(accumulatorStream)
@ -334,10 +357,22 @@ test.cb(
{ ts: 6, key: "g" }, { ts: 6, key: "g" },
]; ];
const firstFlush = [{ ts: 0, key: "a" }]; const firstFlush = [{ ts: 0, key: "a" }];
const secondFlush = [{ ts: 0, key: "a" }, { ts: 2, key: "c" }]; const secondFlush = [
const thirdFlush = [{ ts: 2, key: "c" }, { ts: 3, key: "d" }]; { ts: 0, key: "a" },
const fourthFlush = [{ ts: 3, key: "d" }, { ts: 5, key: "f" }]; { ts: 2, key: "c" },
const fifthFlush = [{ ts: 5, key: "f" }, { ts: 6, key: "g" }]; ];
const thirdFlush = [
{ ts: 2, key: "c" },
{ ts: 3, key: "d" },
];
const fourthFlush = [
{ ts: 3, key: "d" },
{ ts: 5, key: "f" },
];
const fifthFlush = [
{ ts: 5, key: "f" },
{ ts: 6, key: "g" },
];
const flushes = [ const flushes = [
firstFlush, firstFlush,
@ -469,7 +504,10 @@ test.cb("accumulatorBy() sliding", t => {
{ ts: 6, key: "g" }, { ts: 6, key: "g" },
]; ];
const firstFlush = [{ ts: 0, key: "a" }]; const firstFlush = [{ ts: 0, key: "a" }];
const secondFlush = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }]; const secondFlush = [
{ ts: 0, key: "a" },
{ ts: 1, key: "b" },
];
const thirdFlush = [ const thirdFlush = [
{ ts: 0, key: "a" }, { ts: 0, key: "a" },
{ ts: 1, key: "b" }, { ts: 1, key: "b" },
@ -480,8 +518,14 @@ test.cb("accumulatorBy() sliding", t => {
{ ts: 2, key: "c" }, { ts: 2, key: "c" },
{ ts: 3, key: "d" }, { ts: 3, key: "d" },
]; ];
const fifthFlush = [{ ts: 3, key: "d" }, { ts: 5, key: "f" }]; const fifthFlush = [
const sixthFlush = [{ ts: 5, key: "f" }, { ts: 6, key: "g" }]; { ts: 3, key: "d" },
{ ts: 5, key: "f" },
];
const sixthFlush = [
{ ts: 5, key: "f" },
{ ts: 6, key: "g" },
];
const flushes = [ const flushes = [
firstFlush, firstFlush,

View File

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

View File

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

View File

@ -1,8 +1,7 @@
import { Readable } from "stream"; import { Readable } from "stream";
import test from "ava"; import test from "ava";
import { expect } from "chai"; import { expect } from "chai";
import mhysa from "../src"; import { collect } from "../src";
const { collect } = mhysa();
test.cb( test.cb(
"collect() collects streamed elements into an array (object, flowing mode)", "collect() collects streamed elements into an array (object, flowing mode)",
@ -59,7 +58,7 @@ test.cb(
const source = new Readable({ objectMode: false }); const source = new Readable({ objectMode: false });
source source
.pipe(collect()) .pipe(collect({ objectMode: false }))
.on("data", collected => { .on("data", collected => {
expect(collected).to.deep.equal(Buffer.from("abc")); expect(collected).to.deep.equal(Buffer.from("abc"));
t.pass(); t.pass();

View File

@ -2,9 +2,8 @@ import * as test from "ava";
import { expect } from "chai"; import { expect } from "chai";
import { sleep } from "../src/helpers"; import { sleep } from "../src/helpers";
import { Readable, Writable } from "stream"; import { Readable, Writable } from "stream";
import mhysa from "../src"; import { compose, map, fromArray } from "../src";
import { performance } from "perf_hooks"; import { performance } from "perf_hooks";
const { compose, map, fromArray } = mhysa({ objectMode: true });
test.cb("compose() chains two streams together in the correct order", t => { test.cb("compose() chains two streams together in the correct order", t => {
t.plan(3); t.plan(3);
@ -114,13 +113,9 @@ test("compose() writable length should be less than highWaterMark when handing w
return chunk; return chunk;
}); });
const composed = compose( const composed = compose([first, second], undefined, {
[first, second],
undefined,
{
highWaterMark: 2, highWaterMark: 2,
}, });
);
composed.on("error", err => { composed.on("error", err => {
reject(); reject();
}); });
@ -168,13 +163,9 @@ test("compose() should emit drain event ~rate * highWaterMark ms for every write
return chunk; return chunk;
}); });
const composed = compose( const composed = compose([first, second], undefined, {
[first, second],
undefined,
{
highWaterMark, highWaterMark,
}, });
);
composed.on("error", err => { composed.on("error", err => {
reject(); reject();
}); });
@ -221,13 +212,9 @@ test.cb(
return chunk; return chunk;
}); });
const composed = compose( const composed = compose([first, second], undefined, {
[first, second],
undefined,
{
highWaterMark: 5, highWaterMark: 5,
}, });
);
composed.on("error", err => { composed.on("error", err => {
t.end(err); t.end(err);
@ -284,13 +271,9 @@ test.cb(
{ highWaterMark: 1 }, { highWaterMark: 1 },
); );
const composed = compose( const composed = compose([first, second], undefined, {
[first, second],
undefined,
{
highWaterMark: 5, highWaterMark: 5,
}, });
);
composed.on("error", err => { composed.on("error", err => {
t.end(err); t.end(err);
}); });
@ -355,13 +338,9 @@ test.cb(
{ highWaterMark: 2 }, { highWaterMark: 2 },
); );
const composed = compose( const composed = compose([first, second], undefined, {
[first, second],
undefined,
{
highWaterMark: 5, highWaterMark: 5,
}, });
);
composed.on("error", err => { composed.on("error", err => {
t.end(err); t.end(err);
}); });
@ -414,13 +393,9 @@ test.cb(
return chunk; return chunk;
}); });
const composed = compose( const composed = compose([first, second], undefined, {
[first, second],
undefined,
{
highWaterMark: 6, highWaterMark: 6,
}, });
);
composed.on("error", err => { composed.on("error", err => {
t.end(err); t.end(err);
@ -470,12 +445,9 @@ test.cb("compose() should be 'destroyable'", t => {
return chunk; return chunk;
}); });
const composed = compose( const composed = compose([first, second], (err: any) => {
[first, second],
(err: any) => {
t.pass(); t.pass();
}, });
);
const fakeSource = new Readable({ const fakeSource = new Readable({
objectMode: true, objectMode: true,
@ -531,13 +503,9 @@ test.cb("compose() `finish` and `end` propagates", t => {
return chunk; return chunk;
}); });
const composed = compose( const composed = compose([first, second], undefined, {
[first, second],
undefined,
{
highWaterMark: 3, highWaterMark: 3,
}, });
);
const fakeSource = new Readable({ const fakeSource = new Readable({
objectMode: true, objectMode: true,

View File

@ -1,8 +1,7 @@
import { Readable } from "stream"; import { Readable } from "stream";
import test from "ava"; import test from "ava";
import { expect } from "chai"; import { expect } from "chai";
import mhysa from "../src"; import { concat, collect } from "../src";
const { concat, collect } = mhysa();
test.cb( test.cb(
"concat() concatenates multiple readable streams (object, flowing mode)", "concat() concatenates multiple readable streams (object, flowing mode)",
@ -172,7 +171,7 @@ test.cb(
test.cb("concat() concatenates empty list of readable streams", t => { test.cb("concat() concatenates empty list of readable streams", t => {
t.plan(0); t.plan(0);
concat() concat()
.pipe(collect()) .pipe(collect({ objectMode: false }))
.on("data", _ => { .on("data", _ => {
t.fail(); t.fail();
}) })

View File

@ -1,21 +1,30 @@
import { Readable } from "stream"; import { Readable } from "stream";
import test from "ava"; import test from "ava";
import mhysa from "../src"; import { batch as _batch, instance as strom } from "../src";
const withDefaultOptions = mhysa({ objectMode: true }); const withDefaultOptions = strom({ objectMode: false });
const withoutOptions = mhysa();
test("Mhysa instances can have default options", t => { test("strom instances can have default options", t => {
let batch = withDefaultOptions.batch(); 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); 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); 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._readableState.objectMode);
t.true(batch._writableState.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._readableState.objectMode);
t.false(batch._writableState.objectMode); t.false(batch._writableState.objectMode);
}); });

View File

@ -1,11 +1,10 @@
import test from "ava"; import test from "ava";
import { expect } from "chai"; import { expect } from "chai";
import mhysa from "../src"; import { demux, map, fromArray } from "../src";
import { Writable, Readable } from "stream"; import { Writable, Readable } from "stream";
const sinon = require("sinon"); import * as sinon from "sinon";
const { sleep } = require("../src/helpers"); import { sleep } from "../src/helpers";
import { performance } from "perf_hooks"; import { performance } from "perf_hooks";
const { demux, map, fromArray } = mhysa({ objectMode: true });
interface Test { interface Test {
key: string; key: string;

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@ -1,8 +1,7 @@
import { Readable } from "stream"; import { Readable } from "stream";
import test from "ava"; import test from "ava";
import { expect } from "chai"; import { expect } from "chai";
import mhysa from "../src"; import { last } from "../src";
const { last } = mhysa();
test("last() resolves to the last chunk streamed by the given readable stream", async t => { test("last() resolves to the last chunk streamed by the given readable stream", async t => {
const source = new Readable({ objectMode: true }); const source = new Readable({ objectMode: true });

View File

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

View File

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

View File

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

View File

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

View File

@ -1,12 +1,11 @@
import { Readable } from "stream"; import { Readable } from "stream";
import { performance } from "perf_hooks"; import { performance } from "perf_hooks";
import test from "ava"; import test from "ava";
import { expect } from "chai"; import { rate } from "../src";
import mhysa from "../src"; import { sleep } from "../src/helpers";
const { rate } = mhysa({ objectMode: true });
test.cb("rate() sends data at a rate of 150", t => { test.cb("rate() sends data at a rate of 150", t => {
t.plan(5); t.plan(15);
const targetRate = 150; const targetRate = 150;
const source = new Readable({ objectMode: true }); const source = new Readable({ objectMode: true });
const expectedElements = ["a", "b", "c", "d", "e"]; const expectedElements = ["a", "b", "c", "d", "e"];
@ -15,10 +14,10 @@ test.cb("rate() sends data at a rate of 150", t => {
source source
.pipe(rate(targetRate)) .pipe(rate(targetRate))
.on("data", (element: string[]) => { .on("data", (element: string) => {
const currentRate = (i / (performance.now() - start)) * 1000; 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(); t.pass();
i++; 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 => { test.cb("rate() sends data at a rate of 50", t => {
t.plan(5); t.plan(15);
const targetRate = 50; const targetRate = 50;
const source = new Readable({ objectMode: true }); const source = new Readable({ objectMode: true });
const expectedElements = ["a", "b", "c", "d", "e"]; const expectedElements = ["a", "b", "c", "d", "e"];
@ -43,10 +42,10 @@ test.cb("rate() sends data at a rate of 50", t => {
source source
.pipe(rate(targetRate)) .pipe(rate(targetRate))
.on("data", (element: string[]) => { .on("data", (element: string) => {
const currentRate = (i / (performance.now() - start)) * 1000; 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(); t.pass();
i++; 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 => { test.cb("rate() sends data at a rate of 1", t => {
t.plan(5); t.plan(15);
const targetRate = 1; const targetRate = 1;
const source = new Readable({ objectMode: true }); const source = new Readable({ objectMode: true });
const expectedElements = ["a", "b", "c", "d", "e"]; const expectedElements = ["a", "b", "c", "d", "e"];
@ -71,10 +70,10 @@ test.cb("rate() sends data at a rate of 1", t => {
source source
.pipe(rate(targetRate)) .pipe(rate(targetRate))
.on("data", (element: string[]) => { .on("data", (element: string) => {
const currentRate = (i / (performance.now() - start)) * 1000; 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(); t.pass();
i++; i++;
}) })
@ -88,3 +87,41 @@ test.cb("rate() sends data at a rate of 1", t => {
source.push("e"); source.push("e");
source.push(null); 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);
});

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@ -1,7 +1,6 @@
import test from "ava"; import test from "ava";
import { collected } from "../../src/utils"; import { collected } from "../../src/utils";
import mhysa from "../../src"; import { fromArray, collect } from "../../src";
const { fromArray, collect } = mhysa({ objectMode: true });
test("collected returns a promise for the first data point", async t => { test("collected returns a promise for the first data point", async t => {
const data = collected(fromArray([1, 2, 3, 4]).pipe(collect())); const data = collected(fromArray([1, 2, 3, 4]).pipe(collect()));

View File

@ -1934,9 +1934,6 @@ merge2@^1.2.3, merge2@^1.3.0:
resolved "https://npm.dev.jogogo.co/merge2/-/merge2-1.3.0.tgz#5b366ee83b2f1582c48f87e47cf1a9352103ca81" resolved "https://npm.dev.jogogo.co/merge2/-/merge2-1.3.0.tgz#5b366ee83b2f1582c48f87e47cf1a9352103ca81"
integrity sha512-2j4DAdlBOkiSZIsaXk4mTE3sRS02yBHAtfy127xRV3bQUFqXkjHCHLW6Scv7DwNRbIWNHH8zpnz9zMaKXIdvYw== integrity sha512-2j4DAdlBOkiSZIsaXk4mTE3sRS02yBHAtfy127xRV3bQUFqXkjHCHLW6Scv7DwNRbIWNHH8zpnz9zMaKXIdvYw==
mhysa@./:
version "2.0.0-alpha.1"
micromatch@^4.0.2: micromatch@^4.0.2:
version "4.0.2" version "4.0.2"
resolved "https://npm.dev.jogogo.co/micromatch/-/micromatch-4.0.2.tgz#4fcb0999bf9fbc2fcbdd212f6d629b9a56c39259" resolved "https://npm.dev.jogogo.co/micromatch/-/micromatch-4.0.2.tgz#4fcb0999bf9fbc2fcbdd212f6d629b9a56c39259"
@ -2719,6 +2716,9 @@ strip-json-comments@~2.0.1:
resolved "https://registry.yarnpkg.com/strip-json-comments/-/strip-json-comments-2.0.1.tgz#3c531942e908c2697c0ec344858c286c7ca0a60a" resolved "https://registry.yarnpkg.com/strip-json-comments/-/strip-json-comments-2.0.1.tgz#3c531942e908c2697c0ec344858c286c7ca0a60a"
integrity sha1-PFMZQukIwml8DsNEhYwobHygpgo= integrity sha1-PFMZQukIwml8DsNEhYwobHygpgo=
stromjs@./:
version "0.5.1"
supertap@^1.0.0: supertap@^1.0.0:
version "1.0.0" version "1.0.0"
resolved "https://registry.yarnpkg.com/supertap/-/supertap-1.0.0.tgz#bd9751c7fafd68c68cf8222a29892206a119fa9e" resolved "https://registry.yarnpkg.com/supertap/-/supertap-1.0.0.tgz#bd9751c7fafd68c68cf8222a29892206a119fa9e"