import { Buffer } from "../buffer.ts";import finished from "./end_of_stream.ts";import Writable from "../_stream/writable.ts";import { deferred } from "../../async/mod.ts";import { assert, assertEquals, assertStrictEquals, assertThrows,} from "../../testing/asserts.ts";
Deno.test("Writable stream writes correctly", async () => { let callback: undefined | ((error?: Error | null | undefined) => void);
let writeExecuted = 0; const writeExecutedExpected = 1; const writeExpectedExecutions = deferred();
let writevExecuted = 0; const writevExecutedExpected = 1; const writevExpectedExecutions = deferred();
const writable = new Writable({ write: (chunk, encoding, cb) => { writeExecuted++; if (writeExecuted == writeExecutedExpected) { writeExpectedExecutions.resolve(); } assert(chunk instanceof Buffer); assertStrictEquals(encoding, "buffer"); assertStrictEquals(String(chunk), "ABC"); callback = cb; }, writev: (chunks) => { writevExecuted++; if (writevExecuted == writevExecutedExpected) { writevExpectedExecutions.resolve(); } assertStrictEquals(chunks.length, 2); assertStrictEquals(chunks[0].encoding, "buffer"); assertStrictEquals(chunks[1].encoding, "buffer"); assertStrictEquals(chunks[0].chunk + chunks[1].chunk, "DEFGHI"); }, });
writable.write(new TextEncoder().encode("ABC")); writable.write(new TextEncoder().encode("DEF")); writable.end(new TextEncoder().encode("GHI")); callback?.();
const writeTimeout = setTimeout( () => writeExpectedExecutions.reject(), 1000, ); const writevTimeout = setTimeout( () => writevExpectedExecutions.reject(), 1000, ); await writeExpectedExecutions; await writevExpectedExecutions; clearTimeout(writeTimeout); clearTimeout(writevTimeout); assertEquals(writeExecuted, writeExecutedExpected); assertEquals(writevExecuted, writevExecutedExpected);});
Deno.test("Writable stream writes Uint8Array in object mode", async () => { let writeExecuted = 0; const writeExecutedExpected = 1; const writeExpectedExecutions = deferred();
const ABC = new TextEncoder().encode("ABC");
const writable = new Writable({ objectMode: true, write: (chunk, encoding, cb) => { writeExecuted++; if (writeExecuted == writeExecutedExpected) { writeExpectedExecutions.resolve(); } assert(!(chunk instanceof Buffer)); assert(chunk instanceof Uint8Array); assertEquals(chunk, ABC); assertEquals(encoding, "utf8"); cb(); }, });
writable.end(ABC);
const writeTimeout = setTimeout( () => writeExpectedExecutions.reject(), 1000, ); await writeExpectedExecutions; clearTimeout(writeTimeout); assertEquals(writeExecuted, writeExecutedExpected);});
Deno.test("Writable stream throws on unexpected close", async () => { let finishedExecuted = 0; const finishedExecutedExpected = 1; const finishedExpectedExecutions = deferred();
const writable = new Writable({ write: () => {}, }); writable.writable = false; writable.destroy();
finished(writable, (err) => { finishedExecuted++; if (finishedExecuted == finishedExecutedExpected) { finishedExpectedExecutions.resolve(); } assertEquals(err?.code, "ERR_STREAM_PREMATURE_CLOSE"); });
const finishedTimeout = setTimeout( () => finishedExpectedExecutions.reject(), 1000, ); await finishedExpectedExecutions; clearTimeout(finishedTimeout); assertEquals(finishedExecuted, finishedExecutedExpected);});
Deno.test("Writable stream finishes correctly", async () => { let finishedExecuted = 0; const finishedExecutedExpected = 1; const finishedExpectedExecutions = deferred();
const w = new Writable({ write(_chunk, _encoding, cb) { cb(); }, autoDestroy: false, });
w.end("asd");
queueMicrotask(() => { finished(w, () => { finishedExecuted++; if (finishedExecuted == finishedExecutedExpected) { finishedExpectedExecutions.resolve(); } }); });
const finishedTimeout = setTimeout( () => finishedExpectedExecutions.reject(), 1000, ); await finishedExpectedExecutions; clearTimeout(finishedTimeout); assertEquals(finishedExecuted, finishedExecutedExpected);});
Deno.test("Writable stream finishes correctly after error", async () => { let errorExecuted = 0; const errorExecutedExpected = 1; const errorExpectedExecutions = deferred();
let finishedExecuted = 0; const finishedExecutedExpected = 1; const finishedExpectedExecutions = deferred();
const w = new Writable({ write(_chunk, _encoding, cb) { cb(new Error()); }, autoDestroy: false, }); w.write("asd"); w.on("error", () => { errorExecuted++; if (errorExecuted == errorExecutedExpected) { errorExpectedExecutions.resolve(); } finished(w, () => { finishedExecuted++; if (finishedExecuted == finishedExecutedExpected) { finishedExpectedExecutions.resolve(); } }); });
const errorTimeout = setTimeout( () => errorExpectedExecutions.reject(), 1000, ); const finishedTimeout = setTimeout( () => finishedExpectedExecutions.reject(), 1000, ); await finishedExpectedExecutions; await errorExpectedExecutions; clearTimeout(finishedTimeout); clearTimeout(errorTimeout); assertEquals(finishedExecuted, finishedExecutedExpected); assertEquals(errorExecuted, errorExecutedExpected);});
Deno.test("Writable stream fails on 'write' null value", () => { const writable = new Writable(); assertThrows(() => writable.write(null));});