Stream #

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

// Building a stream does nothing. This tap would print on every element,
// but nothing prints until the stream is run.
const numbers = Stream.range(1, 3).pipe(
  Stream.tap((n) => Effect.sync(() => console.log("producing", n)))
)

const program = Effect.gen(function* () {
  console.log("built, nothing produced yet")

  const collected = yield* Stream.runCollect(numbers)      // now it runs
  console.log("collected", collected)

  console.log("make", yield* Stream.runCollect(Stream.make("a", "b")))
  console.log("fromIterable", yield* Stream.runCollect(Stream.fromIterable(new Set([1, 1, 2]))))
  console.log("fromEffect", yield* Stream.runCollect(Stream.fromEffect(Effect.succeed(42))))

  // A queue is a source that someone else fills. Queue.end tells the stream "no more".
  const queue = yield* Queue.unbounded<string, Cause.Done>()
  yield* Queue.offerAll(queue, ["job-1", "job-2"])
  yield* Queue.end(queue)
  console.log("fromQueue", yield* Stream.runCollect(Stream.fromQueue(queue)))
})

Effect.runPromise(program)

import { Effect, Stream } from "effect"

const transactions = Stream.make(50, -20, 30, -5, 100, -60)

const program = Effect.gen(function* () {
  const bigDeposits = yield* transactions.pipe(
    Stream.filter((n) => n > 0),               // keep deposits
    Stream.map((n) => n * 100),                // to cents
    Stream.take(2),                            // first two only
    Stream.runCollect
  )
  console.log("first two deposits in cents", bigDeposits)

  const untilFirstWithdrawal = yield* transactions.pipe(
    Stream.takeWhile((n) => n > 0),
    Stream.runCollect
  )
  console.log("until first withdrawal", untilFirstWithdrawal)

  const balances = yield* transactions.pipe(
    Stream.scan(0, (balance, n) => balance + n),   // emits 0, then every running balance
    Stream.drop(1),                                // skip the initial 0
    Stream.runCollect
  )
  console.log("running balance", balances)
})

Effect.runPromise(program)

import { Effect, Option, Stream } from "effect"

const scores = Stream.make(72, 95, 88, 61)

const program = Effect.gen(function* () {
  const first = yield* Stream.runHead(scores)
  console.log("head", Option.getOrElse(first, () => -1))

  const last = yield* Stream.runLast(scores)
  console.log("last", Option.getOrElse(last, () => -1))

  const empty = yield* Stream.runHead(Stream.empty)
  console.log("head of empty", Option.isNone(empty) ? "none" : "some")

  // runFold: the initial value is a function so every run starts fresh
  const best = yield* Stream.runFold(scores, () => 0, (max, n) => Math.max(max, n))
  console.log("best", best)

  console.log("count", yield* Stream.runCount(scores))

  yield* Stream.runForEach(scores, (n) => Effect.sync(() => console.log("grade", n >= 70 ? "pass" : "fail")))

  // runDrain: values are thrown away, only the tap's effect is kept
  yield* scores.pipe(
    Stream.tap((n) => Effect.sync(() => console.log("seen", n))),
    Stream.take(2),
    Stream.runDrain
  )
})

Effect.runPromise(program)

import { Effect, Option, Schedule, Stream } from "effect"

// Every natural number. Only a description; nothing is computed here.
const naturals = Stream.iterate(0, (n) => n + 1)

// A fake paginated API: pages of two, four pages. Counts how often it is hit.
let fetches = 0
const pages: Record<number, Array<string>> = { 0: ["a", "b"], 1: ["c", "d"], 2: ["e", "f"], 3: ["g"] }
const fetchPage = (page: number) =>
  Effect.sync(() => {
    fetches++
    const next = page < 3 ? Option.some(page + 1) : Option.none<number>()
    return [pages[page]!, next] as const      // [items on this page, next cursor]
  })

