Merge pull request #4 from Jogogoplay/feature/demux-pipe

Allow demux to be piped ie muxed
This commit is contained in:
Lewis Diamond 2020-02-28 17:23:59 -05:00 committed by GitHub
commit 2841f4e182
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 424 additions and 207 deletions

View File

@ -1,4 +1,5 @@
import { pipeline, TransformOptions, Transform } from "stream"; import { pipeline, TransformOptions, Transform } from "stream";
import { AllStreams, isReadable } from "../helpers";
export function compose( export function compose(
streams: Array< streams: Array<
@ -21,18 +22,6 @@ enum EventSubscription {
Self, Self,
} }
type AllStreams =
| NodeJS.ReadableStream
| NodeJS.ReadWriteStream
| NodeJS.WritableStream;
function isReadable(stream: AllStreams): stream is NodeJS.WritableStream {
return (
(stream as NodeJS.ReadableStream).pipe !== undefined &&
(stream as any).readable === true
);
}
export class Compose extends Transform { export class Compose extends Transform {
private first: AllStreams; private first: AllStreams;
private last: AllStreams; private last: AllStreams;

View File

@ -1,4 +1,6 @@
import { WritableOptions, Writable } from "stream"; import { DuplexOptions, Duplex, Transform, Writable } from "stream";
import { isReadable } from "../helpers";
enum EventSubscription { enum EventSubscription {
Last = 0, Last = 0,
@ -8,54 +10,67 @@ enum EventSubscription {
Unhandled, Unhandled,
} }
const eventsTarget = {
close: EventSubscription.Self,
data: EventSubscription.All,
drain: EventSubscription.Self,
end: EventSubscription.Self,
error: EventSubscription.Self,
finish: EventSubscription.Self,
pause: EventSubscription.Self,
pipe: EventSubscription.Self,
readable: EventSubscription.Self,
resume: EventSubscription.Self,
unpipe: EventSubscription.Self,
};
type DemuxStreams = NodeJS.WritableStream | NodeJS.ReadWriteStream; type DemuxStreams = NodeJS.WritableStream | NodeJS.ReadWriteStream;
interface DemuxOptions extends DuplexOptions {
remultiplex?: boolean;
}
export function demux( export function demux(
construct: (destKey?: string) => DemuxStreams, construct: (destKey?: string) => DemuxStreams,
demuxBy: string | ((chunk: any) => string), demuxBy: string | ((chunk: any) => string),
options?: WritableOptions, options?: DemuxOptions,
): Writable { ): Duplex {
return new Demux(construct, demuxBy, options); return new Demux(construct, demuxBy, options);
} }
// @TODO handle pipe event ie) Multiplex class Demux extends Duplex {
class Demux extends Writable {
private streamsByKey: { private streamsByKey: {
[key: string]: DemuxStreams; [key: string]: DemuxStreams;
}; };
private demuxer: (chunk: any) => string; private demuxer: (chunk: any) => string;
private construct: (destKey?: string) => DemuxStreams; private construct: (destKey?: string) => DemuxStreams;
private remultiplex: boolean;
private transform: Transform;
constructor( constructor(
construct: (destKey?: string) => DemuxStreams, construct: (destKey?: string) => DemuxStreams,
demuxBy: string | ((chunk: any) => string), demuxBy: string | ((chunk: any) => string),
options: WritableOptions = {}, options: DemuxOptions = {},
) { ) {
super(options); super(options);
this.demuxer = this.demuxer =
typeof demuxBy === "string" ? chunk => chunk[demuxBy] : demuxBy; typeof demuxBy === "string" ? chunk => chunk[demuxBy] : demuxBy;
this.construct = construct; this.construct = construct;
this.remultiplex =
options.remultiplex === undefined ? true : options.remultiplex;
this.streamsByKey = {}; this.streamsByKey = {};
this.transform = new Transform({
...options,
transform: (d, _, cb) => {
this.push(d);
cb(null);
},
});
this.on("unpipe", () => this._flush());
} }
public _read(size: number) {}
public async _write(chunk: any, encoding: any, cb: any) { public async _write(chunk: any, encoding: any, cb: any) {
const destKey = this.demuxer(chunk); const destKey = this.demuxer(chunk);
if (this.streamsByKey[destKey] === undefined) { if (this.streamsByKey[destKey] === undefined) {
this.streamsByKey[destKey] = await this.construct(destKey); const newPipeline = await this.construct(destKey);
this.streamsByKey[destKey] = 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`,
);
}
} }
if (!this.streamsByKey[destKey].write(chunk, encoding)) { if (!this.streamsByKey[destKey].write(chunk, encoding)) {
this.streamsByKey[destKey].once("drain", () => { this.streamsByKey[destKey].once("drain", () => {
cb(); cb();
@ -65,35 +80,24 @@ class Demux extends Writable {
} }
} }
public on(event: string, cb: any) { public _flush() {
switch (eventsTarget[event]) { const pipelines = Object.values(this.streamsByKey);
case EventSubscription.Self: let totalEnded = 0;
super.on(event, cb); pipelines.forEach(pipeline => {
break; pipeline.once("end", () => {
case EventSubscription.All: totalEnded++;
Object.keys(this.streamsByKey).forEach(key => if (pipelines.length === totalEnded) {
this.streamsByKey[key].on(event, cb), this.push(null);
); this.emit("end");
break; }
default: });
super.on(event, cb); });
} pipelines.forEach(pipeline => pipeline.end());
return this;
} }
public once(event: string, cb: any) { public _destroy(error: any, cb: (error?: any) => void) {
switch (eventsTarget[event]) { const pipelines = Object.values(this.streamsByKey);
case EventSubscription.Self: pipelines.forEach(p => (p as any).destroy());
super.once(event, cb); cb(error);
break;
case EventSubscription.All:
Object.keys(this.streamsByKey).forEach(key =>
this.streamsByKey[key].once(event, cb),
);
break;
default:
super.once(event, cb);
}
return this;
} }
} }

View File

@ -1,3 +1,17 @@
export async function sleep(time: number): Promise<{} | null> { export async function sleep(time: number): Promise<{} | null> {
return time > 0 ? new Promise(resolve => setTimeout(resolve, time)) : null; return time > 0 ? new Promise(resolve => setTimeout(resolve, time)) : null;
} }
export type AllStreams =
| NodeJS.ReadableStream
| NodeJS.ReadWriteStream
| NodeJS.WritableStream;
export function isReadable(
stream: AllStreams,
): stream is NodeJS.WritableStream {
return (
(stream as NodeJS.ReadableStream).pipe !== undefined &&
(stream as any).readable === true
);
}

View File

@ -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(
highWaterMark: 2, [first, second],
}); undefined,
{
highWaterMark: 2,
},
);
composed.on("error", err => { composed.on("error", err => {
reject(); reject();
}); });
@ -171,9 +175,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(
highWaterMark, [first, second],
}); undefined,
{
highWaterMark,
},
);
composed.on("error", err => { composed.on("error", err => {
reject(); reject();
}); });
@ -237,9 +245,13 @@ test.cb(
return chunk; return chunk;
}); });
const composed = compose([first, second], undefined, { const composed = compose(
highWaterMark: 5, [first, second],
}); undefined,
{
highWaterMark: 5,
},
);
composed.on("error", err => { composed.on("error", err => {
t.end(err); t.end(err);
@ -301,9 +313,13 @@ test.cb(
{ highWaterMark: 1 }, { highWaterMark: 1 },
); );
const composed = compose([first, second], undefined, { const composed = compose(
highWaterMark: 5, [first, second],
}); undefined,
{
highWaterMark: 5,
},
);
composed.on("error", err => { composed.on("error", err => {
t.end(err); t.end(err);
}); });
@ -368,9 +384,13 @@ test.cb(
{ highWaterMark: 2 }, { highWaterMark: 2 },
); );
const composed = compose([first, second], undefined, { const composed = compose(
highWaterMark: 5, [first, second],
}); undefined,
{
highWaterMark: 5,
},
);
composed.on("error", err => { composed.on("error", err => {
t.end(err); t.end(err);
}); });
@ -423,9 +443,13 @@ test.cb(
return chunk; return chunk;
}); });
const composed = compose([first, second], undefined, { const composed = compose(
highWaterMark: 6, [first, second],
}); undefined,
{
highWaterMark: 6,
},
);
composed.on("error", err => { composed.on("error", err => {
t.end(err); t.end(err);
@ -475,9 +499,12 @@ test.cb("compose() should be 'destroyable'", t => {
return chunk; return chunk;
}); });
const composed = compose([first, second], (err: any) => { const composed = compose(
t.pass(); [first, second],
}); (err: any) => {
t.pass();
},
);
const fakeSource = new Readable({ const fakeSource = new Readable({
objectMode: true, objectMode: true,
@ -533,9 +560,13 @@ test.cb("compose() `finish` and `end` propagates", t => {
return chunk; return chunk;
}); });
const composed = compose([first, second], undefined, { const composed = compose(
highWaterMark: 3, [first, second],
}); undefined,
{
highWaterMark: 3,
},
);
const fakeSource = new Readable({ const fakeSource = new Readable({
objectMode: true, objectMode: true,

View File

@ -1,11 +1,11 @@
import test from "ava"; import test from "ava";
import { expect } from "chai"; import { expect } from "chai";
import mhysa from "../src"; import mhysa from "../src";
import { Writable } from "stream"; import { Writable, Readable } from "stream";
const sinon = require("sinon"); const sinon = require("sinon");
const { sleep } = require("../src/helpers"); const { sleep } = require("../src/helpers");
import { performance } from "perf_hooks"; import { performance } from "perf_hooks";
const { demux, map } = mhysa(); const { demux, map, fromArray } = mhysa({ objectMode: true });
interface Test { interface Test {
key: string; key: string;
@ -31,7 +31,7 @@ test.cb("demux() constructor should be called once per key", t => {
return dest; return dest;
}); });
const demuxed = demux(construct, "key", { objectMode: true }); const demuxed = demux(construct, "key", {});
demuxed.on("finish", () => { demuxed.on("finish", () => {
expect(construct.withArgs("a").callCount).to.equal(1); expect(construct.withArgs("a").callCount).to.equal(1);
@ -41,8 +41,7 @@ test.cb("demux() constructor should be called once per key", t => {
t.end(); t.end();
}); });
input.forEach(event => demuxed.write(event)); fromArray(input).pipe(demuxed);
demuxed.end();
}); });
test.cb("demux() should send input through correct pipeline", t => { test.cb("demux() should send input through correct pipeline", t => {
@ -66,7 +65,7 @@ test.cb("demux() should send input through correct pipeline", t => {
return dest; return dest;
}; };
const demuxed = demux(construct, "key", { objectMode: true }); const demuxed = demux(construct, "key", {});
demuxed.on("finish", () => { demuxed.on("finish", () => {
pipelineSpies["a"].getCalls().forEach(call => { pipelineSpies["a"].getCalls().forEach(call => {
@ -84,8 +83,7 @@ test.cb("demux() should send input through correct pipeline", t => {
t.end(); t.end();
}); });
input.forEach(event => demuxed.write(event)); fromArray(input).pipe(demuxed);
demuxed.end();
}); });
test.cb("demux() constructor should be called once per key using keyBy", t => { test.cb("demux() constructor should be called once per key using keyBy", t => {
@ -108,7 +106,7 @@ test.cb("demux() constructor should be called once per key using keyBy", t => {
return dest; return dest;
}); });
const demuxed = demux(construct, item => item.key, { objectMode: true }); const demuxed = demux(construct, item => item.key, {});
demuxed.on("finish", () => { demuxed.on("finish", () => {
expect(construct.withArgs("a").callCount).to.equal(1); expect(construct.withArgs("a").callCount).to.equal(1);
@ -118,8 +116,7 @@ test.cb("demux() constructor should be called once per key using keyBy", t => {
t.end(); t.end();
}); });
input.forEach(event => demuxed.write(event)); fromArray(input).pipe(demuxed);
demuxed.end();
}); });
test.cb("demux() should send input through correct pipeline using keyBy", t => { test.cb("demux() should send input through correct pipeline using keyBy", t => {
@ -143,7 +140,7 @@ test.cb("demux() should send input through correct pipeline using keyBy", t => {
return dest; return dest;
}; };
const demuxed = demux(construct, item => item.key, { objectMode: true }); const demuxed = demux(construct, item => item.key, {});
demuxed.on("finish", () => { demuxed.on("finish", () => {
pipelineSpies["a"].getCalls().forEach(call => { pipelineSpies["a"].getCalls().forEach(call => {
@ -161,11 +158,10 @@ test.cb("demux() should send input through correct pipeline using keyBy", t => {
t.end(); t.end();
}); });
input.forEach(event => demuxed.write(event)); fromArray(input).pipe(demuxed);
demuxed.end();
}); });
test("demux() write should return false after if it has >= highWaterMark items buffered and drain should be emitted", t => { test("demux() write should return false and emit drain if more than highWaterMark items are buffered", t => {
return new Promise(async (resolve, reject) => { return new Promise(async (resolve, reject) => {
t.plan(7); t.plan(7);
interface Chunk { interface Chunk {
@ -189,7 +185,7 @@ test("demux() write should return false after if it has >= highWaterMark items b
await sleep(slowProcessorSpeed); await sleep(slowProcessorSpeed);
return { ...chunk, mapped: [1] }; return { ...chunk, mapped: [1] };
}, },
{ highWaterMark: 1, objectMode: true }, { highWaterMark: 1 },
); );
first.on("data", chunk => { first.on("data", chunk => {
@ -205,7 +201,6 @@ test("demux() write should return false after if it has >= highWaterMark items b
}; };
const _demux = demux(construct, "key", { const _demux = demux(construct, "key", {
objectMode: true,
highWaterMark, highWaterMark,
}); });
@ -229,7 +224,7 @@ test("demux() write should return false after if it has >= highWaterMark items b
}); });
}); });
test("demux() should emit one drain event after slowProcessorSpeed * highWaterMark ms", t => { test("demux() should emit one drain event after slowProcessorSpeed * highWaterMark ms when first stream is bottleneck", t => {
return new Promise(async (resolve, reject) => { return new Promise(async (resolve, reject) => {
t.plan(7); t.plan(7);
interface Chunk { interface Chunk {
@ -255,7 +250,7 @@ test("demux() should emit one drain event after slowProcessorSpeed * highWaterMa
chunk.mapped.push(1); chunk.mapped.push(1);
return chunk; return chunk;
}, },
{ highWaterMark: 1, objectMode: true }, { highWaterMark: 1 },
); );
first.on("data", () => { first.on("data", () => {
@ -268,7 +263,6 @@ test("demux() should emit one drain event after slowProcessorSpeed * highWaterMa
return first; return first;
}; };
const _demux = demux(construct, "key", { const _demux = demux(construct, "key", {
objectMode: true,
highWaterMark, highWaterMark,
}); });
_demux.on("error", err => { _demux.on("error", err => {
@ -296,7 +290,7 @@ test("demux() should emit one drain event after slowProcessorSpeed * highWaterMa
test("demux() should emit one drain event when writing 6 items with highWaterMark of 5", t => { test("demux() should emit one drain event when writing 6 items with highWaterMark of 5", t => {
return new Promise(async (resolve, reject) => { return new Promise(async (resolve, reject) => {
t.plan(7); t.plan(1);
interface Chunk { interface Chunk {
key: string; key: string;
mapped: number[]; mapped: number[];
@ -318,12 +312,11 @@ test("demux() should emit one drain event when writing 6 items with highWaterMar
chunk.mapped.push(2); chunk.mapped.push(2);
return chunk; return chunk;
}, },
{ highWaterMark: 1, objectMode: true }, { highWaterMark: 1 },
); );
first.on("data", () => { first.on("data", () => {
pendingReads--; pendingReads--;
t.pass();
if (pendingReads === 0) { if (pendingReads === 0) {
resolve(); resolve();
} }
@ -331,7 +324,6 @@ test("demux() should emit one drain event when writing 6 items with highWaterMar
return first; return first;
}; };
const _demux = demux(construct, "key", { const _demux = demux(construct, "key", {
objectMode: true,
highWaterMark: 5, highWaterMark: 5,
}); });
@ -355,9 +347,10 @@ test("demux() should emit one drain event when writing 6 items with highWaterMar
}); });
}); });
test.cb.only( test.cb(
"demux() should emit drain event when third stream is bottleneck", "demux() should emit drain event when second stream is bottleneck after (highWaterMark - 2) * slowProcessorSpeed ms",
t => { t => {
// ie) first two items are pushed directly into first and second streams (highWaterMark - 2 remain in demux)
t.plan(8); t.plan(8);
const slowProcessorSpeed = 100; const slowProcessorSpeed = 100;
const highWaterMark = 5; const highWaterMark = 5;
@ -383,7 +376,7 @@ test.cb.only(
chunk.mapped.push(1); chunk.mapped.push(1);
return chunk; return chunk;
}, },
{ objectMode: true, highWaterMark: 1 }, { highWaterMark: 1 },
); );
const second = map( const second = map(
@ -392,25 +385,23 @@ test.cb.only(
chunk.mapped.push(2); chunk.mapped.push(2);
return chunk; return chunk;
}, },
{ objectMode: true, highWaterMark: 1 }, { highWaterMark: 1 },
); );
first.pipe(second).pipe(sink); first.pipe(second).pipe(sink);
return first; return first;
}; };
const _demux = demux(construct, () => "a", { const _demux = demux(construct, () => "a", {
objectMode: true,
highWaterMark, highWaterMark,
}); });
_demux.on("error", err => { _demux.on("error", err => {
t.end(err); t.end(err);
}); });
// This event should be received after at least 5 * slowProcessorSpeed (two are read immediately by first and second, 5 remaining in demux before drain event)
_demux.on("drain", () => { _demux.on("drain", () => {
expect(_demux._writableState.length).to.be.equal(0); expect(_demux._writableState.length).to.be.equal(0);
expect(performance.now() - start).to.be.greaterThan( expect(performance.now() - start).to.be.greaterThan(
slowProcessorSpeed * (input.length - 2), slowProcessorSpeed * 3,
); );
t.pass(); t.pass();
}); });
@ -427,15 +418,14 @@ test.cb.only(
let pendingReads = input.length; let pendingReads = input.length;
const start = performance.now(); const start = performance.now();
input.forEach(item => { fromArray(input).pipe(_demux);
_demux.write(item);
});
}, },
); );
test.cb( test.cb(
"demux() should emit drain event when second stream is bottleneck", "demux() should emit drain event when third stream is bottleneck",
t => { t => {
// @TODO investigate why drain is emitted after slowProcessorSpeed
t.plan(8); t.plan(8);
const slowProcessorSpeed = 100; const slowProcessorSpeed = 100;
const highWaterMark = 5; const highWaterMark = 5;
@ -446,7 +436,7 @@ test.cb(
const sink = new Writable({ const sink = new Writable({
objectMode: true, objectMode: true,
write(chunk, encoding, cb) { write(chunk, encoding, cb) {
expect(chunk.mapped).to.deep.equal([1, 2]); expect(chunk.mapped).to.deep.equal([1, 2, 3]);
t.pass(); t.pass();
pendingReads--; pendingReads--;
if (pendingReads === 0) { if (pendingReads === 0) {
@ -461,14 +451,14 @@ test.cb(
chunk.mapped.push(1); chunk.mapped.push(1);
return chunk; return chunk;
}, },
{ objectMode: true, highWaterMark: 1 }, { highWaterMark: 1 },
); );
const second = map( const second = map(
(chunk: Chunk) => { (chunk: Chunk) => {
chunk.mapped.push(2); chunk.mapped.push(2);
return chunk; return chunk;
}, },
{ objectMode: true, highWaterMark: 1 }, { highWaterMark: 1 },
); );
const third = map( const third = map(
@ -477,7 +467,7 @@ test.cb(
chunk.mapped.push(3); chunk.mapped.push(3);
return chunk; return chunk;
}, },
{ objectMode: true, highWaterMark: 1 }, { highWaterMark: 1 },
); );
first first
@ -487,18 +477,16 @@ test.cb(
return first; return first;
}; };
const _demux = demux(construct, () => "a", { const _demux = demux(construct, () => "a", {
objectMode: true,
highWaterMark, highWaterMark,
}); });
_demux.on("error", err => { _demux.on("error", err => {
t.end(err); t.end(err);
}); });
// This event should be received after at least 3 * slowProcessorSpeed (two are read immediately by first and second, 3 remaining in demux before drain event)
_demux.on("drain", () => { _demux.on("drain", () => {
expect(_demux._writableState.length).to.be.equal(0); expect(_demux._writableState.length).to.be.equal(0);
expect(performance.now() - start).to.be.greaterThan( expect(performance.now() - start).to.be.greaterThan(
slowProcessorSpeed * (input.length - 4), slowProcessorSpeed,
); );
t.pass(); t.pass();
}); });
@ -515,9 +503,7 @@ test.cb(
let pendingReads = input.length; let pendingReads = input.length;
const start = performance.now(); const start = performance.now();
input.forEach(item => { fromArray(input).pipe(_demux);
_demux.write(item);
});
}, },
); );
@ -536,36 +522,37 @@ test("demux() should be blocked by slowest pipeline", t => {
chunk.mapped.push(1); chunk.mapped.push(1);
return chunk; return chunk;
}, },
{ objectMode: true, highWaterMark: 1 }, { highWaterMark: 1 },
); );
first.on("data", chunk => {
pendingReads--;
if (chunk.key === "b") {
expect(performance.now() - start).to.be.greaterThan(
slowProcessorSpeed * totalItems,
);
t.pass();
expect(pendingReads).to.equal(0);
resolve();
}
});
return first; return first;
}; };
const _demux = demux(construct, "key", { const _demux = demux(construct, "key", {
objectMode: true,
highWaterMark: 1, highWaterMark: 1,
}); });
_demux.on("error", err => { _demux.on("error", err => {
reject(err); reject(err);
}); });
_demux.on("data", async chunk => {
pendingReads--;
if (chunk.key === "b") {
expect(performance.now() - start).to.be.greaterThan(
slowProcessorSpeed * totalItems,
);
t.pass();
expect(pendingReads).to.equal(0);
resolve();
}
});
const input = [ const input = [
{ key: "a", mapped: [] }, { key: "a", mapped: [] },
{ key: "a", mapped: [] }, { key: "a", mapped: [] },
{ key: "c", mapped: [] }, { key: "c", mapped: [] },
{ key: "c", mapped: [] }, { key: "c", mapped: [] },
{ key: "c", mapped: [] },
{ key: "b", mapped: [] }, { key: "b", mapped: [] },
]; ];
@ -584,74 +571,266 @@ test("demux() should be blocked by slowest pipeline", t => {
}); });
}); });
test("demux() should emit drain event when second stream in pipeline is bottleneck", t => { test.cb("Demux should remux to sink", t => {
t.plan(5); t.plan(6);
const highWaterMark = 3; let i = 0;
return new Promise(async (resolve, reject) => { const input = [
interface Chunk { { key: "a", visited: [] },
key: string; { key: "b", visited: [] },
mapped: number[]; { key: "a", visited: [] },
{ key: "c", visited: [] },
{ key: "a", visited: [] },
{ key: "b", visited: [] },
];
const result = [
{ key: "a", visited: ["a"] },
{ key: "b", visited: ["b"] },
{ key: "a", visited: ["a"] },
{ key: "c", visited: ["c"] },
{ key: "a", visited: ["a"] },
{ key: "b", visited: ["b"] },
];
const construct = (destKey: string) => {
const dest = map((chunk: any) => {
chunk.visited.push(destKey);
return chunk;
});
return dest;
};
const sink = map(d => {
t.deepEqual(d, result[i]);
i++;
if (i === input.length) {
t.end();
} }
const sink = new Writable({ });
objectMode: true,
write(chunk, encoding, cb) { const demuxed = demux(construct, "key", {});
expect(chunk.mapped).to.deep.equal([1, 2]);
t.pass(); fromArray(input)
cb(); .pipe(demuxed)
if (pendingReads === 0) { .pipe(sink);
resolve(); });
}
}, test.cb("Demux should send data events", t => {
t.plan(6);
let i = 0;
const input = [
{ key: "a", visited: [] },
{ key: "b", visited: [] },
{ key: "a", visited: [] },
{ key: "c", visited: [] },
{ key: "a", visited: [] },
{ key: "b", visited: [] },
];
const result = [
{ key: "a", visited: ["a"] },
{ key: "b", visited: ["b"] },
{ key: "a", visited: ["a"] },
{ key: "c", visited: ["c"] },
{ key: "a", visited: ["a"] },
{ key: "b", visited: ["b"] },
];
const construct = (destKey: string) => {
const dest = map((chunk: any) => {
chunk.visited.push(destKey);
return chunk;
}); });
const construct = (destKey: string) => { return dest;
const first = map( };
(chunk: Chunk) => {
expect(first._readableState.length).to.be.at.most(2);
chunk.mapped.push(1);
return chunk;
},
{ objectMode: true, highWaterMark: 2 },
);
const second = map( const demuxed = demux(construct, "key", {});
async (chunk: Chunk) => {
await sleep(100);
chunk.mapped.push(2);
expect(second._writableState.length).to.be.equal(1);
pendingReads--;
return chunk;
},
{ objectMode: true, highWaterMark: 1 },
);
first.pipe(second).pipe(sink); fromArray(input).pipe(demuxed);
return first;
};
const _demux = demux(construct, "key", { demuxed.on("data", d => {
objectMode: true, t.deepEqual(d, result[i]);
highWaterMark, i++;
}); if (i === input.length) {
_demux.on("error", err => { t.end();
reject(); }
});
_demux.on("drain", () => {
expect(_demux._writableState.length).to.be.equal(0);
t.pass();
});
const input = [
{ key: "a", mapped: [] },
{ key: "a", mapped: [] },
{ key: "a", mapped: [] },
{ key: "a", mapped: [] },
];
let pendingReads = input.length;
input.forEach(item => {
_demux.write(item);
});
}); });
}); });
test.cb("demux() `finish` and `end` propagates", t => {
interface Chunk {
key: string;
mapped: number[];
}
t.plan(9);
const construct = (destKey: string) => {
const dest = map((chunk: any) => {
chunk.mapped.push(destKey);
return chunk;
});
return dest;
};
const _demux = demux(construct, "key", {
highWaterMark: 3,
});
const fakeSource = new Readable({
objectMode: true,
read() {
return;
},
});
const sink = map((d: any) => {
const curr = input.shift();
t.is(curr.key, d.key);
t.deepEqual(d.mapped, [d.key]);
});
fakeSource.pipe(_demux).pipe(sink);
fakeSource.on("end", () => {
t.pass();
});
_demux.on("finish", () => {
t.pass();
});
_demux.on("unpipe", () => {
t.pass();
});
_demux.on("end", () => {
t.pass();
t.end();
});
sink.on("finish", () => {
t.pass();
});
const input = [
{ key: "a", mapped: [] },
{ key: "b", mapped: [] },
{ key: "a", mapped: [] },
{ key: "a", mapped: [] },
{ key: "b", mapped: [] },
];
fakeSource.push(input[0]);
fakeSource.push(input[1]);
fakeSource.push(null);
});
test.cb("demux() `unpipe` propagates", t => {
interface Chunk {
key: string;
mapped: number[];
}
t.plan(7);
const construct = (destKey: string) => {
const dest = map((chunk: any) => {
chunk.mapped.push(destKey);
return chunk;
});
return dest;
};
const _demux = demux(construct, "key", {
highWaterMark: 3,
});
const fakeSource = new Readable({
objectMode: true,
read() {
return;
},
});
const sink = map((d: any) => {
const curr = input.shift();
t.is(curr.key, d.key);
t.deepEqual(d.mapped, [d.key]);
});
fakeSource.pipe(_demux).pipe(sink);
_demux.on("unpipe", () => {
t.pass();
});
sink.on("unpipe", () => {
t.pass();
});
sink.on("finish", () => {
t.pass();
t.end();
});
const input = [
{ key: "a", mapped: [] },
{ key: "b", mapped: [] },
{ key: "a", mapped: [] },
{ key: "a", mapped: [] },
{ key: "b", mapped: [] },
];
fakeSource.push(input[0]);
fakeSource.push(input[1]);
fakeSource.push(null);
});
test.cb("demux() should be 'destroyable'", t => {
t.plan(2);
const _sleep = 100;
interface Chunk {
key: string;
mapped: string[];
}
const construct = (destKey: string) => {
const first = map(async (chunk: Chunk) => {
await sleep(_sleep);
chunk.mapped.push(destKey);
return chunk;
});
return first;
};
const _demux = demux(construct, "key");
const fakeSource = new Readable({
objectMode: true,
read() {
return;
},
});
const fakeSink = new Writable({
objectMode: true,
write(data, enc, cb) {
const cur = input.shift();
t.is(cur.key, data.key);
t.deepEqual(cur.mapped, ["a"]);
if (cur.key === "a") {
_demux.destroy();
}
cb();
},
});
_demux.on("close", t.end);
fakeSource.pipe(_demux).pipe(fakeSink);
const input = [
{ key: "a", mapped: [] },
{ key: "b", mapped: [] },
{ key: "c", mapped: [] },
{ key: "d", mapped: [] },
{ key: "e", mapped: [] },
];
fakeSource.push(input[0]);
fakeSource.push(input[1]);
fakeSource.push(input[2]);
fakeSource.push(input[3]);
fakeSource.push(input[4]);
});