ldiamond/strom
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Allow CB to be called by construction streams

Browse Source
This commit is contained in:
Jerry Kurian 2019-08-29 14:39:08 -04:00
parent 9765e6cb49
commit 2524d51aa7
3 changed files with 17 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/functions/compose.ts
View File

@ -52,8 +52,10 @@ class Compose extends Duplex {
return (this.last as NodeJS.ReadableStream).pipe(dest); return (this.last as NodeJS.ReadableStream).pipe(dest);
} }
public write(chunk: any) { public _write(chunk: any, encoding: string, cb: any) {
return (this.first as NodeJS.WritableStream).write(chunk); const res = (this.first as NodeJS.WritableStream).write(chunk);
cb();
return res;
} }
public on(event: string, cb: any) { public on(event: string, cb: any) {

20
src/functions/demux.ts
View File

@ -14,7 +14,9 @@ export function demux(
} }
class Demux extends Writable { class Demux extends Writable {
private keyMap: object; private keyMap: {
[key: string]: NodeJS.WritableStream | NodeJS.ReadWriteStream;
};
private demuxer: (chunk: any) => string; private demuxer: (chunk: any) => string;
private construct: ( private construct: (
destKey?: string, destKey?: string,
@ -28,22 +30,22 @@ class Demux extends Writable {
) { ) {
super(options); super(options);
if (demuxBy.keyBy === undefined && demuxBy.key === undefined) { if (demuxBy.keyBy === undefined && demuxBy.key === undefined) {
throw new Error("Need one"); throw new Error(
"keyBy or key must be provided in second parameter",
);
} }
this.demuxer = demuxBy.keyBy || ((chunk: any) => chunk[demuxBy.key!]); this.demuxer = demuxBy.keyBy || ((chunk: any) => chunk[demuxBy.key!]);
this.construct = construct; this.construct = construct;
this.keyMap = {}; this.keyMap = {};
} }
public write(chunk: any, encoding?: any, cb?: any): boolean { public _write(chunk: any, encoding: string, cb: any) {
const destKey = this.demuxer(chunk); const destKey = this.demuxer(chunk);
if (this.keyMap[destKey] === undefined) { if (this.keyMap[destKey] === undefined) {
this.keyMap[destKey] = this.construct(destKey); this.keyMap[destKey] = this.construct(destKey).on("error", e => {
this.emit("error", e);
});
} }
const writeRes = this.keyMap[destKey].write(chunk); return this.keyMap[destKey].write(chunk, encoding, cb);
if (cb !== undefined) {
cb();
}
return writeRes;
} }
} }

11
tests/demux.spec.ts
View File

@ -26,14 +26,10 @@ test.cb("should spread per key", t => {
const sink = new Writable({ const sink = new Writable({
objectMode: true, objectMode: true,
write(chunk, enc, cb) { write(chunk, enc, cb) {
i++;
expect(results).to.deep.include(chunk); expect(results).to.deep.include(chunk);
expect(input).to.not.deep.include(chunk); expect(input).to.not.deep.include(chunk);
t.pass(); t.pass();
cb(); cb();
if (i === 4) {
t.end();
}
}, },
}); });
const construct = (destKey: string) => { const construct = (destKey: string) => {
@ -53,6 +49,7 @@ test.cb("should spread per key", t => {
demuxed.on("finish", () => { demuxed.on("finish", () => {
expect(destinationStreamKeys).to.deep.equal(["a", "b", "c"]); expect(destinationStreamKeys).to.deep.equal(["a", "b", "c"]);
t.pass(); t.pass();
t.end();
}); });
input.forEach(event => demuxed.write(event)); input.forEach(event => demuxed.write(event));
@ -74,18 +71,13 @@ test.cb("should spread per key using keyBy", t => {
{ key: "c", val: 5 }, { key: "c", val: 5 },
]; ];
const destinationStreamKeys = []; const destinationStreamKeys = [];
let i = 0;
const sink = new Writable({ const sink = new Writable({
objectMode: true, objectMode: true,
write(chunk, enc, cb) { write(chunk, enc, cb) {
i++;
expect(results).to.deep.include(chunk); expect(results).to.deep.include(chunk);
expect(input).to.not.deep.include(chunk); expect(input).to.not.deep.include(chunk);
t.pass(); t.pass();
cb(); cb();
if (i === 4) {
t.end();
}
}, },
}); });
const construct = (destKey: string) => { const construct = (destKey: string) => {
@ -109,6 +101,7 @@ test.cb("should spread per key using keyBy", t => {
demuxed.on("finish", () => { demuxed.on("finish", () => {
expect(destinationStreamKeys).to.deep.equal(["a", "b", "c"]); expect(destinationStreamKeys).to.deep.equal(["a", "b", "c"]);
t.pass(); t.pass();
t.end();
}); });
input.forEach(event => demuxed.write(event)); input.forEach(event => demuxed.write(event));