Compare commits
3 Commits
master
...
feature/de
Author | SHA1 | Date | |
---|---|---|---|
|
46d118d4a5 | ||
|
5dd856deed | ||
|
d0a9d35fe7 |
206
README.md
206
README.md
@ -1,16 +1,15 @@
|
||||
# Strom
|
||||
# Mhysa
|
||||
|
||||
**Dependency-free stream utils for Node.js**
|
||||
|
||||
<sub>Released under the [MIT](LICENSE) license.</sub>
|
||||
<sub>Released under the [MIT](https://github.com/Wenzil/Mhysa/blob/master/LICENSE) license.</sub>
|
||||
|
||||
```sh
|
||||
yarn add stromjs
|
||||
```
|
||||
```sh
|
||||
npm add stromjs
|
||||
yarn add mhysa
|
||||
```
|
||||
|
||||
<sub>Tested with Node.js versions 8+</sub>
|
||||
|
||||
## fromArray(array)
|
||||
Convert an array into a `Readable` stream of its elements
|
||||
|
||||
@ -19,14 +18,14 @@ Convert an array into a `Readable` stream of its elements
|
||||
| `array` | `T[]` | Array of elements to stream |
|
||||
|
||||
```js
|
||||
strom.fromArray(["a", "b"])
|
||||
Mhysa.fromArray(["a", "b"])
|
||||
.pipe(process.stdout);
|
||||
// ab is printed out
|
||||
```
|
||||
|
||||
|
||||
## map(mapper, options)
|
||||
Returns a `ReadWrite` stream that maps streamed chunks
|
||||
Return a `ReadWrite` stream that maps streamed chunks
|
||||
|
||||
| Param | Type | Description |
|
||||
| --- | --- | --- |
|
||||
@ -36,15 +35,15 @@ Returns a `ReadWrite` stream that maps streamed chunks
|
||||
| `options.writableObjectMode` | `boolean` | Whether this stream should behave as a writable stream of objects |
|
||||
|
||||
```js
|
||||
strom.fromArray(["a", "b"])
|
||||
.pipe(strom.map(s => s.toUpperCase()))
|
||||
Mhysa.fromArray(["a", "b"])
|
||||
.pipe(Mhysa.map(s => s.toUpperCase()))
|
||||
.pipe(process.stdout);
|
||||
// AB is printed out
|
||||
```
|
||||
|
||||
|
||||
## flatMap(mapper, options)
|
||||
Returns a `ReadWrite` stream that flat maps streamed chunks
|
||||
Return a `ReadWrite` stream that flat maps streamed chunks
|
||||
|
||||
| Param | Type | Description |
|
||||
| --- | --- | --- |
|
||||
@ -54,15 +53,15 @@ Returns a `ReadWrite` stream that flat maps streamed chunks
|
||||
| `options.writableObjectMode` | `boolean` | Whether this stream should behave as a writable stream of objects |
|
||||
|
||||
```js
|
||||
strom.fromArray(["a", "AA"])
|
||||
.pipe(strom.flatMap(s => new Array(s.length).fill(s)))
|
||||
Mhysa.fromArray(["a", "AA"])
|
||||
.pipe(Mhysa.flatMap(s => new Array(s.length).fill(s)))
|
||||
.pipe(process.stdout);
|
||||
// aAAAA is printed out
|
||||
```
|
||||
|
||||
|
||||
## filter(predicate, options)
|
||||
Returns a `ReadWrite` stream that filters out streamed chunks for which the predicate does not hold
|
||||
Return a `ReadWrite` stream that filters out streamed chunks for which the predicate does not hold
|
||||
|
||||
| Param | Type | Description |
|
||||
| --- | --- | --- |
|
||||
@ -71,15 +70,15 @@ Returns a `ReadWrite` stream that filters out streamed chunks for which the pred
|
||||
| `options.objectMode` | `boolean` | `boolean` | Whether this stream should behave as a stream of objects |
|
||||
|
||||
```js
|
||||
strom.fromArray(["a", "b", "c"])
|
||||
.pipe(strom.filter(s => s !== "b"))
|
||||
Mhysa.fromArray(["a", "b", "c"])
|
||||
.pipe(Mhysa.filter(s => s !== "b"))
|
||||
.pipe(process.stdout);
|
||||
// ac is printed out
|
||||
```
|
||||
|
||||
|
||||
## reduce(iteratee, initialValue, options)
|
||||
Returns a `ReadWrite` stream that reduces streamed chunks down to a single value and yield that
|
||||
Return a `ReadWrite` stream that reduces streamed chunks down to a single value and yield that
|
||||
value
|
||||
|
||||
| Param | Type | Description |
|
||||
@ -91,16 +90,16 @@ value
|
||||
| `options.writableObjectMode` | `boolean` | Whether this stream should behave as a writable stream of objects |
|
||||
|
||||
```js
|
||||
strom.fromArray(["a", "b", "cc"])
|
||||
.pipe(strom.reduce((acc, s) => ({ ...acc, [s]: s.length }), {}))
|
||||
.pipe(strom.stringify())
|
||||
Mhysa.fromArray(["a", "b", "cc"])
|
||||
.pipe(Mhysa.reduce((acc, s) => ({ ...acc, [s]: s.length }), {}))
|
||||
.pipe(Mhysa.stringify())
|
||||
.pipe(process.stdout);
|
||||
// {"a":1,"b":1","c":2} is printed out
|
||||
```
|
||||
|
||||
|
||||
## split(separator)
|
||||
Returns a `ReadWrite` stream that splits streamed chunks using the given separator
|
||||
Return a `ReadWrite` stream that splits streamed chunks using the given separator
|
||||
|
||||
| Param | Type | Description |
|
||||
| --- | --- | --- |
|
||||
@ -109,16 +108,16 @@ Returns a `ReadWrite` stream that splits streamed chunks using the given separat
|
||||
| `options.encoding` | `string` | Character encoding to use for decoding chunks. Defaults to utf8
|
||||
|
||||
```js
|
||||
strom.fromArray(["a,b", "c,d"])
|
||||
.pipe(strom.split(","))
|
||||
.pipe(strom.join("|"))
|
||||
Mhysa.fromArray(["a,b", "c,d"])
|
||||
.pipe(Mhysa.split(","))
|
||||
.pipe(Mhysa.join("|"))
|
||||
.pipe(process.stdout);
|
||||
// a|bc|d is printed out
|
||||
```
|
||||
|
||||
|
||||
## join(separator)
|
||||
Returns a `ReadWrite` stream that joins streamed chunks using the given separator
|
||||
Return a `ReadWrite` stream that joins streamed chunks using the given separator
|
||||
|
||||
| Param | Type | Description |
|
||||
| --- | --- | --- |
|
||||
@ -127,15 +126,15 @@ Returns a `ReadWrite` stream that joins streamed chunks using the given separato
|
||||
| `options.encoding` | `string` | Character encoding to use for decoding chunks. Defaults to utf8
|
||||
|
||||
```js
|
||||
strom.fromArray(["a", "b", "c"])
|
||||
.pipe(strom.join(","))
|
||||
Mhysa.fromArray(["a", "b", "c"])
|
||||
.pipe(Mhysa.join(","))
|
||||
.pipe(process.stdout);
|
||||
// a,b,c is printed out
|
||||
```
|
||||
|
||||
|
||||
## replace(searchValue, replaceValue)
|
||||
Returns a `ReadWrite` stream that replaces occurrences of the given string or regular expression in
|
||||
Return 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 |
|
||||
@ -146,37 +145,37 @@ the streamed chunks with the specified replacement string
|
||||
| `options.encoding` | `string` | Character encoding to use for decoding chunks. Defaults to utf8
|
||||
|
||||
```js
|
||||
strom.fromArray(["a1", "b22", "c333"])
|
||||
.pipe(strom.replace(/b\d+/, "B"))
|
||||
Mhysa.fromArray(["a1", "b22", "c333"])
|
||||
.pipe(Mhysa.replace(/b\d+/, "B"))
|
||||
.pipe(process.stdout);
|
||||
// a1Bc333 is printed out
|
||||
```
|
||||
|
||||
|
||||
## parse()
|
||||
Returns a `ReadWrite` stream that parses the streamed chunks as JSON
|
||||
Return a `ReadWrite` stream that parses the streamed chunks as JSON
|
||||
|
||||
```js
|
||||
strom.fromArray(['{ "a": "b" }'])
|
||||
.pipe(strom.parse())
|
||||
Mhysa.fromArray(['{ "a": "b" }'])
|
||||
.pipe(Mhysa.parse())
|
||||
.once("data", object => console.log(object));
|
||||
// { a: 'b' } is printed out
|
||||
```
|
||||
|
||||
|
||||
## stringify()
|
||||
Returns a `ReadWrite` stream that stringifies the streamed chunks to JSON
|
||||
Return a `ReadWrite` stream that stringifies the streamed chunks to JSON
|
||||
|
||||
```js
|
||||
strom.fromArray([{ a: "b" }])
|
||||
.pipe(strom.stringify())
|
||||
Mhysa.fromArray([{ a: "b" }])
|
||||
.pipe(Mhysa.stringify())
|
||||
.pipe(process.stdout);
|
||||
// {"a":"b"} is printed out
|
||||
```
|
||||
|
||||
|
||||
## collect(options)
|
||||
Returns a `ReadWrite` stream that collects streamed chunks into an array or buffer
|
||||
Return a `ReadWrite` stream that collects streamed chunks into an array or buffer
|
||||
|
||||
| Param | Type | Description |
|
||||
| --- | --- | --- |
|
||||
@ -184,15 +183,15 @@ Returns a `ReadWrite` stream that collects streamed chunks into an array or buff
|
||||
| `options.objectMode` | `boolean` | Whether this stream should behave as a stream of objects |
|
||||
|
||||
```js
|
||||
strom.fromArray(["a", "b", "c"])
|
||||
.pipe(strom.collect({ objectMode: true }))
|
||||
Mhysa.fromArray(["a", "b", "c"])
|
||||
.pipe(Mhysa.collect({ objectMode: true }))
|
||||
.once("data", object => console.log(object));
|
||||
// [ 'a', 'b', 'c' ] is printed out
|
||||
```
|
||||
|
||||
|
||||
## concat(streams)
|
||||
Returns a `Readable` stream of readable streams concatenated together
|
||||
Return a `Readable` stream of readable streams concatenated together
|
||||
|
||||
| Param | Type | Description |
|
||||
| --- | --- | --- |
|
||||
@ -201,7 +200,7 @@ Returns a `Readable` stream of readable streams concatenated together
|
||||
```js
|
||||
const source1 = new Readable();
|
||||
const source2 = new Readable();
|
||||
strom.concat(source1, source2).pipe(process.stdout)
|
||||
Mhysa.concat(source1, source2).pipe(process.stdout)
|
||||
source1.push("a1 ");
|
||||
source2.push("c3 ");
|
||||
source1.push("b2 ");
|
||||
@ -213,7 +212,7 @@ source2.push(null);
|
||||
|
||||
|
||||
## merge(streams)
|
||||
Returns a `Readable` stream of readable streams merged together in chunk arrival order
|
||||
Return a `Readable` stream of readable streams merged together in chunk arrival order
|
||||
|
||||
| Param | Type | Description |
|
||||
| --- | --- | --- |
|
||||
@ -222,7 +221,7 @@ Returns a `Readable` stream of readable streams merged together in chunk arrival
|
||||
```js
|
||||
const source1 = new Readable({ read() {} });
|
||||
const source2 = new Readable({ read() {} });
|
||||
strom.merge(source1, source2).pipe(process.stdout);
|
||||
Mhysa.merge(source1, source2).pipe(process.stdout);
|
||||
source1.push("a1 ");
|
||||
setTimeout(() => source2.push("c3 "), 10);
|
||||
setTimeout(() => source1.push("b2 "), 20);
|
||||
@ -234,7 +233,7 @@ setTimeout(() => source2.push(null), 50);
|
||||
|
||||
|
||||
## duplex(writable, readable)
|
||||
Returns a `Duplex` stream from a writable stream that is assumed to somehow, when written to,
|
||||
Return 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 |
|
||||
@ -244,15 +243,15 @@ cause the given readable stream to yield chunks
|
||||
|
||||
```js
|
||||
const catProcess = require("child_process").exec("grep -o ab");
|
||||
strom.fromArray(["a", "b", "c"])
|
||||
.pipe(strom.duplex(catProcess.stdin, catProcess.stdout))
|
||||
Mhysa.fromArray(["a", "b", "c"])
|
||||
.pipe(Mhysa.duplex(catProcess.stdin, catProcess.stdout))
|
||||
.pipe(process.stdout);
|
||||
// ab is printed out
|
||||
```
|
||||
|
||||
|
||||
## child(childProcess)
|
||||
Returns a `Duplex` stream from a child process' stdin and stdout
|
||||
Return a `Duplex` stream from a child process' stdin and stdout
|
||||
|
||||
| Param | Type | Description |
|
||||
| --- | --- | --- |
|
||||
@ -260,15 +259,15 @@ Returns a `Duplex` stream from a child process' stdin and stdout
|
||||
|
||||
```js
|
||||
const catProcess = require("child_process").exec("grep -o ab");
|
||||
strom.fromArray(["a", "b", "c"])
|
||||
.pipe(strom.child(catProcess))
|
||||
Mhysa.fromArray(["a", "b", "c"])
|
||||
.pipe(Mhysa.child(catProcess))
|
||||
.pipe(process.stdout);
|
||||
// ab is printed out
|
||||
```
|
||||
|
||||
|
||||
## last(readable)
|
||||
Returns a `Promise` resolving to the last streamed chunk of the given readable stream, after it has
|
||||
Return a `Promise` resolving to the last streamed chunk of the given readable stream, after it has
|
||||
ended
|
||||
|
||||
| Param | Type | Description |
|
||||
@ -277,112 +276,9 @@ ended
|
||||
|
||||
```js
|
||||
let f = async () => {
|
||||
const source = strom.fromArray(["a", "b", "c"]);
|
||||
console.log(await strom.last(source));
|
||||
const source = Mhysa.fromArray(["a", "b", "c"]);
|
||||
console.log(await Mhysa.last(source));
|
||||
};
|
||||
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();
|
||||
```
|
||||
|
34
package.json
34
package.json
@ -1,26 +1,24 @@
|
||||
{
|
||||
"name": "stromjs",
|
||||
"version": "0.5.1",
|
||||
"description": "Dependency-free streams utils for Node.js",
|
||||
"name": "@jogogo/mhysa",
|
||||
"version": "2.0.0-alpha.3",
|
||||
"description": "Streams and event emitter utils for Node.js",
|
||||
"keywords": [
|
||||
"promise",
|
||||
"stream",
|
||||
"event emitter",
|
||||
"utils"
|
||||
],
|
||||
"author": {
|
||||
"name": "Wenzil"
|
||||
},
|
||||
"contributors": [
|
||||
{
|
||||
"name": "Sami Turcotte",
|
||||
"url": "https://github.com/Wenzil"
|
||||
"name": "jerry",
|
||||
"email": "jerry@jogogo.co"
|
||||
},
|
||||
{
|
||||
"name": "Jerry Kurian",
|
||||
"email": "jerrykurian@protonmail.com",
|
||||
"url": "https://github.com/jkurian"
|
||||
},
|
||||
{
|
||||
"name": "Lewis Diamond",
|
||||
"email": "stromjs@lewisdiamond.com",
|
||||
"url": "https://github.com/lewisdiamond"
|
||||
"name": "lewis",
|
||||
"email": "lewis@jogogo.co"
|
||||
}
|
||||
],
|
||||
"license": "MIT",
|
||||
@ -29,16 +27,18 @@
|
||||
"files": [
|
||||
"dist"
|
||||
],
|
||||
"publishConfig": {
|
||||
"registry": "https://npm.dev.jogogo.co/"
|
||||
},
|
||||
"repository": {
|
||||
"url": "https://github.com/lewisdiamond/stromjs",
|
||||
"url": "git@github.com:Jogogoplay/mhysa.git",
|
||||
"type": "git"
|
||||
},
|
||||
"scripts": {
|
||||
"test": "ava",
|
||||
"lint": "tslint -p tsconfig.json",
|
||||
"validate:tslint": "tslint-config-prettier-check ./tslint.json",
|
||||
"prepublishOnly": "yarn lint && yarn test && yarn tsc -d",
|
||||
"prepare": "tsc"
|
||||
"prepublishOnly": "yarn lint && yarn test && yarn tsc -d"
|
||||
},
|
||||
"dependencies": {},
|
||||
"devDependencies": {
|
||||
@ -47,7 +47,7 @@
|
||||
"@types/sinon": "^7.0.13",
|
||||
"ava": "^2.4.0",
|
||||
"chai": "^4.2.0",
|
||||
"stromjs": "./",
|
||||
"mhysa": "./",
|
||||
"prettier": "^1.14.3",
|
||||
"sinon": "^7.4.2",
|
||||
"ts-node": "^8.3.0",
|
||||
|
@ -1,9 +1,8 @@
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
|
||||
strom
|
||||
.fromArray(["a", "b", "c"])
|
||||
.pipe(strom.map(s => Promise.resolve(s + s)))
|
||||
.pipe(strom.flatMap(s => Promise.resolve([s, s.toUpperCase()])))
|
||||
.pipe(strom.filter(s => Promise.resolve(s !== "bb")))
|
||||
.pipe(strom.join(","))
|
||||
Mhysa.fromArray(["a", "b", "c"])
|
||||
.pipe(Mhysa.map(s => Promise.resolve(s + s)))
|
||||
.pipe(Mhysa.flatMap(s => Promise.resolve([s, s.toUpperCase()])))
|
||||
.pipe(Mhysa.filter(s => Promise.resolve(s !== "bb")))
|
||||
.pipe(Mhysa.join(","))
|
||||
.pipe(process.stdout);
|
||||
|
@ -1,7 +1,6 @@
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
const catProcess = require("child_process").exec("grep -o ab");
|
||||
|
||||
strom
|
||||
.fromArray(["a", "b", "c"])
|
||||
.pipe(strom.child(catProcess))
|
||||
Mhysa.fromArray(["a", "b", "c"])
|
||||
.pipe(Mhysa.child(catProcess))
|
||||
.pipe(process.stdout);
|
||||
|
@ -1,6 +1,5 @@
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
|
||||
strom
|
||||
.fromArray(["a", "b", "c"])
|
||||
.pipe(strom.collect({ objectMode: true }))
|
||||
Mhysa.fromArray(["a", "b", "c"])
|
||||
.pipe(Mhysa.collect({ objectMode: true }))
|
||||
.on("data", object => console.log(object));
|
||||
|
@ -1,9 +1,9 @@
|
||||
const { Readable } = require("stream");
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
|
||||
const source1 = new Readable();
|
||||
const source2 = new Readable();
|
||||
strom.concat(source1, source2).pipe(process.stdout);
|
||||
Mhysa.concat(source1, source2).pipe(process.stdout);
|
||||
source1.push("a1 ");
|
||||
source2.push("c3 ");
|
||||
source1.push("b2 ");
|
||||
|
@ -1,7 +1,6 @@
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
const catProcess = require("child_process").exec("grep -o ab");
|
||||
|
||||
strom
|
||||
.fromArray(["a", "b", "c"])
|
||||
.pipe(strom.duplex(catProcess.stdin, catProcess.stdout))
|
||||
Mhysa.fromArray(["a", "b", "c"])
|
||||
.pipe(Mhysa.duplex(catProcess.stdin, catProcess.stdout))
|
||||
.pipe(process.stdout);
|
||||
|
@ -1,7 +1,6 @@
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
|
||||
strom
|
||||
.fromArray(["a", "b", "c"])
|
||||
.pipe(strom.filter(s => s !== "b"))
|
||||
.pipe(strom.join(","))
|
||||
Mhysa.fromArray(["a", "b", "c"])
|
||||
.pipe(Mhysa.filter(s => s !== "b"))
|
||||
.pipe(Mhysa.join(","))
|
||||
.pipe(process.stdout);
|
||||
|
@ -1,7 +1,6 @@
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
|
||||
strom
|
||||
.fromArray(["a", "AA"])
|
||||
.pipe(strom.flatMap(s => new Array(s.length).fill(s)))
|
||||
.pipe(strom.join(","))
|
||||
Mhysa.fromArray(["a", "AA"])
|
||||
.pipe(Mhysa.flatMap(s => new Array(s.length).fill(s)))
|
||||
.pipe(Mhysa.join(","))
|
||||
.pipe(process.stdout);
|
||||
|
@ -1,6 +1,5 @@
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
|
||||
strom
|
||||
.fromArray(["a", "b", "c"])
|
||||
.pipe(strom.join(","))
|
||||
Mhysa.fromArray(["a", "b", "c"])
|
||||
.pipe(Mhysa.join(","))
|
||||
.pipe(process.stdout);
|
||||
|
@ -1,7 +1,7 @@
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
|
||||
let f = async () => {
|
||||
const source = strom.fromArray(["a", "b", "c"]);
|
||||
console.log(await strom.last(source));
|
||||
const source = Mhysa.fromArray(["a", "b", "c"]);
|
||||
console.log(await Mhysa.last(source));
|
||||
};
|
||||
f();
|
||||
|
@ -1,7 +1,6 @@
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
|
||||
strom
|
||||
.fromArray(["a", "b"])
|
||||
.pipe(strom.map(s => s.toUpperCase()))
|
||||
.pipe(strom.join(","))
|
||||
Mhysa.fromArray(["a", "b"])
|
||||
.pipe(Mhysa.map(s => s.toUpperCase()))
|
||||
.pipe(Mhysa.join(","))
|
||||
.pipe(process.stdout);
|
||||
|
@ -1,9 +1,9 @@
|
||||
const { Readable } = require("stream");
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
|
||||
const source1 = new Readable({ read() {} });
|
||||
const source2 = new Readable({ read() {} });
|
||||
strom.merge(source1, source2).pipe(process.stdout);
|
||||
Mhysa.merge(source1, source2).pipe(process.stdout);
|
||||
source1.push("a1 ");
|
||||
setTimeout(() => source2.push("c3 "), 10);
|
||||
setTimeout(() => source1.push("b2 "), 20);
|
||||
|
@ -1,15 +0,0 @@
|
||||
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);
|
@ -1,6 +1,5 @@
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
|
||||
strom
|
||||
.fromArray(['{ "a": "b" }'])
|
||||
.pipe(strom.parse())
|
||||
Mhysa.fromArray(['{ "a": "b" }'])
|
||||
.pipe(Mhysa.parse())
|
||||
.on("data", object => console.log(object));
|
||||
|
@ -1,21 +0,0 @@
|
||||
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();
|
@ -1,7 +1,6 @@
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
|
||||
strom
|
||||
.fromArray(["a", "b", "cc"])
|
||||
.pipe(strom.reduce((acc, s) => ({ ...acc, [s]: s.length }), {}))
|
||||
.pipe(strom.stringify())
|
||||
Mhysa.fromArray(["a", "b", "cc"])
|
||||
.pipe(Mhysa.reduce((acc, s) => ({ ...acc, [s]: s.length }), {}))
|
||||
.pipe(Mhysa.stringify())
|
||||
.pipe(process.stdout);
|
||||
|
@ -1,6 +1,5 @@
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
|
||||
strom
|
||||
.fromArray(["a1", "b22", "c333"])
|
||||
.pipe(strom.replace(/b\d+/, "B"))
|
||||
Mhysa.fromArray(["a1", "b22", "c333"])
|
||||
.pipe(Mhysa.replace(/b\d+/, "B"))
|
||||
.pipe(process.stdout);
|
||||
|
@ -1,7 +1,6 @@
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
|
||||
strom
|
||||
.fromArray(["a,b", "c,d"])
|
||||
.pipe(strom.split(","))
|
||||
.pipe(strom.join("|"))
|
||||
Mhysa.fromArray(["a,b", "c,d"])
|
||||
.pipe(Mhysa.split(","))
|
||||
.pipe(Mhysa.join("|"))
|
||||
.pipe(process.stdout);
|
||||
|
@ -1,6 +1,5 @@
|
||||
const strom = require("stromjs");
|
||||
const Mhysa = require("mhysa");
|
||||
|
||||
strom
|
||||
.fromArray([{ a: "b" }])
|
||||
.pipe(strom.stringify())
|
||||
Mhysa.fromArray([{ a: "b" }])
|
||||
.pipe(Mhysa.stringify())
|
||||
.pipe(process.stdout);
|
||||
|
@ -2,7 +2,7 @@ import { Transform, TransformOptions } from "stream";
|
||||
|
||||
export function batch(
|
||||
batchSize: number = 1000,
|
||||
maxBatchAge: number = 0,
|
||||
maxBatchAge: number = 500,
|
||||
options: TransformOptions = {},
|
||||
): Transform {
|
||||
let buffer: any[] = [];
|
||||
@ -23,7 +23,7 @@ export function batch(
|
||||
buffer.push(chunk);
|
||||
if (buffer.length === batchSize) {
|
||||
sendChunk(this);
|
||||
} else if (maxBatchAge) {
|
||||
} else {
|
||||
if (timer === null) {
|
||||
timer = setInterval(() => {
|
||||
sendChunk(this);
|
||||
|
@ -6,45 +6,36 @@ type DemuxStreams = NodeJS.WritableStream | NodeJS.ReadWriteStream;
|
||||
|
||||
export interface DemuxOptions extends DuplexOptions {
|
||||
remultiplex?: boolean;
|
||||
purgeIdleInterval?: number;
|
||||
maxIdleTime?: number;
|
||||
}
|
||||
|
||||
export function demux(
|
||||
pipelineConstructor: (
|
||||
destKey?: string,
|
||||
chunk?: any,
|
||||
) => DemuxStreams | DemuxStreams[],
|
||||
construct: (destKey?: string, chunk?: any) => DemuxStreams,
|
||||
demuxBy: string | ((chunk: any) => string),
|
||||
options?: DemuxOptions,
|
||||
): Duplex {
|
||||
return new Demux(pipelineConstructor, demuxBy, options);
|
||||
return new Demux(construct, demuxBy, options);
|
||||
}
|
||||
|
||||
class Demux extends Duplex {
|
||||
private streamsByKey: {
|
||||
[key: string]: DemuxStreams[];
|
||||
[key: string]: { stream: DemuxStreams; lastWrite: number };
|
||||
};
|
||||
private demuxer: (chunk: any) => string;
|
||||
private pipelineConstructor: (
|
||||
destKey?: string,
|
||||
chunk?: any,
|
||||
) => DemuxStreams[];
|
||||
private construct: (destKey?: string, chunk?: any) => DemuxStreams;
|
||||
private remultiplex: boolean;
|
||||
private transform: Transform;
|
||||
private maxIdleTime: number;
|
||||
constructor(
|
||||
pipelineConstructor: (
|
||||
destKey?: string,
|
||||
chunk?: any,
|
||||
) => DemuxStreams | DemuxStreams[],
|
||||
construct: (destKey?: string) => DemuxStreams,
|
||||
demuxBy: string | ((chunk: any) => string),
|
||||
options: DemuxOptions = {},
|
||||
) {
|
||||
super(options);
|
||||
this.demuxer =
|
||||
typeof demuxBy === "string" ? chunk => chunk[demuxBy] : demuxBy;
|
||||
this.pipelineConstructor = (destKey: string, chunk?: any) => {
|
||||
const pipeline = pipelineConstructor(destKey, chunk);
|
||||
return Array.isArray(pipeline) ? pipeline : [pipeline];
|
||||
};
|
||||
this.construct = construct;
|
||||
this.remultiplex =
|
||||
options.remultiplex === undefined ? true : options.remultiplex;
|
||||
this.streamsByKey = {};
|
||||
@ -55,77 +46,74 @@ class Demux extends Duplex {
|
||||
cb(null);
|
||||
},
|
||||
});
|
||||
this.maxIdleTime = options.maxIdleTime || 600000;
|
||||
const purgeIdleInterval = options.purgeIdleInterval || 600000;
|
||||
setInterval(() => {
|
||||
this._destroyIdle();
|
||||
}, purgeIdleInterval);
|
||||
|
||||
this.on("unpipe", () => this._flush());
|
||||
}
|
||||
|
||||
private _destroyIdle() {
|
||||
for (let key in this.streamsByKey) {
|
||||
const curTime = Date.now();
|
||||
const pipeline = this.streamsByKey[key];
|
||||
|
||||
if (curTime - pipeline.lastWrite > this.maxIdleTime) {
|
||||
delete this.streamsByKey[key];
|
||||
}
|
||||
}
|
||||
}
|
||||
// tslint:disable-next-line
|
||||
public _read(size: number) {}
|
||||
|
||||
public async _write(chunk: any, encoding: any, cb: any) {
|
||||
const destKey = this.demuxer(chunk);
|
||||
if (this.streamsByKey[destKey] === undefined) {
|
||||
const newPipelines = this.pipelineConstructor(destKey, chunk);
|
||||
this.streamsByKey[destKey] = newPipelines;
|
||||
|
||||
newPipelines.forEach(newPipeline => {
|
||||
if (this.remultiplex && isReadable(newPipeline)) {
|
||||
(newPipeline as NodeJS.ReadWriteStream).pipe(
|
||||
this.transform,
|
||||
);
|
||||
} else if (this.remultiplex) {
|
||||
console.error(
|
||||
`Pipeline construct for ${destKey} does not implement readable interface`,
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
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();
|
||||
});
|
||||
}),
|
||||
const newPipeline = await this.construct(destKey, chunk);
|
||||
this.streamsByKey[destKey] = {
|
||||
stream: newPipeline,
|
||||
lastWrite: Date.now(),
|
||||
};
|
||||
if (this.remultiplex && isReadable(newPipeline)) {
|
||||
(newPipeline as NodeJS.ReadWriteStream).pipe(this.transform);
|
||||
} else if (this.remultiplex) {
|
||||
console.error(
|
||||
`Pipeline construct for ${destKey} does not implement readable interface`,
|
||||
);
|
||||
}
|
||||
});
|
||||
await Promise.all(pendingDrains);
|
||||
cb();
|
||||
} else {
|
||||
this.streamsByKey[destKey].lastWrite = Date.now();
|
||||
}
|
||||
|
||||
if (!this.streamsByKey[destKey].stream.write(chunk, encoding)) {
|
||||
this.streamsByKey[destKey].stream.once("drain", () => {
|
||||
cb();
|
||||
});
|
||||
} else {
|
||||
cb();
|
||||
}
|
||||
}
|
||||
|
||||
public _flush() {
|
||||
const pipelines: DemuxStreams[] = Array.prototype.concat.apply(
|
||||
[],
|
||||
Object.values(this.streamsByKey),
|
||||
);
|
||||
const flushPromises: Array<Promise<void>> = [];
|
||||
pipelines.forEach(pipeline => {
|
||||
flushPromises.push(
|
||||
new Promise(resolve => {
|
||||
pipeline.once("end", () => {
|
||||
resolve();
|
||||
});
|
||||
}),
|
||||
);
|
||||
});
|
||||
pipelines.forEach(pipeline => pipeline.end());
|
||||
Promise.all(flushPromises).then(() => {
|
||||
this.push(null);
|
||||
this.emit("end");
|
||||
const pipelines = Object.values(this.streamsByKey);
|
||||
let totalEnded = 0;
|
||||
pipelines.forEach(({ stream }) => {
|
||||
stream.once("end", () => {
|
||||
totalEnded++;
|
||||
if (pipelines.length === totalEnded) {
|
||||
this.push(null);
|
||||
this.emit("end");
|
||||
}
|
||||
});
|
||||
});
|
||||
pipelines.forEach(({ stream }) => stream.end());
|
||||
}
|
||||
|
||||
public _destroy(error: any, cb: (error?: any) => void) {
|
||||
const pipelines: DemuxStreams[] = [].concat.apply(
|
||||
[],
|
||||
Object.values(this.streamsByKey),
|
||||
);
|
||||
pipelines.forEach(p => (p as any).destroy());
|
||||
const pipelines = Object.values(this.streamsByKey);
|
||||
pipelines.forEach(({ stream }) => (stream as any).destroy());
|
||||
cb(error);
|
||||
}
|
||||
}
|
||||
|
@ -28,7 +28,7 @@ import { unbatch } from "./unbatch";
|
||||
import { compose } from "./compose";
|
||||
import { demux } from "./demux";
|
||||
|
||||
export function strom(defaultOptions: TransformOptions = { objectMode: true }) {
|
||||
export default function mhysa(defaultOptions?: TransformOptions) {
|
||||
function withDefaultOptions<T extends any[], R>(
|
||||
n: number,
|
||||
fn: (...args: T) => R,
|
||||
@ -256,17 +256,15 @@ export function strom(defaultOptions: TransformOptions = { objectMode: true }) {
|
||||
|
||||
/**
|
||||
* Composes multiple streams together. Writing occurs on first stream, piping occurs from last stream.
|
||||
* @param construct Constructor for new output source. Should return a Writable or ReadWrite stream.
|
||||
* @param demuxBy
|
||||
* @param demuxBy.key? Key to fetch value from source chunks to demultiplex source.
|
||||
* @param demuxBy.keyBy? Function to fetch value from source chunks to demultiplex source.
|
||||
* @param options Writable stream options
|
||||
* @param {Function} construct Constructor for new output source. Should return a Writable or ReadWrite stream.
|
||||
* @param {String | Function} demuxBy
|
||||
* @param {string} demuxBy.key? Key to fetch value from source chunks to demultiplex source.
|
||||
* @param {Function} demuxBy.keyBy? Function to fetch value from source chunks to demultiplex source.
|
||||
* @param {Object} options Demux stream options
|
||||
* @param {boolean} options.remultiplex? If demux should be remultiplexed into a single destination
|
||||
* @param {number} options.purgeIdleInterval? Interval at which a purge for idle pipelines will occur
|
||||
* @param {number} options.maxIdleTime? Min time a demuxed pipeline must be idle for to be purged
|
||||
*/
|
||||
demux: withDefaultOptions(2, demux),
|
||||
|
||||
/**
|
||||
* Create a new strom instance overriding the defaults
|
||||
*/
|
||||
instance: strom,
|
||||
};
|
||||
}
|
||||
|
@ -2,41 +2,20 @@ 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 & RateOptions,
|
||||
options?: TransformOptions,
|
||||
): Transform {
|
||||
const deltaMS = ((1 / targetRate) * 1000) / period; // Skip a full period
|
||||
let total = 0;
|
||||
const window = options?.window || Infinity;
|
||||
const behavior = options?.behavior || Behavior.BUFFER;
|
||||
let start = performance.now();
|
||||
const start = performance.now();
|
||||
return new Transform({
|
||||
...options,
|
||||
async transform(data, encoding, callback) {
|
||||
const now = performance.now();
|
||||
if (now - start >= window) {
|
||||
start = now - window;
|
||||
}
|
||||
const currentRate = (total / (now - start)) * 1000;
|
||||
const currentRate = (total / (performance.now() - start)) * 1000;
|
||||
if (targetRate && currentRate > targetRate) {
|
||||
if (behavior === Behavior.DROP) {
|
||||
callback(undefined);
|
||||
return;
|
||||
} else {
|
||||
await sleep(deltaMS);
|
||||
}
|
||||
await sleep(deltaMS);
|
||||
}
|
||||
total += 1;
|
||||
callback(undefined, data);
|
||||
|
35
src/index.ts
35
src/index.ts
@ -1,29 +1,6 @@
|
||||
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();
|
||||
import mhysa from "./functions";
|
||||
import * as _utils from "./utils";
|
||||
export default mhysa;
|
||||
|
||||
// @TODO fix this with proper import export
|
||||
export const utils = { ..._utils };
|
||||
|
@ -1,9 +1,10 @@
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { Readable } from "stream";
|
||||
import { accumulator, accumulatorBy } from "../src";
|
||||
import mhysa from "../src";
|
||||
import { FlushStrategy } from "../src/functions/accumulator";
|
||||
import { performance } from "perf_hooks";
|
||||
const { accumulator, accumulatorBy } = mhysa({ objectMode: true });
|
||||
|
||||
test.cb("accumulator() rolling", t => {
|
||||
t.plan(3);
|
||||
@ -13,14 +14,8 @@ test.cb("accumulator() rolling", t => {
|
||||
key: string;
|
||||
}
|
||||
const source = new Readable({ objectMode: true });
|
||||
const firstFlush = [
|
||||
{ ts: 0, key: "a" },
|
||||
{ ts: 1, key: "b" },
|
||||
];
|
||||
const secondFlush = [
|
||||
{ ts: 2, key: "d" },
|
||||
{ ts: 3, key: "e" },
|
||||
];
|
||||
const firstFlush = [{ 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 flushes = [firstFlush, secondFlush, thirdFlush];
|
||||
|
||||
@ -93,10 +88,7 @@ test.cb(
|
||||
"nonExistingKey",
|
||||
{ objectMode: true },
|
||||
);
|
||||
const input = [
|
||||
{ ts: 0, key: "a" },
|
||||
{ ts: 1, key: "b" },
|
||||
];
|
||||
const input = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }];
|
||||
|
||||
source
|
||||
.pipe(accumulatorStream)
|
||||
@ -190,10 +182,7 @@ test.cb("accumulator() sliding", t => {
|
||||
{ ts: 4, key: "d" },
|
||||
];
|
||||
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 = [
|
||||
{ ts: 0, key: "a" },
|
||||
{ ts: 1, key: "b" },
|
||||
@ -243,10 +232,7 @@ test.cb("accumulator() sliding with key", t => {
|
||||
{ ts: 6, key: "g" },
|
||||
];
|
||||
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 = [
|
||||
{ ts: 0, key: "a" },
|
||||
{ ts: 1, key: "b" },
|
||||
@ -257,14 +243,8 @@ test.cb("accumulator() sliding with key", t => {
|
||||
{ ts: 2, key: "c" },
|
||||
{ ts: 3, key: "d" },
|
||||
];
|
||||
const fifthFlush = [
|
||||
{ ts: 3, key: "d" },
|
||||
{ ts: 5, key: "f" },
|
||||
];
|
||||
const sixthFlush = [
|
||||
{ ts: 5, key: "f" },
|
||||
{ ts: 6, key: "g" },
|
||||
];
|
||||
const fifthFlush = [{ ts: 3, key: "d" }, { ts: 5, key: "f" }];
|
||||
const sixthFlush = [{ ts: 5, key: "f" }, { ts: 6, key: "g" }];
|
||||
|
||||
const flushes = [
|
||||
firstFlush,
|
||||
@ -306,10 +286,7 @@ test.cb(
|
||||
"nonExistingKey",
|
||||
{ objectMode: true },
|
||||
);
|
||||
const input = [
|
||||
{ ts: 0, key: "a" },
|
||||
{ ts: 1, key: "b" },
|
||||
];
|
||||
const input = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }];
|
||||
|
||||
source
|
||||
.pipe(accumulatorStream)
|
||||
@ -357,22 +334,10 @@ test.cb(
|
||||
{ ts: 6, key: "g" },
|
||||
];
|
||||
const firstFlush = [{ ts: 0, key: "a" }];
|
||||
const secondFlush = [
|
||||
{ ts: 0, key: "a" },
|
||||
{ ts: 2, key: "c" },
|
||||
];
|
||||
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 secondFlush = [{ ts: 0, key: "a" }, { ts: 2, key: "c" }];
|
||||
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 = [
|
||||
firstFlush,
|
||||
@ -504,10 +469,7 @@ test.cb("accumulatorBy() sliding", t => {
|
||||
{ ts: 6, key: "g" },
|
||||
];
|
||||
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 = [
|
||||
{ ts: 0, key: "a" },
|
||||
{ ts: 1, key: "b" },
|
||||
@ -518,14 +480,8 @@ test.cb("accumulatorBy() sliding", t => {
|
||||
{ ts: 2, key: "c" },
|
||||
{ ts: 3, key: "d" },
|
||||
];
|
||||
const fifthFlush = [
|
||||
{ ts: 3, key: "d" },
|
||||
{ ts: 5, key: "f" },
|
||||
];
|
||||
const sixthFlush = [
|
||||
{ ts: 5, key: "f" },
|
||||
{ ts: 6, key: "g" },
|
||||
];
|
||||
const fifthFlush = [{ ts: 3, key: "d" }, { ts: 5, key: "f" }];
|
||||
const sixthFlush = [{ ts: 5, key: "f" }, { ts: 6, key: "g" }];
|
||||
|
||||
const flushes = [
|
||||
firstFlush,
|
||||
|
@ -1,7 +1,8 @@
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { batch, map, fromArray } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { batch, map, fromArray } = mhysa({ objectMode: true });
|
||||
|
||||
test.cb("batch() batches chunks together", t => {
|
||||
t.plan(3);
|
||||
@ -38,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, 500))
|
||||
.pipe(batch(3))
|
||||
.on("data", (element: string[]) => {
|
||||
t.deepEqual(element, expectedElements[i]);
|
||||
i++;
|
||||
|
@ -2,7 +2,8 @@ import * as cp from "child_process";
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { child } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { child } = mhysa();
|
||||
|
||||
test.cb(
|
||||
"child() allows easily writing to child process stdin and reading from its stdout",
|
||||
|
@ -1,7 +1,8 @@
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { collect } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { collect } = mhysa();
|
||||
|
||||
test.cb(
|
||||
"collect() collects streamed elements into an array (object, flowing mode)",
|
||||
@ -58,7 +59,7 @@ test.cb(
|
||||
const source = new Readable({ objectMode: false });
|
||||
|
||||
source
|
||||
.pipe(collect({ objectMode: false }))
|
||||
.pipe(collect())
|
||||
.on("data", collected => {
|
||||
expect(collected).to.deep.equal(Buffer.from("abc"));
|
||||
t.pass();
|
||||
|
@ -2,8 +2,9 @@ import * as test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { sleep } from "../src/helpers";
|
||||
import { Readable, Writable } from "stream";
|
||||
import { compose, map, fromArray } from "../src";
|
||||
import mhysa from "../src";
|
||||
import { performance } from "perf_hooks";
|
||||
const { compose, map, fromArray } = mhysa({ objectMode: true });
|
||||
|
||||
test.cb("compose() chains two streams together in the correct order", t => {
|
||||
t.plan(3);
|
||||
@ -113,9 +114,13 @@ test("compose() writable length should be less than highWaterMark when handing w
|
||||
return chunk;
|
||||
});
|
||||
|
||||
const composed = compose([first, second], undefined, {
|
||||
highWaterMark: 2,
|
||||
});
|
||||
const composed = compose(
|
||||
[first, second],
|
||||
undefined,
|
||||
{
|
||||
highWaterMark: 2,
|
||||
},
|
||||
);
|
||||
composed.on("error", err => {
|
||||
reject();
|
||||
});
|
||||
@ -163,9 +168,13 @@ test("compose() should emit drain event ~rate * highWaterMark ms for every write
|
||||
return chunk;
|
||||
});
|
||||
|
||||
const composed = compose([first, second], undefined, {
|
||||
highWaterMark,
|
||||
});
|
||||
const composed = compose(
|
||||
[first, second],
|
||||
undefined,
|
||||
{
|
||||
highWaterMark,
|
||||
},
|
||||
);
|
||||
composed.on("error", err => {
|
||||
reject();
|
||||
});
|
||||
@ -212,9 +221,13 @@ test.cb(
|
||||
return chunk;
|
||||
});
|
||||
|
||||
const composed = compose([first, second], undefined, {
|
||||
highWaterMark: 5,
|
||||
});
|
||||
const composed = compose(
|
||||
[first, second],
|
||||
undefined,
|
||||
{
|
||||
highWaterMark: 5,
|
||||
},
|
||||
);
|
||||
|
||||
composed.on("error", err => {
|
||||
t.end(err);
|
||||
@ -271,9 +284,13 @@ test.cb(
|
||||
{ highWaterMark: 1 },
|
||||
);
|
||||
|
||||
const composed = compose([first, second], undefined, {
|
||||
highWaterMark: 5,
|
||||
});
|
||||
const composed = compose(
|
||||
[first, second],
|
||||
undefined,
|
||||
{
|
||||
highWaterMark: 5,
|
||||
},
|
||||
);
|
||||
composed.on("error", err => {
|
||||
t.end(err);
|
||||
});
|
||||
@ -338,9 +355,13 @@ test.cb(
|
||||
{ highWaterMark: 2 },
|
||||
);
|
||||
|
||||
const composed = compose([first, second], undefined, {
|
||||
highWaterMark: 5,
|
||||
});
|
||||
const composed = compose(
|
||||
[first, second],
|
||||
undefined,
|
||||
{
|
||||
highWaterMark: 5,
|
||||
},
|
||||
);
|
||||
composed.on("error", err => {
|
||||
t.end(err);
|
||||
});
|
||||
@ -393,9 +414,13 @@ test.cb(
|
||||
return chunk;
|
||||
});
|
||||
|
||||
const composed = compose([first, second], undefined, {
|
||||
highWaterMark: 6,
|
||||
});
|
||||
const composed = compose(
|
||||
[first, second],
|
||||
undefined,
|
||||
{
|
||||
highWaterMark: 6,
|
||||
},
|
||||
);
|
||||
|
||||
composed.on("error", err => {
|
||||
t.end(err);
|
||||
@ -445,9 +470,12 @@ test.cb("compose() should be 'destroyable'", t => {
|
||||
return chunk;
|
||||
});
|
||||
|
||||
const composed = compose([first, second], (err: any) => {
|
||||
t.pass();
|
||||
});
|
||||
const composed = compose(
|
||||
[first, second],
|
||||
(err: any) => {
|
||||
t.pass();
|
||||
},
|
||||
);
|
||||
|
||||
const fakeSource = new Readable({
|
||||
objectMode: true,
|
||||
@ -503,9 +531,13 @@ test.cb("compose() `finish` and `end` propagates", t => {
|
||||
return chunk;
|
||||
});
|
||||
|
||||
const composed = compose([first, second], undefined, {
|
||||
highWaterMark: 3,
|
||||
});
|
||||
const composed = compose(
|
||||
[first, second],
|
||||
undefined,
|
||||
{
|
||||
highWaterMark: 3,
|
||||
},
|
||||
);
|
||||
|
||||
const fakeSource = new Readable({
|
||||
objectMode: true,
|
||||
|
@ -1,7 +1,8 @@
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { concat, collect } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { concat, collect } = mhysa();
|
||||
|
||||
test.cb(
|
||||
"concat() concatenates multiple readable streams (object, flowing mode)",
|
||||
@ -171,7 +172,7 @@ test.cb(
|
||||
test.cb("concat() concatenates empty list of readable streams", t => {
|
||||
t.plan(0);
|
||||
concat()
|
||||
.pipe(collect({ objectMode: false }))
|
||||
.pipe(collect())
|
||||
.on("data", _ => {
|
||||
t.fail();
|
||||
})
|
||||
|
@ -1,30 +1,21 @@
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { batch as _batch, instance as strom } from "../src";
|
||||
import mhysa from "../src";
|
||||
|
||||
const withDefaultOptions = strom({ objectMode: false });
|
||||
const withDefaultOptions = mhysa({ objectMode: true });
|
||||
const withoutOptions = mhysa();
|
||||
|
||||
test("strom instances can have default options", t => {
|
||||
test("Mhysa instances can have default options", t => {
|
||||
let batch = withDefaultOptions.batch();
|
||||
t.false(batch._readableState.objectMode);
|
||||
t.false(batch._writableState.objectMode);
|
||||
t.true(batch._readableState.objectMode);
|
||||
t.true(batch._writableState.objectMode);
|
||||
batch = withDefaultOptions.batch(3);
|
||||
t.false(batch._readableState.objectMode);
|
||||
t.false(batch._writableState.objectMode);
|
||||
t.true(batch._readableState.objectMode);
|
||||
t.true(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 = _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 });
|
||||
batch = withDefaultOptions.batch(3, 1, { objectMode: false });
|
||||
t.false(batch._readableState.objectMode);
|
||||
t.false(batch._writableState.objectMode);
|
||||
});
|
||||
|
@ -1,10 +1,11 @@
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { demux, map, fromArray } from "../src";
|
||||
import mhysa from "../src";
|
||||
import { Writable, Readable } from "stream";
|
||||
import * as sinon from "sinon";
|
||||
import { sleep } from "../src/helpers";
|
||||
const sinon = require("sinon");
|
||||
const { sleep } = require("../src/helpers");
|
||||
import { performance } from "perf_hooks";
|
||||
const { demux, map, fromArray } = mhysa({ objectMode: true });
|
||||
|
||||
interface Test {
|
||||
key: string;
|
||||
@ -861,3 +862,39 @@ test.cb("demux() should be 'destroyable'", t => {
|
||||
fakeSource.push(input[3]);
|
||||
fakeSource.push(input[4]);
|
||||
});
|
||||
|
||||
test.cb("Should delete idle pipelines", t => {
|
||||
t.plan(5);
|
||||
const input = [
|
||||
{ key: "a", visited: [] },
|
||||
{ key: "b", visited: [] },
|
||||
{ key: "b", visited: [] },
|
||||
{ key: "a", visited: [] },
|
||||
{ key: "b", visited: [] },
|
||||
{ key: "c", visited: [] },
|
||||
];
|
||||
const construct = sinon.spy((destKey: string) => {
|
||||
const dest = map((chunk: Test) => {
|
||||
chunk.visited.push(1);
|
||||
return chunk;
|
||||
});
|
||||
t.pass();
|
||||
|
||||
return dest;
|
||||
});
|
||||
|
||||
const demuxed = demux(construct, "key", {
|
||||
maxIdleTime: 110,
|
||||
purgeIdleInterval: 110,
|
||||
});
|
||||
|
||||
demuxed.on("data", data => {
|
||||
if (data.key === "c") t.end();
|
||||
});
|
||||
|
||||
for (let i = 0; i < input.length; i++) {
|
||||
setTimeout(() => {
|
||||
demuxed.write(input[i]);
|
||||
}, i * 100);
|
||||
}
|
||||
});
|
||||
|
@ -2,7 +2,8 @@ import * as cp from "child_process";
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { duplex } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { duplex } = mhysa();
|
||||
|
||||
test.cb(
|
||||
"duplex() combines a writable and readable stream into a ReadWrite stream",
|
||||
|
@ -1,7 +1,8 @@
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { Readable } from "stream";
|
||||
import { filter } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { filter } = mhysa();
|
||||
|
||||
test.cb("filter() filters elements synchronously", t => {
|
||||
t.plan(2);
|
||||
|
@ -1,7 +1,8 @@
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { flatMap } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { flatMap } = mhysa({ objectMode: true });
|
||||
|
||||
test.cb("flatMap() maps elements synchronously", t => {
|
||||
t.plan(6);
|
||||
|
@ -1,6 +1,7 @@
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { fromArray } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { fromArray } = mhysa();
|
||||
|
||||
test.cb("fromArray() streams array elements in flowing mode", t => {
|
||||
t.plan(3);
|
||||
|
@ -1,7 +1,8 @@
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { join } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { join } = mhysa();
|
||||
|
||||
test.cb("join() joins chunks using the specified separator", t => {
|
||||
t.plan(9);
|
||||
|
@ -1,7 +1,8 @@
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { last } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { last } = mhysa();
|
||||
|
||||
test("last() resolves to the last chunk streamed by the given readable stream", async t => {
|
||||
const source = new Readable({ objectMode: true });
|
||||
|
@ -1,7 +1,8 @@
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { map } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { map } = mhysa();
|
||||
|
||||
test.cb("map() maps elements synchronously", t => {
|
||||
t.plan(3);
|
||||
|
@ -1,7 +1,8 @@
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { merge } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { merge } = mhysa();
|
||||
|
||||
test.cb(
|
||||
"merge() merges multiple readable streams in chunk arrival order",
|
||||
|
@ -2,8 +2,9 @@ import { Readable } from "stream";
|
||||
import { performance } from "perf_hooks";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { parallelMap } from "../src";
|
||||
import mhysa from "../src";
|
||||
import { sleep } from "../src/helpers";
|
||||
const { parallelMap } = mhysa({ objectMode: true });
|
||||
|
||||
test.cb("parallelMap() parallel mapping", t => {
|
||||
t.plan(6);
|
||||
|
@ -1,7 +1,8 @@
|
||||
import { Readable, finished } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { parse } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { parse } = mhysa();
|
||||
|
||||
test.cb("parse() parses the streamed elements as JSON", t => {
|
||||
t.plan(3);
|
||||
|
@ -1,11 +1,12 @@
|
||||
import { Readable } from "stream";
|
||||
import { performance } from "perf_hooks";
|
||||
import test from "ava";
|
||||
import { rate } from "../src";
|
||||
import { sleep } from "../src/helpers";
|
||||
import { expect } from "chai";
|
||||
import mhysa from "../src";
|
||||
const { rate } = mhysa({ objectMode: true });
|
||||
|
||||
test.cb("rate() sends data at a rate of 150", t => {
|
||||
t.plan(15);
|
||||
t.plan(5);
|
||||
const targetRate = 150;
|
||||
const source = new Readable({ objectMode: true });
|
||||
const expectedElements = ["a", "b", "c", "d", "e"];
|
||||
@ -14,10 +15,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;
|
||||
t.is(element, expectedElements[i]);
|
||||
t.true(currentRate <= targetRate);
|
||||
expect(element).to.deep.equal(expectedElements[i]);
|
||||
expect(currentRate).lessThan(targetRate);
|
||||
t.pass();
|
||||
i++;
|
||||
})
|
||||
@ -33,7 +34,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(15);
|
||||
t.plan(5);
|
||||
const targetRate = 50;
|
||||
const source = new Readable({ objectMode: true });
|
||||
const expectedElements = ["a", "b", "c", "d", "e"];
|
||||
@ -42,10 +43,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;
|
||||
t.is(element, expectedElements[i]);
|
||||
t.true(currentRate <= targetRate);
|
||||
expect(element).to.deep.equal(expectedElements[i]);
|
||||
expect(currentRate).lessThan(targetRate);
|
||||
t.pass();
|
||||
i++;
|
||||
})
|
||||
@ -61,7 +62,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(15);
|
||||
t.plan(5);
|
||||
const targetRate = 1;
|
||||
const source = new Readable({ objectMode: true });
|
||||
const expectedElements = ["a", "b", "c", "d", "e"];
|
||||
@ -70,10 +71,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;
|
||||
t.is(element, expectedElements[i]);
|
||||
t.true(currentRate <= targetRate);
|
||||
expect(element).to.deep.equal(expectedElements[i]);
|
||||
expect(currentRate).lessThan(targetRate);
|
||||
t.pass();
|
||||
i++;
|
||||
})
|
||||
@ -87,41 +88,3 @@ 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);
|
||||
});
|
||||
|
@ -1,7 +1,8 @@
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { reduce } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { reduce } = mhysa({ objectMode: true });
|
||||
|
||||
test.cb("reduce() reduces elements synchronously", t => {
|
||||
t.plan(1);
|
||||
|
@ -1,7 +1,8 @@
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { replace } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { replace } = mhysa();
|
||||
|
||||
test.cb(
|
||||
"replace() replaces occurrences of the given string in the streamed elements with the specified " +
|
||||
|
@ -1,7 +1,8 @@
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { split } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { split } = mhysa();
|
||||
|
||||
test.cb("split() splits chunks using the default separator (\\n)", t => {
|
||||
t.plan(5);
|
||||
|
@ -1,7 +1,8 @@
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { stringify } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { stringify } = mhysa();
|
||||
|
||||
test.cb("stringify() stringifies the streamed elements as JSON", t => {
|
||||
t.plan(4);
|
||||
|
@ -1,7 +1,8 @@
|
||||
import { Readable } from "stream";
|
||||
import test from "ava";
|
||||
import { expect } from "chai";
|
||||
import { unbatch, batch } from "../src";
|
||||
import mhysa from "../src";
|
||||
const { unbatch, batch } = mhysa({ objectMode: true });
|
||||
|
||||
test.cb("unbatch() unbatches", t => {
|
||||
t.plan(3);
|
||||
|
@ -1,6 +1,7 @@
|
||||
import test from "ava";
|
||||
import { collected } from "../../src/utils";
|
||||
import { fromArray, collect } from "../../src";
|
||||
import mhysa from "../../src";
|
||||
const { fromArray, collect } = mhysa({ objectMode: true });
|
||||
|
||||
test("collected returns a promise for the first data point", async t => {
|
||||
const data = collected(fromArray([1, 2, 3, 4]).pipe(collect()));
|
||||
|
@ -1934,6 +1934,9 @@ merge2@^1.2.3, merge2@^1.3.0:
|
||||
resolved "https://npm.dev.jogogo.co/merge2/-/merge2-1.3.0.tgz#5b366ee83b2f1582c48f87e47cf1a9352103ca81"
|
||||
integrity sha512-2j4DAdlBOkiSZIsaXk4mTE3sRS02yBHAtfy127xRV3bQUFqXkjHCHLW6Scv7DwNRbIWNHH8zpnz9zMaKXIdvYw==
|
||||
|
||||
mhysa@./:
|
||||
version "2.0.0-alpha.1"
|
||||
|
||||
micromatch@^4.0.2:
|
||||
version "4.0.2"
|
||||
resolved "https://npm.dev.jogogo.co/micromatch/-/micromatch-4.0.2.tgz#4fcb0999bf9fbc2fcbdd212f6d629b9a56c39259"
|
||||
@ -2716,9 +2719,6 @@ strip-json-comments@~2.0.1:
|
||||
resolved "https://registry.yarnpkg.com/strip-json-comments/-/strip-json-comments-2.0.1.tgz#3c531942e908c2697c0ec344858c286c7ca0a60a"
|
||||
integrity sha1-PFMZQukIwml8DsNEhYwobHygpgo=
|
||||
|
||||
stromjs@./:
|
||||
version "0.5.1"
|
||||
|
||||
supertap@^1.0.0:
|
||||
version "1.0.0"
|
||||
resolved "https://registry.yarnpkg.com/supertap/-/supertap-1.0.0.tgz#bd9751c7fafd68c68cf8222a29892206a119fa9e"
|
||||
|
Loading…
Reference in New Issue
Block a user