Async Duplex

A pair of connected async streams where one acts as reader and the other as writer, and reverse() swaps their roles.

Install

npm install @silyze/async-duplex

Overview

AsyncDuplex<TOutput, TInput = TOutput> provides a simple duplex abstraction over two connected asynchronous streams. It supports:

• Reading from one stream and writing to the other.
• Flipping roles using reverse() — useful for building symmetrical or bidirectional systems.
• Composing with AsyncTransform for streaming transformations.

API

new AsyncDuplex<TOutput, TInput>()

Creates a new duplex stream with output type TOutput and input type TInput.

read(signal?: AbortSignal): AsyncIterable<TOutput>

Returns an async iterable for reading from the output side.

write(input: TInput): Promise<void>

Writes data to the input side.

transform(): AsyncTransform<TOutput>

Returns a transform object for piping or modifying output data.

reverse(): AsyncStream<TInput, TOutput>

Returns a new AsyncStream where the original input becomes readable and the original output becomes writable — reversing the flow.

Example: Duplex Console Echo

import { AsyncDuplex } from "@silyze/async-duplex";
import { AsyncTransform } from "@mojsoski/async-stream";
import readline from "readline";

// Create a duplex console
const console = new AsyncDuplex<string>();
const consoleImpl = console.reverse(); // reversed interface

// Set up signal aborts on process exit
const stdoutAbortController = new AbortController();
process.stdout.once("close", () => stdoutAbortController.abort());

const stdinAbortController = new AbortController();
process.stdin.once("close", () => stdinAbortController.abort());

// Pipe data written to the duplex into stdout
consoleImpl.transform().pipe(
  {
    async write(input) {
      await new Promise<void>((resolve, reject) => {
        process.stdout.write(input, (err) => (err ? reject(err) : resolve()));
      });
    },
  },
  { signal: stdoutAbortController.signal }
);

// Read from stdin using readline and pipe into the duplex
const inputInterface = readline.promises.createInterface(process.stdin);
AsyncTransform.from(inputInterface).pipe(consoleImpl, {
  signal: stdinAbortController.signal,
});

// Read from the duplex, prefix the input, and write it back to the stream
console
  .transform()
  .map((item) => `prefix: ${item}\n`)
  .pipe(console, { signal: stdinAbortController.signal })
  .then(() => process.exit(0));