4 Commits

Author SHA1 Message Date
Jerry Kurian
3c75ef88b4 Minor changes to types 2020-05-11 21:13:03 -04:00
Jerry Kurian
7113400cb1 version 2020-05-11 10:33:17 -04:00
Jerry Kurian
a42560edfc Rename construct 2020-05-08 16:21:20 -04:00
Jerry Kurian
58a95a91d0 Demux handles arrays by key 2020-05-08 15:58:31 -04:00
46 changed files with 236 additions and 241 deletions

View File

@@ -1,13 +1,15 @@
# Strom # Mhysa
**Dependency-free stream utils for Node.js** **Dependency-free stream utils for Node.js**
<sub>Released under the [MIT](https://git.lewis.id/strom/blob/master/LICENSE) license.</sub> <sub>Released under the [MIT](https://github.com/Wenzil/Mhysa/blob/master/LICENSE) license.</sub>
```sh ```sh
yarn add strom // Name TBD yarn add mhysa
``` ```
<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
@@ -16,7 +18,7 @@ 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
strom.fromArray(["a", "b"]) Mhysa.fromArray(["a", "b"])
.pipe(process.stdout); .pipe(process.stdout);
// ab is printed out // ab is printed out
``` ```
@@ -33,8 +35,8 @@ 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
strom.fromArray(["a", "b"]) Mhysa.fromArray(["a", "b"])
.pipe(strom.map(s => s.toUpperCase())) .pipe(Mhysa.map(s => s.toUpperCase()))
.pipe(process.stdout); .pipe(process.stdout);
// AB is printed out // AB is printed out
``` ```
@@ -51,8 +53,8 @@ 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
strom.fromArray(["a", "AA"]) Mhysa.fromArray(["a", "AA"])
.pipe(strom.flatMap(s => new Array(s.length).fill(s))) .pipe(Mhysa.flatMap(s => new Array(s.length).fill(s)))
.pipe(process.stdout); .pipe(process.stdout);
// aAAAA is printed out // aAAAA is printed out
``` ```
@@ -68,8 +70,8 @@ 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
strom.fromArray(["a", "b", "c"]) Mhysa.fromArray(["a", "b", "c"])
.pipe(strom.filter(s => s !== "b")) .pipe(Mhysa.filter(s => s !== "b"))
.pipe(process.stdout); .pipe(process.stdout);
// ac is printed out // ac is printed out
``` ```
@@ -88,9 +90,9 @@ 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
strom.fromArray(["a", "b", "cc"]) Mhysa.fromArray(["a", "b", "cc"])
.pipe(strom.reduce((acc, s) => ({ ...acc, [s]: s.length }), {})) .pipe(Mhysa.reduce((acc, s) => ({ ...acc, [s]: s.length }), {}))
.pipe(strom.stringify()) .pipe(Mhysa.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
``` ```
@@ -106,9 +108,9 @@ 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
strom.fromArray(["a,b", "c,d"]) Mhysa.fromArray(["a,b", "c,d"])
.pipe(strom.split(",")) .pipe(Mhysa.split(","))
.pipe(strom.join("|")) .pipe(Mhysa.join("|"))
.pipe(process.stdout); .pipe(process.stdout);
// a|bc|d is printed out // a|bc|d is printed out
``` ```
@@ -124,8 +126,8 @@ 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
strom.fromArray(["a", "b", "c"]) Mhysa.fromArray(["a", "b", "c"])
.pipe(strom.join(",")) .pipe(Mhysa.join(","))
.pipe(process.stdout); .pipe(process.stdout);
// a,b,c is printed out // a,b,c is printed out
``` ```
@@ -143,8 +145,8 @@ 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
strom.fromArray(["a1", "b22", "c333"]) Mhysa.fromArray(["a1", "b22", "c333"])
.pipe(strom.replace(/b\d+/, "B")) .pipe(Mhysa.replace(/b\d+/, "B"))
.pipe(process.stdout); .pipe(process.stdout);
// a1Bc333 is printed out // a1Bc333 is printed out
``` ```
@@ -154,8 +156,8 @@ strom.fromArray(["a1", "b22", "c333"])
Return a `ReadWrite` stream that parses the streamed chunks as JSON Return a `ReadWrite` stream that parses the streamed chunks as JSON
```js ```js
strom.fromArray(['{ "a": "b" }']) Mhysa.fromArray(['{ "a": "b" }'])
.pipe(strom.parse()) .pipe(Mhysa.parse())
.once("data", object => console.log(object)); .once("data", object => console.log(object));
// { a: 'b' } is printed out // { a: 'b' } is printed out
``` ```
@@ -165,8 +167,8 @@ strom.fromArray(['{ "a": "b" }'])
Return a `ReadWrite` stream that stringifies the streamed chunks to JSON Return a `ReadWrite` stream that stringifies the streamed chunks to JSON
```js ```js
strom.fromArray([{ a: "b" }]) Mhysa.fromArray([{ a: "b" }])
.pipe(strom.stringify()) .pipe(Mhysa.stringify())
.pipe(process.stdout); .pipe(process.stdout);
// {"a":"b"} is printed out // {"a":"b"} is printed out
``` ```
@@ -181,8 +183,8 @@ 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
strom.fromArray(["a", "b", "c"]) Mhysa.fromArray(["a", "b", "c"])
.pipe(strom.collect({ objectMode: true })) .pipe(Mhysa.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
``` ```
@@ -198,7 +200,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();
strom.concat(source1, source2).pipe(process.stdout) Mhysa.concat(source1, source2).pipe(process.stdout)
source1.push("a1 "); source1.push("a1 ");
source2.push("c3 "); source2.push("c3 ");
source1.push("b2 "); source1.push("b2 ");
@@ -219,7 +221,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() {} });
strom.merge(source1, source2).pipe(process.stdout); Mhysa.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);
@@ -241,8 +243,8 @@ 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");
strom.fromArray(["a", "b", "c"]) Mhysa.fromArray(["a", "b", "c"])
.pipe(strom.duplex(catProcess.stdin, catProcess.stdout)) .pipe(Mhysa.duplex(catProcess.stdin, catProcess.stdout))
.pipe(process.stdout); .pipe(process.stdout);
// ab is printed out // ab is printed out
``` ```
@@ -257,8 +259,8 @@ 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");
strom.fromArray(["a", "b", "c"]) Mhysa.fromArray(["a", "b", "c"])
.pipe(strom.child(catProcess)) .pipe(Mhysa.child(catProcess))
.pipe(process.stdout); .pipe(process.stdout);
// ab is printed out // ab is printed out
``` ```
@@ -274,8 +276,8 @@ ended
```js ```js
let f = async () => { let f = async () => {
const source = strom.fromArray(["a", "b", "c"]); const source = Mhysa.fromArray(["a", "b", "c"]);
console.log(await strom.last(source)); console.log(await Mhysa.last(source));
}; };
f(); f();
// c is printed out // c is printed out