const program = Effect.gen(function* () {
  const squares = yield* naturals.pipe(
    Stream.map((n) => n * n),
    Stream.filter((n) => n % 2 === 0),
    Stream.take(5),                             // without this line the program never ends
    Stream.runCollect
  )
  console.log("first five even squares", squares)

  const ticks = yield* Stream.tick("5 millis").pipe(Stream.take(3), Stream.runCount)
  console.log("ticks", ticks)

  const steps = yield* Stream.fromSchedule(Schedule.recurs(3)).pipe(Stream.runCollect)
  console.log("schedule steps", steps)

  const three = yield* Stream.paginate(0, fetchPage).pipe(Stream.take(3), Stream.runCollect)
  console.log("took", three, "with", fetches, "fetches")

  fetches = 0
  const all = yield* Stream.paginate(0, fetchPage).pipe(Stream.runCollect)
  console.log("took", all, "with", fetches, "fetches")
})

Effect.runPromise(program)

import { Effect, Stream } from "effect"

let inFlight = 0
let peak = 0

// A fake API call. Tracks how many are running at the same time.
const fetchUser = (id: number) =>
  Effect.gen(function* () {
    inFlight++
    peak = Math.max(peak, inFlight)
    yield* Effect.sleep("10 millis")
    inFlight--
    return "user-" + id
  })

const ids = Stream.range(1, 6)

const program = Effect.gen(function* () {
  const serial = yield* ids.pipe(Stream.mapEffect(fetchUser), Stream.runCollect)
  console.log("serial    ", serial, "peak", peak)

  peak = 0
  const parallel = yield* ids.pipe(
    Stream.mapEffect(fetchUser, { concurrency: 3 }),   // three at a time, order preserved
    Stream.runCollect
  )
  console.log("concurrent", parallel, "peak", peak)
})

Effect.runPromise(program)

import { Effect, Stream } from "effect"

const names = Stream.make("ada", "lin", "bo")
const errorsFromA = Stream.make("A1", "A2")
const errorsFromB = Stream.make("B1")

const program = Effect.gen(function* () {
  const numbered = yield* names.pipe(
    Stream.zipWith(Stream.iterate(1, (n) => n + 1), (name, i) => i + ". " + name),
    Stream.runCollect
  )
  console.log("zipWith", numbered)

  const merged = yield* errorsFromA.pipe(Stream.merge(errorsFromB), Stream.runCollect)
  console.log("merge", merged.sort())                 // sorted: merge order is not guaranteed

  const expanded = yield* names.pipe(
    Stream.flatMap((name) => name.length === 3 ? Stream.make(name, name.toUpperCase()) : Stream.empty),
    Stream.runCollect
  )
  console.log("flatMap", expanded)

  const batches = yield* Stream.range(1, 7).pipe(Stream.grouped(3), Stream.runCollect)
  console.log("grouped", batches)

  const pairs = yield* names.pipe(Stream.zip(Stream.make(true, false)), Stream.runCollect)
  console.log("zip stops at the shorter", pairs)
})

Effect.runPromise(program)

import { Cause, Effect, PubSub, Queue, Stream } from "effect"

// A fake event source. It calls onEvent 3 times, 1 ms apart, then it calls onDone.
// The returned function stops the source. Most event emitters have this shape.
const listen = (onEvent: (n: number) => void, onDone: () => void) => {
  let count = 0
  const timer = setInterval(() => {
    count++
    onEvent(count)
    if (count === 3) {
      clearInterval(timer)
      onDone()
    }
  }, 1)
  return () => {
    clearInterval(timer)
    console.log("listener removed")
  }
}

// Stream.callback gives you a queue. Push into it from plain callbacks. End it when the source is done.
const events = Stream.callback<number>((queue) =>
  Effect.acquireRelease(
    Effect.sync(() => listen((n) => Queue.offerUnsafe(queue, n), () => Queue.endUnsafe(queue))),
    (stop) => Effect.sync(stop)                  // runs when the stream ends, also after take
  )
)

