Channel

import { Channel } from "jsr:@ende/async";

class

class Channel<T> implements AsyncIterable<T>, Disposable, AsyncDisposable

An async channel for communicating between concurrent tasks with optional bounded buffering and backpressure.

FIFO order: values are received in the order they were sent. When multiple senders or receivers are suspended, they are served in the order they arrived.
Backpressure: Channel.send suspends when the buffer is full (or always, when unbuffered) until a receiver consumes a value.
Close asymmetry: Channel.close accepts an optional reason. Pending and future Channel.receive calls reject with that reason (or a fresh ChannelClosedError when no reason was supplied). Pending Channel.send calls always reject with a ChannelClosedError carrying the unsent value, regardless of the close reason.
undefined values: undefined is a valid channel value, so non-blocking receives use the ChannelReceiveResult discriminated union rather than T | undefined.
Multiple consumers: concurrent Channel.receive calls and multiple Channel.toReadableStream instances each consume values FIFO; every value is delivered to exactly one consumer.

Example: Basic producer/consumer

import { Channel } from "@ende/async/channel";
import { assertEquals } from "@ende/assert";

const ch = new Channel<number>({ capacity: 4 });

await ch.send(1);
await ch.send(2);
ch.close();

const values: number[] = [];
for await (const v of ch) {
  values.push(v);
}
assertEquals(values, [1, 2]);

Example: Using await using for automatic cleanup

import { Channel } from "@ende/async/channel";
import { assert } from "@ende/assert";

let ref: Channel<string>;
{
  await using ch = new Channel<string>({ capacity: 8 });
  ref = ch;
  await ch.send("hello");
}
assert(ref.closed);

Constructors

`constructor`

new Channel(options: ChannelOptions = [UNSUPPORTED])

Creates a new channel.

options: ChannelOptions = [UNSUPPORTED]: Channel options. Defaults to an unbuffered (rendezvous) channel with capacity 0.

Methods

`send`

send(value: T, options?: ChannelSendOptions): Promise<void>

Sends a value into the channel. The returned promise resolves when the value is buffered or handed to a waiting receiver. Suspends if the buffer is full (or if unbuffered, suspends until a receiver calls Channel.receive).

value: T: The value to send into the channel.
options?: ChannelSendOptions: Optional settings for the send operation.

Returns

A promise that resolves when the value has been accepted.

Example: Usage

import { Channel } from "@ende/async/channel";
import { assertEquals } from "@ende/assert";

const ch = new Channel<number>({ capacity: 1 });
await ch.send(42);
assertEquals(ch.size, 1);
ch.close();

Example: Cancelling with an AbortSignal

import { Channel } from "@ende/async/channel";
import { assertRejects } from "@ende/assert";

const ch = new Channel<number>();
const controller = new AbortController();
const p = ch.send(42, { signal: controller.signal });
controller.abort(new Error("cancelled"));
await assertRejects(() => p, Error, "cancelled");

`receive`

receive(options?: ChannelReceiveOptions): Promise<T>

Receives a value from the channel. Suspends if the buffer is empty. Multiple concurrent calls are supported; each value is delivered to exactly one receiver in FIFO order.

options?: ChannelReceiveOptions: Optional settings for the receive operation.

Returns

A promise that resolves with the next value from the channel.

Example: Usage

import { Channel } from "@ende/async/channel";
import { assertEquals } from "@ende/assert";

const ch = new Channel<number>({ capacity: 1 });
await ch.send(42);
assertEquals(await ch.receive(), 42);
ch.close();

Example: Cancelling with an AbortSignal

import { Channel } from "@ende/async/channel";
import { assertRejects } from "@ende/assert";

const ch = new Channel<number>();
const controller = new AbortController();
const p = ch.receive({ signal: controller.signal });
controller.abort(new Error("cancelled"));
await assertRejects(() => p, Error, "cancelled");

`trySend`

trySend(value: T): boolean

Non-blocking send. Does not throw.

value: T: The value to send.

Returns

true if the value was delivered (buffered, or handed directly to a waiting receiver in the unbuffered case). false if the buffer is full, no receiver is waiting (unbuffered), or the channel is closed.

Example: Usage

import { Channel } from "@ende/async/channel";
import { assert, assertFalse } from "@ende/assert";

const ch = new Channel<number>({ capacity: 1 });
assert(ch.trySend(1));
assertFalse(ch.trySend(2));

`tryReceive`

tryReceive(): ChannelReceiveResult<T>

Non-blocking receive. Discriminate on the state field to determine the outcome without ambiguity, even when T itself can be undefined.

Returns

A ChannelReceiveResult — { state: "ok", value } if a value was available, { state: "empty" } if the channel is open but no value is ready, or { state: "closed" } if the channel has been closed and no buffered values remain.

