import { Readable, Transform, Writable } from "./stream.ts";import { Buffer } from "./buffer.ts";import { deferred } from "../async/mod.ts";import { assert, assertEquals } from "../testing/asserts.ts";import { mustCall } from "./_utils.ts";
Deno.test("Readable and Writable stream backpressure test", async () => { let pushes = 0; const total = 65500 + 40 * 1024;
let rsExecuted = 0; const rsExecutedExpected = 11; const rsExpectedExecutions = deferred();
let wsExecuted = 0; const wsExecutedExpected = 410; const wsExpectedExecutions = deferred();
const rs = new Readable({ read: function () { rsExecuted++; if (rsExecuted == rsExecutedExpected) { rsExpectedExecutions.resolve(); }
if (pushes++ === 10) { this.push(null); return; }
assert(this._readableState.length <= total);
this.push(Buffer.alloc(65500)); for (let i = 0; i < 40; i++) { this.push(Buffer.alloc(1024)); } }, });
const ws = new Writable({ write: function (_data, _enc, cb) { wsExecuted++; if (wsExecuted == wsExecutedExpected) { wsExpectedExecutions.resolve(); } cb(); }, });
rs.pipe(ws);
const rsTimeout = setTimeout(() => rsExpectedExecutions.reject(), 1000); const wsTimeout = setTimeout(() => wsExpectedExecutions.reject(), 1000); await rsExpectedExecutions; await wsExpectedExecutions; clearTimeout(rsTimeout); clearTimeout(wsTimeout); assertEquals(rsExecuted, rsExecutedExpected); assertEquals(wsExecuted, wsExecutedExpected);});
Deno.test("Readable can be piped through Transform", async () => { const [readExecution, readCb] = mustCall(function (this: Readable) { this.push("content"); this.push(null); });
const r = new Readable({ read: readCb, });
const [transformExecution, transformCb] = mustCall( function ( this: Transform, chunk: unknown, _e, callback: (error?: Error | null) => void, ) { this.push(chunk); callback(); }, );
const [flushExecution, flushCb] = mustCall( function (this: Transform, callback: (error?: Error | null) => void) { callback(); }, );
const t = new Transform({ transform: transformCb, flush: flushCb, });
r.pipe(t);
const [readableExecution, readableCb] = mustCall(function () { while (true) { const chunk = t.read(); if (!chunk) { break; }
assertEquals(chunk.toString(), "content"); } }, 2);
t.on("readable", readableCb);
await readExecution; await transformExecution; await flushExecution; await readableExecution;});