streams

TypeScript modules that provides utilities for Streams API.

Example

import { from } from "@milly/streams/readable/from";
import { fromMessage } from "@milly/streams/readable/from-message";
import { switchMap } from "@milly/streams/transform/switch-map";
import { take } from "@milly/streams/transform/take";
import { forEach } from "@milly/streams/writable/for-each";
import { postMessage } from "@milly/streams/writable/post-message";
import { assertEquals } from "@std/assert";

async function* gen() {
  yield 1;
  await Promise.resolve();
  yield "foo";
  await Promise.resolve();
  yield true;
}

const { port1, port2 } = new MessageChannel();

from(gen())
  .pipeThrough(switchMap((chunk) => [chunk, typeof chunk]))
  .pipeTo(postMessage(port1))
  .finally(() => port1.close());

const result: unknown[] = [];

await fromMessage(port2)
  .pipeThrough(take(6))
  .pipeTo(forEach((chunk) => {
    result.push(chunk);
  }))
  .finally(() => port2.close());

assertEquals(result, [
  1,
  "number",
  "foo",
  "string",
  true,
  "boolean",
]);

License

This library is licensed under the MIT License. See the LICENSE file for details.