const program = Effect.gen(function* () {
  // 1. A queue. A producer fiber offers, then ends. The stream ends with the queue.
  const queue = yield* Queue.bounded<string, Cause.Done>(2)
  yield* Effect.forkChild(Effect.gen(function* () {
    yield* Queue.offerAll(queue, ["job-1", "job-2", "job-3", "job-4"])   // waits while the queue is full
    yield* Queue.end(queue)
    console.log("producer ended")
  }))
  console.log("queue", yield* Stream.runCollect(Stream.fromQueue(queue)))

  // 2. A PubSub. Subscribe first, in a scope. Then publish. Each subscriber gets each message.
  const pubsub = yield* PubSub.unbounded<string>()
  const seen = yield* Effect.scoped(Effect.gen(function* () {
    const subscription = yield* PubSub.subscribe(pubsub)     // removed when the scope closes
    yield* PubSub.publishAll(pubsub, ["login", "click", "logout"])
    return yield* Stream.fromSubscription(subscription).pipe(Stream.take(3), Stream.runCollect)
  }))
  console.log("pubsub", seen)

  // 3. A callback source. The release runs in both cases.
  console.log("callback", yield* Stream.runCollect(events))
  console.log("callback take 2", yield* events.pipe(Stream.take(2), Stream.runCollect))
})

Effect.runPromise(program)

import { Effect, Stream } from "effect"

// A fake file. acquireRelease pairs the open with the close.
const openFile = (name: string) =>
  Effect.acquireRelease(
    Effect.sync(() => {
      console.log("open", name)
      return ["l1", "l2", "l3", "l4"]
    }),
    () => Effect.sync(() => console.log("close", name))
  )

// Stream.scoped: the resource lives as long as 1 run of the stream
const lines = Stream.scoped(Stream.fromEffect(openFile("a.txt"))).pipe(
  Stream.flatMap((all) => Stream.fromIterable(all)),
  Stream.tap((line) => Effect.sync(() => console.log("read", line))),
  Stream.ensuring(Effect.sync(() => console.log("ensuring: always"))),
  Stream.onEnd(Effect.sync(() => console.log("onEnd: only after a complete run")))
)

const program = Effect.gen(function* () {
  console.log("--- complete run")
  console.log(yield* Stream.runCollect(lines))

  console.log("--- take(2) stops the stream early")
  console.log(yield* lines.pipe(Stream.take(2), Stream.runCollect))

  console.log("--- a failure in the middle")
  const result = yield* lines.pipe(
    Stream.mapEffect((line) => line === "l3" ? Effect.fail("bad " + line) : Effect.succeed(line)),
    Stream.runCollect,
    Effect.catch((e) => Effect.succeed("failed: " + e))
  )
  console.log(result)
})

Effect.runPromise(program)

import { Effect, Stream } from "effect"

const orders = Stream.make(
  { city: "cairo", total: 10 },
  { city: "oslo", total: 5 },
  { city: "cairo", total: 7 },
  { city: "lima", total: 3 },
  { city: "oslo", total: 1 }
)

const program = Effect.gen(function* () {
  // groupByKey: 1 substream for each key. Reduce each substream to 1 value.
  // The groups fill at the same time, so the step that reads them must run with concurrency.
  const perCity = yield* orders.pipe(
    Stream.groupByKey((order) => order.city),
    Stream.mapEffect(
      ([city, group]) =>
        Stream.runFold(group, () => 0, (sum, order) => sum + order.total).pipe(
          Effect.map((sum) => city + "=" + sum)
        ),
      { concurrency: "unbounded" }
    ),
    Stream.runCollect
  )
  console.log("groupByKey", perCity)

  // buffer: the producer runs ahead of the consumer, up to the capacity
  const produce = (label: string) =>
    Stream.range(1, 3).pipe(
      Stream.rechunk(1),                    // 1 element for each chunk, so each element is 1 step
      Stream.tap((n) => Effect.sync(() => console.log(label, "produced", n)))
    )
  const consume = (label: string) =>
    Stream.runForEach((n: number) => Effect.sync(() => console.log(label, "consumed", n)))
  yield* produce("plain").pipe(consume("plain"))
  yield* produce("buffer").pipe(Stream.buffer({ capacity: 4 }), consume("buffer"))

  // broadcast: 1 source, 2 consumers, each consumer sees each element. It needs a scope.
  const [sum, max] = yield* Effect.scoped(Effect.gen(function* () {
    const shared = yield* Stream.broadcast(Stream.make(3, 1, 4, 1, 5), { capacity: 8 })
    return yield* Effect.all([
      Stream.runFold(shared, () => 0, (a, b) => a + b),
      Stream.runFold(shared, () => 0, (a, b) => Math.max(a, b))
    ], { concurrency: "unbounded" })
  }))
  console.log("broadcast sum", sum, "max", max)

  // changes: drop an element that is equal to the previous element
  const status = Stream.make("ok", "ok", "down", "down", "down", "ok")
  console.log("changes", yield* status.pipe(Stream.changes, Stream.runCollect))
})

