1/**
2 * @title Consuming and transforming streams
3 *
4 * How to transform and consume streams using operators like `map`, `flatMap`, `filter`, `mapEffect`, and various `run*` methods.
5 */
6import { Effect, Sink, Stream } from "effect"
7
8interface Order {
9 readonly id: string
10 readonly customerId: string
11 readonly status: "paid" | "refunded"
12 readonly subtotalCents: number
13 readonly shippingCents: number
14 readonly country: "US" | "CA" | "NZ"
15}
16
17interface NormalizedOrder extends Order {
18 readonly totalCents: number
19}
20
21interface EnrichedOrder extends NormalizedOrder {
22 readonly taxCents: number
23 readonly grandTotalCents: number
24 readonly priority: "normal" | "high"
25}
26
27// Start with structured order events from an in-memory source.
28export const orderEvents = Stream.succeed<Order>({
29 id: "ord_1001",
30 customerId: "cus_1",
31 status: "paid",
32 subtotalCents: 4_500,
33 shippingCents: 500,
34 country: "US"
35})
36
37// Use `Stream.map` for pure per-element transforms.
38export const normalizedOrders = orderEvents.pipe(
39 Stream.map((order): NormalizedOrder => ({
40 ...order,
41 totalCents: order.subtotalCents + order.shippingCents
42 }))
43)
44
45// `Stream.filter` lets you exclude elements that don't match a predicate.
46export const paidOrders = normalizedOrders.pipe(
47 Stream.filter((order) => order.status === "paid")
48)
49
50// Use `Stream.flatMap` to transform each element into a stream, and flatten the
51// results.
52export const allOrders = Stream.make("US", "CA", "NZ").pipe(
53 Stream.flatMap(
54 (country) =>
55 Stream.range(1, 50).pipe(
56 Stream.map((i): Order => ({
57 id: `ord_${country}_${i}`,
58 customerId: `cus_${i}`,
59 status: i % 10 === 0 ? "refunded" : "paid",
60 subtotalCents: Math.round(Math.random() * 100_000),
61 shippingCents: Math.round(Math.random() * 10_000),
62 country
63 }))
64 ),
65 // Optionally control the concurrency of the flatMap with the second argument.
66 { concurrency: 2 }
67 )
68)
69
70const enrichOrder = Effect.fn(function*(order: NormalizedOrder): Effect.fn.Return<EnrichedOrder> {
71 // Simulate effectful enrichment (for example, tax/risk lookup).
72 yield* Effect.sleep("5 millis")
73
74 const taxRate = order.country === "US" ? 0.08 : 0.13
75 const taxCents = Math.round(order.totalCents * taxRate)
76
77 return {
78 ...order,
79 taxCents,
80 grandTotalCents: order.totalCents + taxCents,
81 priority: order.totalCents >= 20_000 ? "high" : "normal"
82 }
83})
84
85// `Stream.mapEffect` performs effectful per-element transforms with concurrency control.
86export const enrichedPaidOrders = paidOrders.pipe(
87 Stream.mapEffect(enrichOrder, { concurrency: 4 })
88)
89
90// `runCollect` gathers all stream outputs into an immutable array.
91export const collectedOrders = Stream.runCollect(enrichedPaidOrders)
92
93// `runDrain` runs the stream for its effects, ignoring all outputs.
94export const drained = Stream.runDrain(enrichedPaidOrders)
95
96// `runForEach` executes an effectful consumer for every element.
97export const logOrders = enrichedPaidOrders.pipe(
98 Stream.runForEach((order) => Effect.logInfo(`Order ${order.id} total=$${(order.grandTotalCents / 100).toFixed(2)}`))
99)
100
101// `runFold` reduces the stream to one accumulated value.
102export const totalRevenueCents = enrichedPaidOrders.pipe(
103 Stream.runFold(() => 0, (acc: number, order) => acc + order.grandTotalCents)
104)
105
106// `run` lets you consume a stream through any Sink.
107export const totalRevenueViaSink = enrichedPaidOrders.pipe(
108 Stream.map((order) => order.grandTotalCents),
109 Stream.run(Sink.sum)
110)
111
112// `runHead` and `runLast` capture edge elements as Option values.
113export const firstLargeOrder = enrichedPaidOrders.pipe(
114 Stream.filter((order) => order.priority === "high"),
115 Stream.runHead
116)
117
118export const lastLargeOrder = enrichedPaidOrders.pipe(
119 Stream.filter((order) => order.priority === "high"),
120 Stream.runLast
121)
122
123// Windowing-style operators help shape what downstream consumers see.
124export const firstTwoOrders = enrichedPaidOrders.pipe(
125 Stream.take(2),
126 Stream.runCollect
127)
128
129export const afterWarmupOrder = enrichedPaidOrders.pipe(
130 Stream.drop(1),
131 Stream.runCollect
132)
133
134export const untilLargeOrder = enrichedPaidOrders.pipe(
135 Stream.takeWhile((order) => order.priority === "normal"),
136 Stream.runCollect
137)