Creating Streams

Build a Stream from values, effectful work, or an existing source. Construction stays lazy with respect to stream execution: callbacks that perform Effect work run when the stream is consumed. The introduction explains trusted Stream builders and how producers work with Streams.

Finite and empty streams

// run: Console.log(JSON.stringify([labels.runCollect(), absent.runCollect(), signal.runCount(), pages.runCollect(), readPowers()]))
const labels = Stream.make("queued", "running", "complete")
const absent = Stream.empty
const signal = Stream.succeed(void 0)
const pages = Stream.range(4, 7)
const powers = Stream.iterate(2, (n) => n * 2).take(4)
 
function readPowers() {
  return powers.runCollect() // [2, 4, 8, 16]
}
// Output: [["queued","running","complete"],[],1,[4,5,6,7],[2,4,8,16]]

make accepts individual elements and empty emits none. range includes both endpoints. iterate emits the initial value, then repeatedly applies its pure callback; limit an infinite source before collecting it.

Successes and failures

// run: Console.log(JSON.stringify([ready.runCollect(), readClosed()]))
error SourceClosed
 
const ready = Stream.succeed("ready")
const closed = Stream.fail(new SourceClosed())
 
function readClosed() {
  try {
    return closed.runCollect()
  } catch (error) {
    return ["source unavailable"]
  }
}
// Output: [["ready"],["source unavailable"]]

succeed emits once and completes. fail builds a failing source; it does not throw at construction. A try around consumption handles its typed failure.

Arrays and iterables

// run: provide Catalog = { codes: () => ["A", "B"] } { Console.log(JSON.stringify([fromArray.runCollect(), fromBatches.runCollect(), fromSet.runCollect(), codes.runCollect()])) }
const fromArray = Stream.fromArray([11, 22])
const fromBatches = Stream.fromArrays([11, 22], [33])
const fromSet = Stream.fromIterable(new Set(["east", "west"]))
 
service Catalog {
  codes(): ReadonlyArray<string>!!
}
 
const codes = Stream.fromIterableEffect(() => Catalog.codes())
// Output: [[11,22],[11,22,33],["east","west"],["A","B"]]

fromArray and fromIterable emit collection members. fromArrays preserves the supplied batch boundaries internally. fromIterableEffect first runs the program to obtain an iterable, then emits its members; service requirements belong to consumption.

Async iterables

// TypeScript (.ts)
// rows.ts
// run: return (async () => { const values = []; for await (const row of rows()) values.push(row); console.log(JSON.stringify(values)); try { for await (const row of rows(true)) {} } catch (error) { console.log((error as Error).message) } })()
/** @vibelang trusted */
export function rows(fail = false): AsyncIterable<string> {
  return {
    async *[Symbol.asyncIterator]() {
      if (fail) throw new Error("iterator failed")
      yield "row-a"
      yield "row-b"
    }
  }
}
// Output: ["row-a","row-b"]
// Output: iterator failed

// run: Console.log(JSON.stringify(adaptRows(rows()).runCollect())); try { adaptRows(rows(true)).runCollect() } catch (error) { Console.log(error._tag) }
import { rows } from "./rows.js"
 
error IteratorFailed {
  cause: unknown
}
 
function adaptRows(rows: AsyncIterable<string>) {
  return Stream.fromAsyncIterable(rows, (cause) => new IteratorFailed({ cause }))
}
// Output: ["row-a","row-b"]
// Output: IteratorFailed

Pass an async iterable obtained from a TypeScript adapter or a service. The error mapper turns iteration exceptions and rejections into the stream's failure type. VibeLang has no async generator syntax or for await; use Stream consumers after adaptation.

Effects and callbacks

// run: const clock = Clock.Clock(); provide Clock = { ...clock, sleep: (duration) => clock.sleep(duration), currentTimeMillis: () => 123, currentTimeMillisUnsafe: () => 123 } { Console.log(JSON.stringify([timestamp.runCollect(), readNotices()])) }
const timestamp = Stream.fromEffect(Clock.currentTimeMillis)
 
const notices = Stream.callback<string>((queue) => {
  queue.offer("connected")
  queue.offer("ready")
  queue.end()
})
 
function readNotices() {
  return notices.runCollect() // ["connected", "ready"]
}
// Output: [[123],["connected","ready"]]

fromEffect emits one success value. callback gives its registration program a queue: Queue.offer emits an element, Queue.end completes normally, and Queue.fail terminates with a typed error. Its optional second argument accepts bufferSize and a strategy of "suspend", "dropping", or "sliding".

For a foreign event API, put the raw callback registration in a TypeScript adapter and register unsubscription with Effect.acquireRelease. The registration callback's Scope remains open for the stream lifetime. Platform callbacks cannot call VibeLang programs directly as synchronous JavaScript callbacks.

Scoped sources

