import { Buffer } from "../buffer.ts";import Readable from "../_stream/readable.ts";import { once } from "../events.ts";import { deferred } from "../../async/mod.ts";import { assert, assertEquals, assertStrictEquals,} from "../../testing/asserts.ts";
Deno.test("Readable stream from iterator", async () => { function* generate() { yield "a"; yield "b"; yield "c"; }
const stream = Readable.from(generate());
const expected = ["a", "b", "c"];
for await (const chunk of stream) { assertStrictEquals(chunk, expected.shift()); }});
Deno.test("Readable stream from async iterator", async () => { async function* generate() { yield "a"; yield "b"; yield "c"; }
const stream = Readable.from(generate());
const expected = ["a", "b", "c"];
for await (const chunk of stream) { assertStrictEquals(chunk, expected.shift()); }});
Deno.test("Readable stream from promise", async () => { const promises = [ Promise.resolve("a"), Promise.resolve("b"), Promise.resolve("c"), ];
const stream = Readable.from(promises);
const expected = ["a", "b", "c"];
for await (const chunk of stream) { assertStrictEquals(chunk, expected.shift()); }});
Deno.test("Readable stream from string", async () => { const string = "abc"; const stream = Readable.from(string);
for await (const chunk of stream) { assertStrictEquals(chunk, string); }});
Deno.test("Readable stream from Buffer", async () => { const string = "abc"; const stream = Readable.from(Buffer.from(string));
for await (const chunk of stream) { assertStrictEquals((chunk as Buffer).toString(), string); }});
Deno.test("Readable stream gets destroyed on error", async () => { async function* generate() { throw new Error("kaboom"); }
const stream = Readable.from(generate());
stream.read();
const [err] = await once(stream, "error"); assertStrictEquals(err.message, "kaboom"); assertStrictEquals(stream.destroyed, true);});
Deno.test("Readable stream works as Transform stream", async () => { async function* generate(stream: Readable) { for await (const chunk of stream) { yield (chunk as string).toUpperCase(); } }
const source = new Readable({ objectMode: true, read() { this.push("a"); this.push("b"); this.push("c"); this.push(null); }, });
const stream = Readable.from(generate(source));
const expected = ["A", "B", "C"];
for await (const chunk of stream) { assertStrictEquals(chunk, expected.shift()); }});
Deno.test("Readable stream can be paused", () => { const readable = new Readable();
readable._read = () => {};
assert(!readable.isPaused());
readable.on("data", () => {});
assert(!readable.isPaused());
readable.pause(); assert(readable.isPaused()); readable.resume(); assert(!readable.isPaused());});
Deno.test("Readable stream sets enconding correctly", () => { const readable = new Readable({ read() {}, });
readable.setEncoding("utf8");
readable.push(new TextEncoder().encode("DEF")); readable.unshift(new TextEncoder().encode("ABC"));
assertStrictEquals(readable.read(), "ABCDEF");});
Deno.test("Readable stream sets encoding correctly", () => { const readable = new Readable({ read() {}, });
readable.setEncoding("utf8");
readable.push(new TextEncoder().encode("DEF")); readable.unshift(new TextEncoder().encode("ABC"));
assertStrictEquals(readable.read(), "ABCDEF");});
Deno.test("Readable stream holds up a big push", async () => { let readExecuted = 0; const readExecutedExpected = 3; const readExpectedExecutions = deferred();
let endExecuted = 0; const endExecutedExpected = 1; const endExpectedExecutions = deferred();
const str = "asdfasdfasdfasdfasdf";
const r = new Readable({ highWaterMark: 5, encoding: "utf8", });
let reads = 0;
function _read() { if (reads === 0) { setTimeout(() => { r.push(str); }, 1); reads++; } else if (reads === 1) { const ret = r.push(str); assertEquals(ret, false); reads++; } else { r.push(null); } }
r._read = () => { readExecuted++; if (readExecuted == readExecutedExpected) { readExpectedExecutions.resolve(); } _read(); };
r.on("end", () => { endExecuted++; if (endExecuted == endExecutedExpected) { endExpectedExecutions.resolve(); } });
const ret = r.push(str); assert(!ret); let chunk = r.read(); assertEquals(chunk, str); chunk = r.read(); assertEquals(chunk, null);
r.once("readable", () => { chunk = r.read(); assertEquals(chunk, str + str);
chunk = r.read(); assertEquals(chunk, null); });
const readTimeout = setTimeout( () => readExpectedExecutions.reject(), 1000, ); const endTimeout = setTimeout( () => endExpectedExecutions.reject(), 1000, ); await readExpectedExecutions; await endExpectedExecutions; clearTimeout(readTimeout); clearTimeout(endTimeout); assertEquals(readExecuted, readExecutedExpected); assertEquals(endExecuted, endExecutedExpected);});
Deno.test("Readable stream: 'on' event", async () => { async function* generate() { yield "a"; yield "b"; yield "c"; }
const stream = Readable.from(generate());
let iterations = 0; const expected = ["a", "b", "c"];
stream.on("data", (chunk) => { iterations++; assertStrictEquals(chunk, expected.shift()); });
await once(stream, "end");
assertStrictEquals(iterations, 3);});
Deno.test("Readable stream: 'data' event", async () => { async function* generate() { yield "a"; yield "b"; yield "c"; }
const stream = Readable.from(generate(), { objectMode: false });
let iterations = 0; const expected = ["a", "b", "c"];
stream.on("data", (chunk) => { iterations++; assertStrictEquals(chunk instanceof Buffer, true); assertStrictEquals(chunk.toString(), expected.shift()); });
await once(stream, "end");
assertStrictEquals(iterations, 3);});
Deno.test("Readable stream: 'data' event on non-object", async () => { async function* generate() { yield "a"; yield "b"; yield "c"; }
const stream = Readable.from(generate(), { objectMode: false });
let iterations = 0; const expected = ["a", "b", "c"];
stream.on("data", (chunk) => { iterations++; assertStrictEquals(chunk instanceof Buffer, true); assertStrictEquals(chunk.toString(), expected.shift()); });
await once(stream, "end");
assertStrictEquals(iterations, 3);});
Deno.test("Readable stream: 'readable' event is emitted but 'read' is not on highWaterMark length exceeded", async () => { let readableExecuted = 0; const readableExecutedExpected = 1; const readableExpectedExecutions = deferred();
const r = new Readable({ highWaterMark: 3, });
r._read = () => { throw new Error("_read must not be called"); }; r.push(Buffer.from("blerg"));
setTimeout(function () { assert(!r._readableState.reading); r.on("readable", () => { readableExecuted++; if (readableExecuted == readableExecutedExpected) { readableExpectedExecutions.resolve(); } }); }, 1);
const readableTimeout = setTimeout( () => readableExpectedExecutions.reject(), 1000, ); await readableExpectedExecutions; clearTimeout(readableTimeout); assertEquals(readableExecuted, readableExecutedExpected);});
Deno.test("Readable stream: 'readable' and 'read' events are emitted on highWaterMark length not reached", async () => { let readableExecuted = 0; const readableExecutedExpected = 1; const readableExpectedExecutions = deferred();
let readExecuted = 0; const readExecutedExpected = 1; const readExpectedExecutions = deferred();
const r = new Readable({ highWaterMark: 3, });
r._read = () => { readExecuted++; if (readExecuted == readExecutedExpected) { readExpectedExecutions.resolve(); } };
r.push(Buffer.from("bl"));
setTimeout(function () { assert(r._readableState.reading); r.on("readable", () => { readableExecuted++; if (readableExecuted == readableExecutedExpected) { readableExpectedExecutions.resolve(); } }); }, 1);
const readableTimeout = setTimeout( () => readableExpectedExecutions.reject(), 1000, ); const readTimeout = setTimeout( () => readExpectedExecutions.reject(), 1000, ); await readableExpectedExecutions; await readExpectedExecutions; clearTimeout(readableTimeout); clearTimeout(readTimeout); assertEquals(readableExecuted, readableExecutedExpected); assertEquals(readExecuted, readExecutedExpected);});
Deno.test("Readable stream: 'readable' event is emitted but 'read' is not on highWaterMark length not reached and stream ended", async () => { let readableExecuted = 0; const readableExecutedExpected = 1; const readableExpectedExecutions = deferred();
const r = new Readable({ highWaterMark: 30, });
r._read = () => { throw new Error("Must not be executed"); };
r.push(Buffer.from("blerg")); r.push(null);
setTimeout(function () { assert(!r._readableState.reading); r.on("readable", () => { readableExecuted++; if (readableExecuted == readableExecutedExpected) { readableExpectedExecutions.resolve(); } }); }, 1);
const readableTimeout = setTimeout( () => readableExpectedExecutions.reject(), 1000, ); await readableExpectedExecutions; clearTimeout(readableTimeout); assertEquals(readableExecuted, readableExecutedExpected);});
Deno.test("Readable stream: 'read' is emitted on empty string pushed in non-object mode", async () => { let endExecuted = 0; const endExecutedExpected = 1; const endExpectedExecutions = deferred();
const underlyingData = ["", "x", "y", "", "z"]; const expected = underlyingData.filter((data) => data); const result: unknown[] = [];
const r = new Readable({ encoding: "utf8", }); r._read = function () { queueMicrotask(() => { if (!underlyingData.length) { this.push(null); } else { this.push(underlyingData.shift()); } }); };
r.on("readable", () => { const data = r.read(); if (data !== null) result.push(data); });
r.on("end", () => { endExecuted++; if (endExecuted == endExecutedExpected) { endExpectedExecutions.resolve(); } assertEquals(result, expected); });
const endTimeout = setTimeout( () => endExpectedExecutions.reject(), 1000, ); await endExpectedExecutions; clearTimeout(endTimeout); assertEquals(endExecuted, endExecutedExpected);});
Deno.test("Readable stream: listeners can be removed", () => { const r = new Readable(); r._read = () => {}; r.on("data", () => {});
r.removeAllListeners("data");
assertEquals(r.eventNames().length, 0);});