View File

@@ -1,16 +1,17 @@
{ {
"name": "strom", "name": "@jogogo/mhysa",
"version": "2.0.0", "version": "2.0.0-alpha.4",
"description": "Streams utils for Node.js", "description": "Streams and event emitter utils for Node.js",
"keywords": [ "keywords": [
"promise", "promise",
"stream", "stream",
"event emitter",
"utils" "utils"
], ],
"contributors": [ "author": {
{
"name": "Wenzil" "name": "Wenzil"
}, },
"contributors": [
{ {
"name": "jerry", "name": "jerry",
"email": "jerry@jogogo.co" "email": "jerry@jogogo.co"
@@ -26,16 +27,18 @@
"files": [ "files": [
"dist" "dist"
], ],
"publishConfig": {
"registry": "https://npm.dev.jogogo.co/"
},
"repository": { "repository": {
"url": "git@git.lewis.id:ldiamond/strom.git", "url": "git@github.com:Jogogoplay/mhysa.git",
"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": {

View File

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

View File

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

View File

@@ -1,5 +1,5 @@
const strom = require("strom").strom(); const Mhysa = require("mhysa");
strom.fromArray(["a", "b", "c"]) Mhysa.fromArray(["a", "b", "c"])
.pipe(strom.collect({ objectMode: true })) .pipe(Mhysa.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 strom = require("strom").strom(); const Mhysa = require("mhysa");
const source1 = new Readable(); const source1 = new Readable();
const source2 = new Readable(); const source2 = new Readable();
strom.concat(source1, source2).pipe(process.stdout); Mhysa.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,6 @@
const strom = require("strom").strom(); const Mhysa = require("mhysa");
const catProcess = require("child_process").exec("grep -o ab"); const catProcess = require("child_process").exec("grep -o ab");
strom.fromArray(["a", "b", "c"]) Mhysa.fromArray(["a", "b", "c"])
.pipe(strom.duplex(catProcess.stdin, catProcess.stdout)) .pipe(Mhysa.duplex(catProcess.stdin, catProcess.stdout))
.pipe(process.stdout); .pipe(process.stdout);

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -1,9 +1,9 @@
const { Readable } = require("stream"); const { Readable } = require("stream");
const strom = require("strom").strom(); const Mhysa = require("mhysa");
const source1 = new Readable({ read() {} }); const source1 = new Readable({ read() {} });
const source2 = 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 "); source1.push("a1 ");
setTimeout(() => source2.push("c3 "), 10); setTimeout(() => source2.push("c3 "), 10);
setTimeout(() => source1.push("b2 "), 20); setTimeout(() => source1.push("b2 "), 20);

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -99,7 +99,7 @@ class Demux extends Duplex {
} }
public _flush() { public _flush() {
const pipelines: DemuxStreams[] = Array.prototype.concat.apply( const pipelines: DemuxStreams[] = [].concat.apply(
[], [],
Object.values(this.streamsByKey), Object.values(this.streamsByKey),
); );

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 function strom(defaultOptions: TransformOptions = { objectMode: true }) { export default function mhysa(defaultOptions?: TransformOptions) {
function withDefaultOptions<T extends any[], R>( function withDefaultOptions<T extends any[], R>(
n: number, n: number,
fn: (...args: T) => R, fn: (...args: T) => R,

View File

@@ -1,2 +1,6 @@
export { strom } from "./functions"; import mhysa from "./functions";
export * from "./utils"; import * as _utils from "./utils";
export default mhysa;
// @TODO fix this with proper import export
export const utils = { ..._utils };

View File

@@ -1,10 +1,10 @@
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 { strom } from "../src"; import mhysa 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 } = strom({ objectMode: true }); const { accumulator, accumulatorBy } = mhysa({ objectMode: true });
test.cb("accumulator() rolling", t => { test.cb("accumulator() rolling", t => {
t.plan(3); t.plan(3);
@@ -14,14 +14,8 @@ test.cb("accumulator() rolling", t => {
key: string; key: string;
} }
const source = new Readable({ objectMode: true }); const source = new Readable({ objectMode: true });
const firstFlush = [ const firstFlush = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }];
{ ts: 0, key: "a" }, const secondFlush = [{ ts: 2, key: "d" }, { ts: 3, key: "e" }];
{ 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];
@@ -94,10 +88,7 @@ test.cb(
"nonExistingKey", "nonExistingKey",
{ objectMode: true }, { objectMode: true },
); );
const input = [ const input = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }];
{ ts: 0, key: "a" },
{ ts: 1, key: "b" },
];
source source
.pipe(accumulatorStream) .pipe(accumulatorStream)
@@ -191,10 +182,7 @@ 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 = [ const secondFlush = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }];
{ 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" },
@@ -244,10 +232,7 @@ 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 = [ const secondFlush = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }];
{ 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" },
@@ -258,14 +243,8 @@ 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 = [ const fifthFlush = [{ ts: 3, key: "d" }, { ts: 5, key: "f" }];
{ ts: 3, key: "d" }, const sixthFlush = [{ ts: 5, key: "f" }, { ts: 6, key: "g" }];
{ ts: 5, key: "f" },
];
const sixthFlush = [
{ ts: 5, key: "f" },
{ ts: 6, key: "g" },
];
const flushes = [ const flushes = [
firstFlush, firstFlush,
@@ -307,10 +286,7 @@ test.cb(
"nonExistingKey", "nonExistingKey",
{ objectMode: true }, { objectMode: true },
); );
const input = [ const input = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }];
{ ts: 0, key: "a" },
{ ts: 1, key: "b" },
];
source source
.pipe(accumulatorStream) .pipe(accumulatorStream)
@@ -358,22 +334,10 @@ test.cb(
{ ts: 6, key: "g" }, { ts: 6, key: "g" },
]; ];
const firstFlush = [{ ts: 0, key: "a" }]; const firstFlush = [{ ts: 0, key: "a" }];
const secondFlush = [ const secondFlush = [{ ts: 0, key: "a" }, { ts: 2, key: "c" }];
{ ts: 0, key: "a" }, const thirdFlush = [{ ts: 2, key: "c" }, { ts: 3, key: "d" }];
{ ts: 2, key: "c" }, const fourthFlush = [{ ts: 3, key: "d" }, { ts: 5, key: "f" }];
]; 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,
@@ -505,10 +469,7 @@ 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 = [ const secondFlush = [{ ts: 0, key: "a" }, { ts: 1, key: "b" }];
{ 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" },
@@ -519,14 +480,8 @@ test.cb("accumulatorBy() sliding", t => {
{ ts: 2, key: "c" }, { ts: 2, key: "c" },
{ ts: 3, key: "d" }, { ts: 3, key: "d" },
]; ];
const fifthFlush = [ const fifthFlush = [{ ts: 3, key: "d" }, { ts: 5, key: "f" }];
{ ts: 3, key: "d" }, const sixthFlush = [{ ts: 5, key: "f" }, { ts: 6, key: "g" }];
{ 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,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { batch, map, fromArray } = strom({ objectMode: true }); 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);

View File

@@ -2,8 +2,8 @@ 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 { strom } from "../src"; import mhysa from "../src";
const { child } = strom(); 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,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { collect } = strom(); 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 +59,7 @@ test.cb(
const source = new Readable({ objectMode: false }); const source = new Readable({ objectMode: false });
source source
.pipe(collect({ objectMode: false })) .pipe(collect())
.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,9 @@ 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 { strom } from "../src"; import mhysa from "../src";
import { performance } from "perf_hooks"; import { performance } from "perf_hooks";
const { compose, map, fromArray } = strom({ objectMode: true }); 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,9 +114,13 @@ test("compose() writable length should be less than highWaterMark when handing w
return chunk; return chunk;
}); });
const composed = compose([first, second], undefined, { const composed = compose(
[first, second],
undefined,
{
highWaterMark: 2, highWaterMark: 2,
}); },
);
composed.on("error", err => { composed.on("error", err => {
reject(); reject();
}); });
@@ -164,9 +168,13 @@ test("compose() should emit drain event ~rate * highWaterMark ms for every write
return chunk; return chunk;
}); });
const composed = compose([first, second], undefined, { const composed = compose(
[first, second],
undefined,
{
highWaterMark, highWaterMark,
}); },
);
composed.on("error", err => { composed.on("error", err => {
reject(); reject();
}); });
@@ -213,9 +221,13 @@ test.cb(
return chunk; return chunk;
}); });
const composed = compose([first, second], undefined, { const composed = compose(
[first, second],
undefined,
{
highWaterMark: 5, highWaterMark: 5,
}); },
);
composed.on("error", err => { composed.on("error", err => {
t.end(err); t.end(err);
@@ -272,9 +284,13 @@ test.cb(
{ highWaterMark: 1 }, { highWaterMark: 1 },
); );
const composed = compose([first, second], undefined, { const composed = compose(
[first, second],
undefined,
{
highWaterMark: 5, highWaterMark: 5,
}); },
);
composed.on("error", err => { composed.on("error", err => {
t.end(err); t.end(err);
}); });
@@ -339,9 +355,13 @@ test.cb(
{ highWaterMark: 2 }, { highWaterMark: 2 },
); );
const composed = compose([first, second], undefined, { const composed = compose(
[first, second],
undefined,
{
highWaterMark: 5, highWaterMark: 5,
}); },
);
composed.on("error", err => { composed.on("error", err => {
t.end(err); t.end(err);
}); });
@@ -394,9 +414,13 @@ test.cb(
return chunk; return chunk;
}); });
const composed = compose([first, second], undefined, { const composed = compose(
[first, second],
undefined,
{
highWaterMark: 6, highWaterMark: 6,
}); },
);
composed.on("error", err => { composed.on("error", err => {
t.end(err); t.end(err);
@@ -446,9 +470,12 @@ test.cb("compose() should be 'destroyable'", t => {
return chunk; return chunk;
}); });
const composed = compose([first, second], (err: any) => { const composed = compose(
[first, second],
(err: any) => {
t.pass(); t.pass();
}); },
);
const fakeSource = new Readable({ const fakeSource = new Readable({
objectMode: true, objectMode: true,
@@ -504,9 +531,13 @@ test.cb("compose() `finish` and `end` propagates", t => {
return chunk; return chunk;
}); });
const composed = compose([first, second], undefined, { const composed = compose(
[first, second],
undefined,
{
highWaterMark: 3, highWaterMark: 3,
}); },
);
const fakeSource = new Readable({ const fakeSource = new Readable({
objectMode: true, objectMode: true,

View File

@@ -1,8 +1,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { concat, collect } = strom(); 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 +172,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({ objectMode: false })) .pipe(collect())
.on("data", _ => { .on("data", _ => {
t.fail(); t.fail();
}) })

View File

@@ -1,11 +1,11 @@
import { Readable } from "stream"; import { Readable } from "stream";
import test from "ava"; import test from "ava";
import { strom } from "../src"; import mhysa from "../src";
const withDefaultOptions = strom({ objectMode: true }); const withDefaultOptions = mhysa({ objectMode: true });
const withoutOptions = strom(); const withoutOptions = mhysa();
test("strom instances can have default options", t => { test("Mhysa instances can have default options", t => {
let batch = withDefaultOptions.batch(); let batch = withDefaultOptions.batch();
t.true(batch._readableState.objectMode); t.true(batch._readableState.objectMode);
t.true(batch._writableState.objectMode); t.true(batch._writableState.objectMode);

View File

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

View File

@@ -2,8 +2,8 @@ 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 { strom } from "../src"; import mhysa from "../src";
const { duplex } = strom(); 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,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { filter } = strom(); 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,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { flatMap } = strom({ objectMode: true }); 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,7 @@
import test from "ava"; import test from "ava";
import { expect } from "chai"; import { expect } from "chai";
import { strom } from "../src"; import mhysa from "../src";
const { fromArray } = strom(); 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,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { join } = strom(); 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,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { last } = strom(); 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,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { map } = strom(); 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,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { merge } = strom(); 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,9 @@ 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 { strom } from "../src"; import mhysa from "../src";
import { sleep } from "../src/helpers"; import { sleep } from "../src/helpers";
const { parallelMap } = strom({ objectMode: true }); 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,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { parse } = strom(); 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

@@ -2,8 +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 { strom } from "../src"; import mhysa from "../src";
const { rate } = strom({ objectMode: true }); 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(5);

View File

@@ -1,8 +1,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { reduce } = strom({ objectMode: true }); 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,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { replace } = strom(); 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,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { split } = strom(); 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,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { stringify } = strom(); 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,8 @@
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 { strom } from "../src"; import mhysa from "../src";
const { unbatch, batch } = strom({ objectMode: true }); 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,7 @@
import test from "ava"; import test from "ava";
import { collected } from "../../src/utils"; import { collected } from "../../src/utils";
import { strom } from "../../src"; import mhysa from "../../src";
const { fromArray, collect } = strom({ objectMode: true }); 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()));