import { Buffer } from "../buffer.ts";import pipeline from "./pipeline.ts";import Readable from "./readable.ts";import Transform from "./transform.ts";import Writable from "./writable.ts";import { mustCall } from "../_utils.ts";import { assert, assertEquals, assertStrictEquals,} from "../../testing/asserts.ts";import type { NodeErrorAbstraction } from "../_errors.ts";
Deno.test("Pipeline ends on stream finished", async () => { let finished = false;
const processed: any[] = []; const expected = [ Buffer.from("a"), Buffer.from("b"), Buffer.from("c"), ];
const read = new Readable({ read() {}, });
const write = new Writable({ write(data, _enc, cb) { processed.push(data); cb(); }, });
write.on("finish", () => { finished = true; });
for (let i = 0; i < expected.length; i++) { read.push(expected[i]); } read.push(null);
const [finishedCompleted, finishedCb] = mustCall( (err?: NodeErrorAbstraction | null) => { assert(!err); assert(finished); assertEquals(processed, expected); }, 1, );
pipeline(read, write, finishedCb);
await finishedCompleted;});
Deno.test("Pipeline fails on stream destroyed", async () => { const read = new Readable({ read() {}, });
const write = new Writable({ write(_data, _enc, cb) { cb(); }, });
read.push("data"); queueMicrotask(() => read.destroy());
const [pipelineExecuted, pipelineCb] = mustCall( (err?: NodeErrorAbstraction | null) => { assert(err); }, 1, ); pipeline(read, write, pipelineCb);
await pipelineExecuted;});
Deno.test("Pipeline exits on stream error", async () => { const read = new Readable({ read() {}, });
const transform = new Transform({ transform(_data, _enc, cb) { cb(new Error("kaboom")); }, });
const write = new Writable({ write(_data, _enc, cb) { cb(); }, });
const [readExecution, readCb] = mustCall(); read.on("close", readCb); const [closeExecution, closeCb] = mustCall(); transform.on("close", closeCb); const [writeExecution, writeCb] = mustCall(); write.on("close", writeCb);
const errorExecutions = [read, transform, write] .map((stream) => { const [execution, cb] = mustCall((err?: NodeErrorAbstraction | null) => { assertEquals(err, new Error("kaboom")); });
stream.on("error", cb); return execution; });
const [pipelineExecution, pipelineCb] = mustCall( (err?: NodeErrorAbstraction | null) => { assertEquals(err, new Error("kaboom")); }, ); const dst = pipeline(read, transform, write, pipelineCb);
assertStrictEquals(dst, write);
read.push("hello");
await readExecution; await closeExecution; await writeExecution; await Promise.all(errorExecutions); await pipelineExecution;});
Deno.test("Pipeline processes iterators correctly", async () => { let res = ""; const w = new Writable({ write(chunk, _encoding, callback) { res += chunk; callback(); }, });
const [pipelineExecution, pipelineCb] = mustCall( (err?: NodeErrorAbstraction | null) => { assert(!err); assertEquals(res, "helloworld"); }, ); pipeline( function* () { yield "hello"; yield "world"; }(), w, pipelineCb, );
await pipelineExecution;});
Deno.test("Pipeline processes async iterators correctly", async () => { let res = ""; const w = new Writable({ write(chunk, _encoding, callback) { res += chunk; callback(); }, });
const [pipelineExecution, pipelineCb] = mustCall( (err?: NodeErrorAbstraction | null) => { assert(!err); assertEquals(res, "helloworld"); }, ); pipeline( async function* () { await Promise.resolve(); yield "hello"; yield "world"; }(), w, pipelineCb, );
await pipelineExecution;});
Deno.test("Pipeline processes generators correctly", async () => { let res = ""; const w = new Writable({ write(chunk, _encoding, callback) { res += chunk; callback(); }, });
const [pipelineExecution, pipelineCb] = mustCall( (err?: NodeErrorAbstraction | null) => { assert(!err); assertEquals(res, "helloworld"); }, ); pipeline( function* () { yield "hello"; yield "world"; }, w, pipelineCb, );
await pipelineExecution;});
Deno.test("Pipeline processes async generators correctly", async () => { let res = ""; const w = new Writable({ write(chunk, _encoding, callback) { res += chunk; callback(); }, });
const [pipelineExecution, pipelineCb] = mustCall( (err?: NodeErrorAbstraction | null) => { assert(!err); assertEquals(res, "helloworld"); }, ); pipeline( async function* () { await Promise.resolve(); yield "hello"; yield "world"; }, w, pipelineCb, );
await pipelineExecution;});
Deno.test("Pipeline handles generator transforms", async () => { let res = "";
const [pipelineExecuted, pipelineCb] = mustCall( (err?: NodeErrorAbstraction | null) => { assert(!err); assertEquals(res, "HELLOWORLD"); }, ); pipeline( async function* () { await Promise.resolve(); yield "hello"; yield "world"; }, async function* (source: string[]) { for await (const chunk of source) { yield chunk.toUpperCase(); } }, async function (source: string[]) { for await (const chunk of source) { res += chunk; } }, pipelineCb, );
await pipelineExecuted;});
Deno.test("Pipeline passes result to final callback", async () => { const [pipelineExecuted, pipelineCb] = mustCall( (err?: NodeErrorAbstraction | null, val?: unknown) => { assert(!err); assertEquals(val, "HELLOWORLD"); }, ); pipeline( async function* () { await Promise.resolve(); yield "hello"; yield "world"; }, async function* (source: string[]) { for await (const chunk of source) { yield chunk.toUpperCase(); } }, async function (source: string[]) { let ret = ""; for await (const chunk of source) { ret += chunk; } return ret; }, pipelineCb, );
await pipelineExecuted;});
Deno.test("Pipeline returns a stream after ending", async () => { const [pipelineExecuted, pipelineCb] = mustCall( (err?: NodeErrorAbstraction | null) => { assertEquals(err, undefined); }, ); const ret = pipeline( async function* () { await Promise.resolve(); yield "hello"; }, async function* (source: string[]) { for await (const chunk of source) { chunk; } }, pipelineCb, );
ret.resume();
assertEquals(typeof ret.pipe, "function");
await pipelineExecuted;});