1/**
2 * @title Creating streams from common data sources
3 *
4 * Learn how to create streams from various data sources. Includes:
5 *
6 * - `Stream.fromIterable` for arrays and other iterables
7 * - `Stream.fromEffectSchedule` for polling effects
8 * - `Stream.paginate` for paginated APIs
9 * - `Stream.fromAsyncIterable` for async iterables
10 * - `Stream.fromEventListener` for DOM events
11 * - `Stream.callback` for any callback-based API
12 * - `NodeStream.fromReadable` for Node.js readable streams
13 */
14import { NodeStream } from "@effect/platform-node"
15import { Array, Effect, Queue, Schedule, Schema, Stream } from "effect"
16import * as Option from "effect/Option"
17import { Readable } from "node:stream"
18
19// `Stream.fromIterable` turns any iterable into a stream.
20export const numbers = Stream.fromIterable<number>([1, 2, 3, 4, 5])
21
22// `Stream.fromEffectSchedule` turns a single effect into a polling stream.
23// This is useful for metrics, health checks, and cache refresh loops.
24export const samples = Stream.fromEffectSchedule(
25 Effect.succeed(3),
26 Schedule.spaced("30 seconds")
27).pipe(
28 // Stream.take limits the number of elements emitted by the stream.
29 Stream.take(3)
30)
31
32// Use `Stream.paginate` when reading APIs that return one page at a time.
33// The function returns the current page of values and optionally the next
34// cursor.
35export const fetchJobsPage = Stream.paginate(
36 0, // start with page 0 (the cursor)
37 Effect.fn(function*(page) {
38 // Simulate network latency
39 yield* Effect.sleep("50 millis")
40
41 const results = Array.range(0, 100).map((i) => `Job ${i + 1 + page * 100}`)
42
43 // only return 10 pages of results
44 const nextPage = page <= 10
45 ? Option.some(page + 1)
46 : Option.none()
47
48 return [results, nextPage] as const
49 })
50)
51
52class LetterError extends Schema.TaggedError<LetterError>()("LetterError", {
53 cause: Schema.Defect()
54}) {}
55
56async function* asyncIterable() {
57 yield "a"
58 yield "b"
59 yield "c"
60}
61
62// Create a stream from an async iterable.
63// The second argument is a function that converts any errors thrown by the
64// async iterable into a typed error.
65export const letters = Stream.fromAsyncIterable(
66 asyncIterable(),
67 (cause) => new LetterError({ cause })
68)
69
70const button = document.getElementById("my-button")!
71
72// `Stream.fromEventListener` creates a stream from an event listener.
73export const events = Stream.fromEventListener<PointerEvent>(button, "click")
74
75// You can also use `Stream.callback` to create a stream from any callback-based
76// API.
77export const callbackStream = Stream.callback<PointerEvent>(Effect.fn(function*(queue) {
78 // You can use the `Queue` apis to emit values into the stream from the
79 // callback.
80 function onEvent(event: PointerEvent) {
81 Queue.offerUnsafe(queue, event)
82 }
83 // register the event listener and add a finalizer to unregister it when the
84 // stream is finished.
85 yield* Effect.acquireRelease(
86 Effect.sync(() => button.addEventListener("click", onEvent)),
87 () => Effect.sync(() => button.removeEventListener("click", onEvent))
88 )
89}))
90
91export class NodeStreamError extends Schema.TaggedError<NodeStreamError>()("NodeStreamError", {
92 cause: Schema.Defect()
93}) {}
94
95// Create a stream from a Node.js readable stream.
96//
97// It takes options to convert any errors emitted by the stream into a typed
98// error, and to evaluate the stream lazily.
99export const nodeStream = NodeStream.fromReadable({
100 evaluate: () => Readable.from(["Hello", " ", "world", "!"]),
101 onError: (cause) => new NodeStreamError({ cause }),
102 closeOnDone: true // true by default
103})