deno.land / std@0.91.0 / node / _stream / readable_test.ts
نووسراو ببینە
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490// Copyright Node.js contributors. All rights reserved. MIT License.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 () => { // deno-lint-ignore require-yield 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();
// _read is a noop, here. readable._read = () => {};
// Default state of a stream is not "paused" assert(!readable.isPaused());
// Make the stream start flowing... readable.on("data", () => {});
// still not paused. 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(); } });
// Push some data in to start. // We've never gotten any read event at this point. const ret = r.push(str); assert(!ret); let chunk = r.read(); assertEquals(chunk, str); chunk = r.read(); assertEquals(chunk, null);
r.once("readable", () => { // This time, we'll get *all* the remaining data, because // it's been added synchronously, as the read WOULD take // us below the hwm, and so it triggered a _read() again, // which synchronously added more, which we then return. 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")); //This ends the stream and triggers end r.push(null);
setTimeout(function () { // Assert we're testing what we think we are 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);});
std
Version Info
3 years ago
External Dependencies
No external dependencies 🎉
