On this page

Multi-consumer

History
M

broadcast

History
broadcast(options?): Object
Attributes
options:Object
budget?:number
Buffer size in bytes. Must be >= 16384. Default: 65536.
backpressure?:string
'strict', 'unbounded', 'drop-oldest', or 'drop-newest'. Default: 'strict'.
signal:AbortSignal
Returns:Object
writer:Writable
broadcast:BroadcastChannel

Create a push-model multi-consumer broadcast channel. A single writer pushes data to multiple consumers. Each consumer has an independent cursor into a shared buffer.

import { broadcast, text } from 'node:stream/iter';

const { writer, broadcast: bc } = broadcast();

// Create consumers before writing
const c1 = bc.push();  // Consumer 1
const c2 = bc.push();  // Consumer 2

// Producer and consumers must run concurrently. Awaited writes
// block when the buffer fills until consumers read.
const producing = (async () => {
  await writer.write('hello');
  await writer.end();
})();

const [r1, r2] = await Promise.all([text(c1), text(c2)]);
console.log(r1); // 'hello'
console.log(r2); // 'hello'
await producing;
JavaScript
const { broadcast, text } = require('node:stream/iter');

async function run() {
  const { writer, broadcast: bc } = broadcast();

  // Create consumers before writing
  const c1 = bc.push();  // Consumer 1
  const c2 = bc.push();  // Consumer 2

  // Producer and consumers must run concurrently. Awaited writes
  // block when the buffer fills until consumers read.
  const producing = (async () => {
    await writer.write('hello');
    await writer.end();
  })();

  const [r1, r2] = await Promise.all([text(c1), text(c2)]);
  console.log(r1); // 'hello'
  console.log(r2); // 'hello'
  await producing;
}

run().catch(console.error);
JavaScript
 
broadcast.cancel(reason?): void
Attributes
reason:Error

Cancel the broadcast. All consumers receive an error.

Attributes
number

The number of active consumers.

broadcast.push(...transforms?, options?): AsyncIterable
Attributes
...transforms:Function | Object
options:Object
signal:AbortSignal
Returns:AsyncIterable
whose chunks fulfill with Uint8Array[]

Create a new consumer. Each consumer receives all data written to the broadcast from the point of subscription onward. Optional transforms are applied to this consumer's view of the data.

broadcast[Symbol.dispose](): void

Alias for broadcast.cancel().

M

Broadcast.from

History
Broadcast.from(input, options?): Object
Attributes
input:AsyncIterable | Iterable | BroadcastChannel
options:Object
Same as broadcast().
Returns:Object
{ writer, broadcast }

Create a BroadcastChannel from an existing source. The source is consumed automatically and pushed to all subscribers.

M

share

History
share(source, options?): Share
Attributes
source:AsyncIterable
The source to share.
options:Object
budget?:number
Buffer size in bytes. Must be >= 16384. Default: 65536.
backpressure?:string
'strict', 'unbounded', 'drop-oldest', or 'drop-newest'. Default: 'strict'.
Returns:Share

Create a pull-model multi-consumer shared stream. Unlike broadcast(), the source is only read when a consumer pulls. Multiple consumers share a single buffer.

import { from, share, text } from 'node:stream/iter';

const shared = share(from('hello'));

const c1 = shared.pull();
const c2 = shared.pull();

// Consume concurrently to avoid deadlock with small buffers.
const [r1, r2] = await Promise.all([text(c1), text(c2)]);
console.log(r1); // 'hello'
console.log(r2); // 'hello'
JavaScript
const { from, share, text } = require('node:stream/iter');

async function run() {
  const shared = share(from('hello'));

  const c1 = shared.pull();
  const c2 = shared.pull();

  // Consume concurrently to avoid deadlock with small buffers.
  const [r1, r2] = await Promise.all([text(c1), text(c2)]);
  console.log(r1); // 'hello'
  console.log(r2); // 'hello'
}

run().catch(console.error);
JavaScript
S

Share.from

History
Share.from(input, options?): Share
Attributes
input:AsyncIterable | Shareable
options:Object
Same as share().
Returns:Share

Create a Share from an existing source.

share.cancel(reason?): void
Attributes
reason:Error

Cancel the share. All consumers receive an error.

Attributes
number

The number of active consumers.

share.pull(...transforms?, options?): AsyncIterable
Attributes
...transforms:Function | Object
options:Object
signal:AbortSignal
Returns:AsyncIterable
whose chunks fulfill with Uint8Array[]

Create a new consumer of the shared source.

share[Symbol.dispose](): void

Alias for share.cancel().

Attributes
Function
that returns a Share.
Attributes
Function
that returns a SyncShare.
M

shareSync

History
shareSync(source, options?): SyncShare
Attributes
source:Iterable
The sync source to share.
options:Object
budget?:number
Must be >= 16384. Default: 65536.
backpressure?:string
Default: 'strict'.
Returns:SyncShare

Synchronous version of share().

S

SyncShare.fromSync

History
SyncShare.fromSync(input, options?): SyncShare
Attributes
input:Iterable | SyncShareable
options:Object
Returns:SyncShare
Attributes
number

The number of chunks currently buffered.

share.cancel(reason?): void
Attributes
reason:Error

Cancel the share. All consumers receive an error.

Attributes
number

The number of active consumers.

share.pull(...transforms?, options?): Iterable
Attributes
...transforms:Function | Object
options:Object
signal:AbortSignal
Returns:Iterable
whose chunks return Uint8Array[]

Create a new consumer of the shared source.

share[Symbol.dispose](): void

Alias for share.cancel().

PreviousUtilitiesNextClassic stream interop