This commit is contained in:
Jerry Kurian 2019-08-22 14:52:39 -04:00
parent d097fa6aa5
commit 1d0e15890c
5 changed files with 59 additions and 57 deletions

View File

@ -19,4 +19,4 @@ export { replace } from "./replace";
export { split } from "./split"; export { split } from "./split";
export { stringify } from "./stringify"; export { stringify } from "./stringify";
export { unbatch } from "./unbatch"; export { unbatch } from "./unbatch";
export { compose, composeDuplex } from "./compose"; export { compose } from "./compose";

View File

@ -2,7 +2,6 @@ import {
pipeline, pipeline,
Transform, Transform,
Writable, Writable,
Pipe,
WritableOptions, WritableOptions,
Readable, Readable,
Duplex, Duplex,
@ -12,7 +11,6 @@ import {
* Return a Readable stream of readable streams concatenated together * Return a Readable stream of readable streams concatenated together
* @param streams Readable streams to concatenate * @param streams Readable streams to concatenate
*/ */
// First Readable --> Readable // First Readable --> Readable
// First Transform | Duplex, Last Writable --> Writable // First Transform | Duplex, Last Writable --> Writable
// //
@ -22,26 +20,23 @@ export function compose(
): Duplex { ): Duplex {
// Maybe just return a new stream here // Maybe just return a new stream here
if (streams.length < 2) { if (streams.length < 2) {
throw new Error("Not enough"); throw new Error("At least two streams are required to compose");
} }
const duplex = new Duplex({
objectMode: true,
write(chunk, enc, cb) {
const first = streams[0] as Writable; const first = streams[0] as Writable;
const last = streams[streams.length - 1] as Readable;
const duplex = new Duplex({
...options,
write(chunk, enc, cb) {
if (!first.write(chunk)) { if (!first.write(chunk)) {
first.on("drain", cb); first.once("drain", cb);
} else { } else {
cb(); cb();
} }
}, },
read(size) { read(size) {
let chunk; if (last.readable) {
while ( this.push(last.read(size));
null !==
(chunk = (streams[streams.length - 1] as Readable).read())
) {
this.push(chunk);
} }
}, },
}); });

View File

@ -297,9 +297,3 @@ export function compose(
options, options,
); );
} }
export function composeDuplex(
streams: Array<Writable | Transform>,
options?: WritableOptions,
) {
return baseFunctions.composeDuplex(streams, options);
}

View File

@ -22,5 +22,4 @@ export {
accumulator, accumulator,
accumulatorBy, accumulatorBy,
compose, compose,
composeDuplex,
} from "./functions"; } from "./functions";

View File

@ -1,54 +1,68 @@
const test = require("ava"); const test = require("ava");
const { expect } = require("chai");
const { compose, composeDuplex, map } = require("../src"); const { compose, composeDuplex, map } = require("../src");
test.cb("compose()", t => { test.cb("compose() chains two streams together in the correct order", t => {
const first = map((chunk: number) => chunk + "x"); t.plan(3);
const second = map((chunk: number) => chunk + "y"); let i = 0;
const first = map((chunk: number) => chunk + 1);
const second = map((chunk: number) => chunk * 2);
const composed = compose( const composed = compose(
[first, second], [first, second],
{ objectMode: true }, { objectMode: true },
); );
const third = map((chunk: number) => chunk + "z");
composed
.pipe(third)
.on("data", data => console.log("Piped composed: ", data));
composed.on("data", data => { composed.on("data", data => {
console.log("data on composed", data); expect(data).to.equal(result[i]);
t.pass();
i++;
if (i === 3) {
t.end();
}
});
composed.on("error", err => {
t.end(err);
});
composed.on("end", () => {
t.end(); t.end();
}); });
composed.on("error", data => {
console.log("ERROR", data);
});
composed.on("end", data => {
console.log("end", data);
});
composed.write(1); const input = [1, 2, 3];
composed.write(2); const result = [4, 6, 8];
input.forEach(item => composed.write(item));
}); });
test.cb.only("composeDuplex()", t => {
const first = map((chunk: number) => chunk + "x");
const second = map((chunk: number) => chunk + "y");
const composed = composeDuplex([first, second], { objectMode: true }); test.cb(
const third = map((chunk: number) => chunk + "z"); "compose() followed by pipe chains streams together in the correct order",
// composed t => {
// .pipe(third) t.plan(3);
// .on("data", data => console.log("Piped composed: ", data)); let i = 0;
const first = map((chunk: number) => chunk + 1);
const second = map((chunk: number) => chunk * 2);
composed.on("data", data => { const composed = compose(
console.log("data on composed", data); [first, second],
{ objectMode: true },
);
const third = map((chunk: number) => chunk + 1);
composed.pipe(third).on("data", data => {
expect(data).to.equal(result[i]);
t.pass();
i++;
if (i === 3) {
t.end(); t.end();
}); }
composed.on("error", data => {
console.log("ERROR", data);
});
composed.on("end", data => {
console.log("end", data);
}); });
composed.write(1); composed.on("error", err => {
composed.write(2); t.end(err);
}); });
const input = [1, 2, 3];
const result = [5, 7, 9];
input.forEach(item => composed.write(item));
},
);