Sink #

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

// A consumer, defined once, with no stream in sight
const printer = Sink.forEach((line: string) => Effect.sync(() => console.log("> " + line)))

const today = Stream.make("deploy", "rollback")
const yesterday = Stream.make("hotfix")

const program = Effect.gen(function* () {
  yield* Stream.run(today, printer)          // same sink...
  yield* Stream.run(yesterday, printer)      // ...different stream

  const amounts = Stream.make(19, 5, 42)
  console.log("sum", yield* Stream.run(amounts, Sink.sum))
  console.log("count", yield* amounts.pipe(Stream.run(Sink.count)))   // pipe style works too
  console.log("collect", yield* amounts.pipe(Stream.run(Sink.collect())))
  console.log("first two", yield* amounts.pipe(Stream.run(Sink.take(2))))

  const first = yield* amounts.pipe(Stream.run(Sink.head()))
  console.log("head", Option.getOrElse(first, () => 0))

  const none = yield* Stream.empty.pipe(Stream.run(Sink.last()))
  console.log("last of empty is", Option.isNone(none) ? "none" : "some")
})

Effect.runPromise(program)

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

interface State {
  readonly min: number
  readonly max: number
  readonly sum: number
  readonly n: number
}

// One pass over the input, no arrays allocated
const stats = Sink.reduce(
  (): State => ({ min: Infinity, max: -Infinity, sum: 0, n: 0 }),   // fresh state per run
  (s: State, x: number): State => ({
    min: Math.min(s.min, x),
    max: Math.max(s.max, x),
    sum: s.sum + x,
    n: s.n + 1
  })
).pipe(
  Sink.map((s) => ({ min: s.min, max: s.max, mean: s.sum / s.n }))   // reshape the result
)

const program = Effect.gen(function* () {
  console.log(yield* Stream.make(4, 8, 15, 16, 23, 42).pipe(Stream.run(stats)))
  console.log(yield* Stream.range(1, 1000).pipe(Stream.run(stats)))

  // reduceWhile stops as soon as the predicate on the state is false
  const untilTen = Sink.reduceWhile(() => 0, (total) => total < 10, (total, x: number) => total + x)
  console.log("first total >= 10:", yield* Stream.range(1, 1_000_000).pipe(Stream.run(untilTen)))
})

Effect.runPromise(program)

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

interface Order {
  readonly id: string
  readonly total: number
  readonly items: number
}

const orders = Stream.make(
  { id: "o1", total: 30, items: 2 },
  { id: "o2", total: 45, items: 1 },
  { id: "o3", total: 25, items: 4 }
)

// Sink.sum wants numbers. mapInput teaches it how to read an Order.
const revenue: Sink.Sink<number, Order> = Sink.sum.pipe(Sink.mapInput((o: Order) => o.total))
const itemsSold: Sink.Sink<number, Order> = Sink.sum.pipe(Sink.mapInput((o: Order) => o.items))

// map changes the result: a number becomes a formatted string
const revenueReport = revenue.pipe(Sink.map((n) => "revenue: $" + n))

// as replaces the result with a constant, keeping the consumption
const ack = Sink.drain.pipe(Sink.as("all orders read"))

const program = Effect.gen(function* () {
  console.log(yield* orders.pipe(Stream.run(revenue)))
  console.log(yield* orders.pipe(Stream.run(itemsSold)))
  console.log(yield* orders.pipe(Stream.run(revenueReport)))
  console.log(yield* orders.pipe(Stream.run(ack)))
})

Effect.runPromise(program)

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

const csv = Stream.make("name,qty", "apple,3", "pear,5", "fig,2")

// Sink 1 reads the header. Sink 2 is built from it and reads the rest.
const table = Sink.take<string>(1).pipe(
  Sink.flatMap((headerLines) => {
    const columns = headerLines[0]!.split(",")
    return Sink.collect<string>().pipe(
      Sink.map((rows) => rows.map((row) => {
        const cells = row.split(",")
        return Object.fromEntries(columns.map((c, i) => [c, cells[i]]))
      }))
    )
  })
)

// A length-prefixed message: first element says how many to read
const prefixed = Sink.head<number>().pipe(
  Sink.flatMap((len) => Sink.take<number>(len._tag === "Some" ? len.value : 0))
)

const program = Effect.gen(function* () {
  const rows = yield* csv.pipe(Stream.run(table))
  console.log(JSON.stringify(rows))

  const payload = yield* Stream.make(3, 10, 20, 30, 40, 50).pipe(Stream.run(prefixed))
  console.log("payload", payload)                          // 40 and 50 are left over, not read
})

Effect.runPromise(program)

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

const letters = Stream.make("a", "b", "c")

const program = letters.pipe(Stream.run(Sink.collectAll()))

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

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

const words = Stream.make("ab", "cde", "fghi")

const totalLength = Sink.sum.pipe(Sink.map((s: string) => s.length))

Effect.runPromise(words.pipe(Stream.run(totalLength))).then((n) => console.log(n))

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

const validated: Sink.Sink<void, number> = Sink.forEach((amount: number) =>
  amount < 0 ? Effect.fail("negative amount: " + amount) : Effect.void
)

const program = Stream.make(10, 20, 30).pipe(Stream.run(validated))

Effect.runPromise(program).then(() => console.log("all amounts valid"))

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

const deposits = Stream.make(40, 35, 30, 50, 10)

const untilHundred = Sink.reduce(() => 0, (total, n: number) => total + n)

Effect.runPromise(deposits.pipe(Stream.run(untilHundred))).then((n) => console.log("stopped at", n))