Effect.runPromise(program)

import { Cause, Clock, Effect, Fiber, Queue, Stream } from "effect"
import { TestClock } from "effect/testing"

// Prints the element with the test-clock time. The time is exact, because the clock is fake.
const stamp = <A>(label: string) => (a: A) =>
  Effect.map(Clock.currentTimeMillis, (ms) => label + " " + ms + "ms " + a)

const program = Effect.gen(function* () {
  // throttle: at most 1 element for each second
  const throttled = yield* Effect.forkChild(
    Stream.range(1, 3).pipe(
      Stream.rechunk(1),                                      // throttle counts chunks, so make each element a chunk
      Stream.throttle({ cost: (chunk) => chunk.length, units: 1, duration: "1 second" }),
      Stream.mapEffect(stamp("throttle")),
      Stream.runCollect
    )
  )
  yield* TestClock.adjust("2 seconds")                        // the fake clock moves 2 seconds at once
  console.log(yield* Fiber.join(throttled))

  // debounce: emit only when the input is quiet for 100 ms
  const keys = yield* Queue.unbounded<string, Cause.Done>()
  const debounced = yield* Effect.forkChild(
    Stream.fromQueue(keys).pipe(Stream.debounce("100 millis"), Stream.runCollect)
  )
  yield* Queue.offerAll(keys, ["k", "ke", "key"])              // 3 keystrokes, no pause
  yield* TestClock.adjust("100 millis")                       // a pause
  yield* Queue.offerAll(keys, ["keyb", "keybo"])
  yield* TestClock.adjust("100 millis")
  yield* Queue.end(keys)
  console.log("debounce", yield* Fiber.join(debounced))

  // timeout: end the stream when no element arrives in 1 second
  const pings = yield* Queue.unbounded<string, Cause.Done>()
  const guarded = yield* Effect.forkChild(
    Stream.fromQueue(pings).pipe(Stream.timeout("1 second"), Stream.runCollect)
  )
  yield* Queue.offer(pings, "ping")
  yield* TestClock.adjust("500 millis")
  yield* Queue.offer(pings, "pong")
  yield* TestClock.adjust("1 second")                         // nothing arrives: the stream ends. The queue never ends.
  console.log("timeout", yield* Fiber.join(guarded))
}).pipe(Effect.provide(TestClock.layer()))

Effect.runPromise(program)

import { Effect, Schedule, Schema, Stream } from "effect"

class Corrupt extends Schema.TaggedError<Corrupt>()("Corrupt", { line: Schema.Number }) {}

// Two good lines, then a failure in the middle of the stream
const lines = Stream.make("ok 1", "ok 2").pipe(
  Stream.concat(Stream.fail(new Corrupt({ line: 3 }))),
  Stream.concat(Stream.make("never reached"))
)

// A source that fails on the first two attempts, then works
let attempts = 0
const flaky = Stream.suspend(() => {
  attempts++
  return attempts < 3 ? Stream.fail("connection refused") : Stream.make("connected on attempt " + attempts)
})

const program = Effect.gen(function* () {
  const recovered = yield* lines.pipe(
    Stream.catchTag("Corrupt", (e) => Stream.make("skipped line " + e.line)),   // E becomes never
    Stream.runCollect
  )
  console.log(recovered)

  const raw = yield* lines.pipe(
    Stream.runCollect,
    Effect.catch((e) => Effect.succeed(["run failed with " + e._tag]))       // or handle after running
  )
  console.log(raw)

  const retried = yield* flaky.pipe(Stream.retry(Schedule.recurs(5)), Stream.runCollect)
  console.log(retried)
})