// run: Console.log(JSON.stringify(sessionNames.runCollect()))
function openSession() {
  Console.log("open session")
  return Effect.acquireRelease(
    () => ({ name: "metrics" }),
    () => Console.log("close session")
  )
}
 
const sessionNames = Stream.fromEffect(openSession()).map((session) => session.name).scoped()
// Output: open session
// Output: close session
// Output: ["metrics"]

Stream.scoped takes a Stream, and supplies its Scope during consumption. Effect.acquireRelease registers release in that Scope. Place all operations that use the acquired resource inside the scoped stream; see Resourceful Streams.

Repeating values, streams, and effects

// run: Console.log(JSON.stringify([pings.take(3).runCollect(), rounds.runCollect(), times.take(2).runCount(), countdown()]))
const pings = Stream.succeed("ping").forever()
const rounds = Stream.make("start", "finish").repeat(Schedule.recurs(2))
const times = Stream.fromEffectRepeat(() => Clock.currentTimeMillis())
 
function countdown() {
  let remaining = 3
  const source = Stream.fromEffectRepeat {
    if (remaining === 0) Cause.done()
    return remaining--
  }
  return source.runCollect() // [3, 2, 1]
}
// Output: [["ping","ping","ping"],["start","finish","start","finish","start","finish"],2,[3,2,1]]

forever restarts its stream indefinitely. repeat runs the original stream once and lets the schedule control additional runs; rounds emits three rounds. fromEffectRepeat reruns its program for each element and treats Cause.Done as normal completion; Cause.done() runs an Effect that signals completion.

Keep mutable per-run state inside a function, as in countdown, so separate consumptions do not share it. Stream.take is another way to bound repetition.

Periodic elements

// run: const clock = Clock.Clock(); provide Clock = { ...clock, sleep: (duration) => clock.sleep(duration), currentTimeMillis: () => 123, currentTimeMillisUnsafe: () => 123 } { Console.log(JSON.stringify(collectSamples())) }
const samples = Stream.tick("250 millis").mapEffect(() => Clock.currentTimeMillis())
 
function collectSamples() {
  return samples.take(4).runCollect()
}
// Output: [123,123,123,123]

tick emits void on an interval. Map each signal to the value you need. Backpressure still applies; a tick stream is not a guarantee that slow work completes at that frequency.

Unfolding state and paginating responses

// run: provide Directory = { page: (cursor) => ({ names: [String(cursor)], next: cursor < 2 ? cursor + 1 : undefined }) } { Console.log(JSON.stringify([halving.runCollect(), names.runCollect()])) }
const halving = Stream.unfold(32, (n) => {
  if (n < 2) return undefined
  return [n, n / 2] as const
})
 
service Directory {
  page(cursor: number): Effect<{
    names: ReadonlyArray<string>
    next: number | undefined
  }, never, never>
}
 
const names = Stream.paginate(0, (cursor) => {
  const page = Directory.page(cursor)
  return [page.names, Option.fromUndefinedOr(page.next)] as const
})
// Output: [[32,16,8,4,2],["0","1","2"]]

unfold returns either [element, nextState] or undefined to stop without emitting. Its callback expects an Effect, so ordinary VibeLang statements can perform work between steps. paginate instead returns [batch, nextStateOption]; it emits the whole batch, including the final page whose next state is None.

Queues and broadcasts

// run: Console.log(JSON.stringify(queuedNumbers())); const bus = PubSub.unbounded<string>({ replay: 1 }); PubSub.publish(bus, "broadcast"); Console.log(JSON.stringify(fromBroadcast(bus).take(1).runCollect()))
function queuedNumbers() {
  const queue = Queue.unbounded<number, Cause.Done>()
  queue.offer(8)
  queue.offer(13)
  queue.end()
  return Stream.fromQueue(queue).runCollect() // [8, 13]
}
 
function fromBroadcast(bus: PubSub.PubSub<string>) {
  return Stream.fromPubSub(bus)
}
// Output: [8,13]
// Output: ["broadcast"]

fromQueue removes items from a shared queue, so competing consumers divide the work. fromPubSub creates a subscription during consumption, so subscribers each receive published elements. A plain PubSub source stays live until the bus ends or the consumer stops; publishing before subscription does not guarantee delivery without replay configuration.

Schedules as sources

// run: Console.log(JSON.stringify(readPulses()))
const policy = Schedule.spaced("400 millis").upTo({ times: 3 })
const pulses = Stream.fromSchedule(policy)
 
function readPulses() {
  return pulses.runCollect() // [0, 1, 2]
}
// Output: [0,1,2]

fromSchedule emits schedule outputs until the schedule stops. Stream.fromEffectSchedule(() => work(), policy) instead emits repeated program results under a schedule. Both preserve the schedule's timing and service requirements.

The matching Effect page: effect.website/docs/v4/stream/creating

Suggest changes to this page