writable_test.ts

// Copyright Node.js contributors. All rights reserved. MIT License.
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));
});