Effect.runPromise(program)

import { Effect, Stream } from "effect"

const pipeline = Stream.range(1, 3).pipe(
  Stream.tap((n) => Effect.sync(() => console.log("processing", n)))
)

Effect.runPromise(pipeline)

import { Effect, Stream } from "effect"

const program = Stream.range(1, 7).pipe(
  Stream.sliding(3),
  Stream.runForEach((batch) => Effect.sync(() => console.log("write", JSON.stringify(batch))))
)

Effect.runPromise(program)

import { Effect, Stream } from "effect"

const prices: Record<string, number> = { apple: 10, pear: 20, fig: 30 }
const lookupPrice = (item: string) => Effect.succeed(prices[item] ?? 0)

const program = Stream.make("apple", "pear", "fig").pipe(
  Stream.map((item) => lookupPrice(item)),
  Stream.runFold(() => 0, (total, price) => total + price)
)

Effect.runPromise(program).then((total) => console.log(total))

import { Effect, Stream } from "effect"

const program = Stream.iterate(0, (n) => n + 1).pipe(
  Stream.take(4),
  Stream.filter((n) => n % 7 === 0),
  Stream.runCollect
)

Effect.runPromise(program).then((xs) => console.log(xs))

import { Effect, Stream } from "effect"

const words = Stream.make("a", "b", "a")

const program = Stream.runFold(words, {} as Record<string, number>, (counts, w) => {
  counts[w] = (counts[w] ?? 0) + 1
  return counts
})

Effect.runPromise(program).then((counts) => console.log(JSON.stringify(counts)))

import { Effect, Schema, Stream } from "effect"

class ParseError extends Schema.TaggedError<ParseError>()("ParseError", { input: Schema.String }) {}

const numbers = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail(new ParseError({ input: "x" })))
)

const program = numbers.pipe(
  Stream.catchTag("ParseFailure", (e) => Stream.make(-1)),
  Stream.runCollect
)

Effect.runPromise(program).then((xs) => console.log(xs))

import { Effect, Stream } from "effect"

const parseNumber = (line: string) =>
  Effect.try({
    try: () => {
      const n = Number(line)
      if (Number.isNaN(n)) throw new Error()
      return n
    },
    catch: () => "not a number: " + line
  })

const parseAll = (lines: Array<string>): Stream.Stream<number> =>
  Stream.fromIterable(lines).pipe(
    Stream.mapEffect(parseNumber)
  )

Effect.runPromise(Stream.runCollect(parseAll(["10", "20", "30"]))).then((xs) => console.log(xs))

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

const program = Effect.gen(function* () {
  const queue = yield* Queue.unbounded<string>()

  yield* Effect.forkChild(Effect.gen(function* () {
    yield* Queue.offerAll(queue, ["job-1", "job-2", "job-3"])
    yield* Queue.end(queue)
  }))

  const jobs = yield* Stream.runCollect(Stream.fromQueue(queue))
  console.log(jobs)
})

Effect.runPromise(program)

import { Effect, Stream } from "effect"

const openFile = Effect.acquireRelease(
  Effect.sync(() => {
    console.log("open")
    return ["l1", "l2", "l3"]
  }),
  () => Effect.sync(() => console.log("close"))
)

const lines = Stream.fromEffect(openFile).pipe(
  Stream.flatMap((all) => Stream.fromIterable(all))
)

const program = Effect.gen(function* () {
  console.log(yield* lines.pipe(Stream.take(2), Stream.runCollect))
  console.log("next task")
})

Effect.runPromise(Effect.scoped(program))

import { Effect, Stream } from "effect"

const program = Effect.gen(function* () {
  const shared = yield* Stream.broadcast(Stream.make(3, 1, 4), { capacity: 8 })

  const [sum, max] = yield* Effect.all([
    Stream.runFold(shared, () => 0, (a, b) => a + b),
    Stream.runFold(shared, () => 0, (a, b) => Math.max(a, b))
  ], { concurrency: "unbounded" })

  console.log("sum", sum, "max", max)
})

Effect.runPromise(program)