effect-io-ai

Package: effect
Module: Stream

Stream.callback

Creates a stream from a callback that can emit values into a queue.

When to use

Use when you need callback-based code to emit stream values by offering to a Queue, or signal stream completion through the Queue module APIs.

By default it uses an “unbounded” buffer size. You can customize the buffer size and strategy by passing an object as the second argument with the bufferSize and strategy fields.

Example (Creating a stream from a callback that can emit values into a queue)

import { Console, Effect, Queue, Stream } from "effect"

const stream = Stream.callback<number>((queue) =>
  Effect.sync(() => {
    // Emit values to the stream
    Queue.offerUnsafe(queue, 1)
    Queue.offerUnsafe(queue, 2)
    Queue.offerUnsafe(queue, 3)
    // Signal completion
    Queue.endUnsafe(queue)
  })
)

const program = Effect.gen(function*() {
  const values = yield* stream.pipe(Stream.runCollect)
  yield* Console.log(values)
  // [ 1, 2, 3 ]
})

Effect.runPromise(program)

Signature

declare const callback: <A, E = never, R = never>(f: (queue: Queue.Queue<A, E | Cause.Done>) => Effect.Effect<unknown, E, R | Scope.Scope>, options?: { readonly bufferSize?: number | undefined; readonly strategy?: "sliding" | "dropping" | "suspend" | undefined; }) => Stream<A, E, Exclude<R, Scope.Scope>>

Source

Since v4.0.0