Example: Usage

import { Channel } from "@ende/async/channel";
import { assertEquals } from "@ende/assert";

const ch = new Channel<number>({ capacity: 1 });
await ch.send(42);
assertEquals(ch.tryReceive(), { state: "ok", value: 42 });
assertEquals(ch.tryReceive(), { state: "empty" });
ch.close();
assertEquals(ch.tryReceive(), { state: "closed" });

`close`

close(): void

Closes the channel. Idempotent — calling close() on an already-closed channel is a no-op. Pending send() calls reject with ChannelClosedError. Pending receive() calls drain remaining buffered values, then reject.

Example: Usage

import { Channel } from "@ende/async/channel";
import { assert } from "@ende/assert";

const ch = new Channel<number>();
ch.close();
assert(ch.closed);

close(reason: unknown): void

Closes the channel with a reason. All pending and future receive() calls reject with reason after draining buffered values. Pending send() calls reject with ChannelClosedError (the reason is intentionally not propagated to senders, so they can recover the unsent value via ChannelClosedError.value). Idempotent.

reason: unknown: The reason to reject pending and future receivers with.

Example: Usage

import { Channel } from "@ende/async/channel";
import { assertRejects } from "@ende/assert";

const ch = new Channel<number>();
ch.close(new Error("upstream failure"));
await assertRejects(() => ch.receive(), Error, "upstream failure");

close(...args: [unknown] | []): void

`closed`

get closed(): boolean

Whether the channel has been closed.

Returns

true if the channel is closed, false otherwise.

Example: Usage

import { Channel } from "@ende/async/channel";
import { assert, assertFalse } from "@ende/assert";

const ch = new Channel<number>();
assertFalse(ch.closed);
ch.close();
assert(ch.closed);

`size`

get size(): number

Number of values currently buffered. Informational only — the value is inherently racy and should not be used for send/receive control flow.

Returns

The number of buffered values.

Example: Usage

import { Channel } from "@ende/async/channel";
import { assertEquals } from "@ende/assert";

const ch = new Channel<number>({ capacity: 4 });
await ch.send(1);
await ch.send(2);
assertEquals(ch.size, 2);
ch.close();

`capacity`

get capacity(): number

Maximum buffer capacity (0 for unbuffered).

Returns

The maximum buffer capacity.

Example: Usage

import { Channel } from "@ende/async/channel";
import { assertEquals } from "@ende/assert";

const ch = new Channel<number>({ capacity: 8 });
assertEquals(ch.capacity, 8);
ch.close();

`[Symbol.asyncIterator]`

[Symbol.asyncIterator](): AsyncIterableIterator<T>

Async iteration drains the channel until it is closed and empty.

Returns

An async iterator that yields values from the channel.

Example: Usage

import { Channel } from "@ende/async/channel";
import { assertEquals } from "@ende/assert";

const ch = new Channel<number>({ capacity: 4 });
await ch.send(1);
await ch.send(2);
ch.close();

const values: number[] = [];
for await (const v of ch) {
  values.push(v);
}
assertEquals(values, [1, 2]);

`toReadableStream`

toReadableStream(): ReadableStream<T>

Creates a ReadableStream that yields values from this channel. The stream closes when the channel closes after draining buffered values. If the channel was closed with a reason, the stream errors with that reason. Cancelling the stream closes the channel; a non-undefined cancel reason is forwarded to Channel.close so other consumers observe it.

Each call returns an independent consumer. If multiple streams (or streams alongside direct Channel.receive calls or async iteration) consume from the same channel concurrently, values are distributed FIFO and each value is delivered to exactly one consumer.

Returns

A readable stream of channel values.

Example: Usage

import { Channel } from "@ende/async/channel";
import { assertEquals } from "@ende/assert";

const ch = new Channel<number>({ capacity: 4 });
await ch.send(1);
await ch.send(2);
ch.close();

const values = await Array.fromAsync(ch.toReadableStream());
assertEquals(values, [1, 2]);

`[Symbol.dispose]`

[Symbol.dispose](): void

Calls Channel.close. Enables using for automatic cleanup.

Example: Usage

import { Channel } from "@ende/async/channel";
import { assert } from "@ende/assert";

const ch = new Channel<number>();
ch[Symbol.dispose]();
assert(ch.closed);

`[Symbol.asyncDispose]`

[Symbol.asyncDispose](): Promise<void>

Calls Channel.close. Enables await using for automatic cleanup in async contexts.

Example: Usage

import { Channel } from "@ende/async/channel";
import { assert } from "@ende/assert";

const ch = new Channel<number>();
await ch[Symbol.asyncDispose]();
assert(ch.closed);

Source