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

Update tests to write to sink

Browse Source
This commit is contained in:
Jerry Kurian 2019-08-29 08:50:11 -04:00
parent 685215bee6
commit 9765e6cb49
1 changed files with 57 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

95
tests/demux.spec.ts
View File

@ -1,6 +1,7 @@
import test from "ava"; import test from "ava";
import { expect } from "chai"; import { expect } from "chai";
import { demux, map } from "../src"; import { demux, map } from "../src";
import { Writable } from "stream";
interface Test { interface Test {
key: string; key: string;
@ -10,45 +11,48 @@ test.cb("should spread per key", t => {
t.plan(5); t.plan(5);
const input = [ const input = [
{ key: "a", val: 1 }, { key: "a", val: 1 },
{ key: "a", val: 2 }, { key: "b", val: 2 },
{ key: "b", val: 3 }, { key: "a", val: 3 },
{ key: "c", val: 4 }, { key: "c", val: 4 },
]; ];
const results = [ const results = [
{ key: "a", val: 2 }, { key: "a", val: 2 },
{ key: "a", val: 3 }, { key: "b", val: 3 },
{ key: "b", val: 4 }, { key: "a", val: 4 },
{ key: "c", val: 5 }, { key: "c", val: 5 },
]; ];
const destKeys = []; const destinationStreamKeys = [];
const dests = [];
let i = 0; let i = 0;
const sink = new Writable({
const construct = (destKey: string) => { objectMode: true,
destKeys.push(destKey); write(chunk, enc, cb) {
const dest = map((chunk: Test) => ({
...chunk,
val: chunk.val + 1,
}))
.on("data", (d: Test) => {
expect(results).to.deep.include(d);
t.pass();
})
.on("end", () => {
i++; i++;
if (i === dests.length) { expect(results).to.deep.include(chunk);
expect(input).to.not.deep.include(chunk);
t.pass();
cb();
if (i === 4) {
t.end(); t.end();
} }
},
}); });
dests.push(dest); const construct = (destKey: string) => {
destinationStreamKeys.push(destKey);
const dest = map((chunk: Test) => {
return {
...chunk,
val: chunk.val + 1,
};
});
dest.pipe(sink);
return dest; return dest;
}; };
const demuxed = demux(construct, { key: "key" }, { objectMode: true }); const demuxed = demux(construct, { key: "key" }, { objectMode: true });
demuxed.on("finish", () => { demuxed.on("finish", () => {
expect(destKeys).to.deep.equal(["a", "b", "c"]); expect(destinationStreamKeys).to.deep.equal(["a", "b", "c"]);
t.pass(); t.pass();
dests.forEach(dest => dest.end());
}); });
input.forEach(event => demuxed.write(event)); input.forEach(event => demuxed.write(event));
@ -59,37 +63,41 @@ test.cb("should spread per key using keyBy", t => {
t.plan(5); t.plan(5);
const input = [ const input = [
{ key: "a", val: 1 }, { key: "a", val: 1 },
{ key: "a", val: 2 }, { key: "b", val: 2 },
{ key: "b", val: 3 }, { key: "a", val: 3 },
{ key: "c", val: 4 }, { key: "c", val: 4 },
]; ];
const results = [ const results = [
{ key: "a", val: 2 }, { key: "a", val: 2 },
{ key: "a", val: 3 }, { key: "b", val: 3 },
{ key: "b", val: 4 }, { key: "a", val: 4 },
{ key: "c", val: 5 }, { key: "c", val: 5 },
]; ];
const destKeys = []; const destinationStreamKeys = [];
const dests = [];
let i = 0; let i = 0;
const sink = new Writable({
const construct = (destKey: string) => { objectMode: true,
destKeys.push(destKey); write(chunk, enc, cb) {
const dest = map((chunk: Test) => ({
...chunk,
val: chunk.val + 1,
}))
.on("data", (d: Test) => {
expect(results).to.deep.include(d);
t.pass();
})
.on("end", () => {
i++; i++;
if (i === dests.length) { expect(results).to.deep.include(chunk);
expect(input).to.not.deep.include(chunk);
t.pass();
cb();
if (i === 4) {
t.end(); t.end();
} }
},
}); });
dests.push(dest); const construct = (destKey: string) => {
destinationStreamKeys.push(destKey);
const dest = map((chunk: Test) => {
return {
...chunk,
val: chunk.val + 1,
};
});
dest.pipe(sink);
return dest; return dest;
}; };
@ -99,9 +107,8 @@ test.cb("should spread per key using keyBy", t => {
{ objectMode: true }, { objectMode: true },
); );
demuxed.on("finish", () => { demuxed.on("finish", () => {
expect(destKeys).to.deep.equal(["a", "b", "c"]); expect(destinationStreamKeys).to.deep.equal(["a", "b", "c"]);
t.pass(); t.pass();
dests.forEach(dest => dest.end());
}); });
input.forEach(event => demuxed.write(event)); input.forEach(event => demuxed.write(event));