104 lines
3.5 KiB
TypeScript
104 lines
3.5 KiB
TypeScript
/**
|
|
* @title Creating streams from common data sources
|
|
*
|
|
* Learn how to create streams from various data sources. Includes:
|
|
*
|
|
* - `Stream.fromIterable` for arrays and other iterables
|
|
* - `Stream.fromEffectSchedule` for polling effects
|
|
* - `Stream.paginate` for paginated APIs
|
|
* - `Stream.fromAsyncIterable` for async iterables
|
|
* - `Stream.fromEventListener` for DOM events
|
|
* - `Stream.callback` for any callback-based API
|
|
* - `NodeStream.fromReadable` for Node.js readable streams
|
|
*/
|
|
import { NodeStream } from "@effect/platform-node"
|
|
import { Array, Effect, Queue, Schedule, Schema, Stream } from "effect"
|
|
import * as Option from "effect/Option"
|
|
import { Readable } from "node:stream"
|
|
|
|
// `Stream.fromIterable` turns any iterable into a stream.
|
|
export const numbers = Stream.fromIterable<number>([1, 2, 3, 4, 5])
|
|
|
|
// `Stream.fromEffectSchedule` turns a single effect into a polling stream.
|
|
// This is useful for metrics, health checks, and cache refresh loops.
|
|
export const samples = Stream.fromEffectSchedule(
|
|
Effect.succeed(3),
|
|
Schedule.spaced("30 seconds")
|
|
).pipe(
|
|
// Stream.take limits the number of elements emitted by the stream.
|
|
Stream.take(3)
|
|
)
|
|
|
|
// Use `Stream.paginate` when reading APIs that return one page at a time.
|
|
// The function returns the current page of values and optionally the next
|
|
// cursor.
|
|
export const fetchJobsPage = Stream.paginate(
|
|
0, // start with page 0 (the cursor)
|
|
Effect.fn(function*(page) {
|
|
// Simulate network latency
|
|
yield* Effect.sleep("50 millis")
|
|
|
|
const results = Array.range(0, 100).map((i) => `Job ${i + 1 + page * 100}`)
|
|
|
|
// only return 10 pages of results
|
|
const nextPage = page <= 10
|
|
? Option.some(page + 1)
|
|
: Option.none()
|
|
|
|
return [results, nextPage] as const
|
|
})
|
|
)
|
|
|
|
class LetterError extends Schema.TaggedErrorClass<LetterError>()("LetterError", {
|
|
cause: Schema.Defect()
|
|
}) {}
|
|
|
|
async function* asyncIterable() {
|
|
yield "a"
|
|
yield "b"
|
|
yield "c"
|
|
}
|
|
|
|
// Create a stream from an async iterable.
|
|
// The second argument is a function that converts any errors thrown by the
|
|
// async iterable into a typed error.
|
|
export const letters = Stream.fromAsyncIterable(
|
|
asyncIterable(),
|
|
(cause) => new LetterError({ cause })
|
|
)
|
|
|
|
const button = document.getElementById("my-button")!
|
|
|
|
// `Stream.fromEventListener` creates a stream from an event listener.
|
|
export const events = Stream.fromEventListener<PointerEvent>(button, "click")
|
|
|
|
// You can also use `Stream.callback` to create a stream from any callback-based
|
|
// API.
|
|
export const callbackStream = Stream.callback<PointerEvent>(Effect.fn(function*(queue) {
|
|
// You can use the `Queue` apis to emit values into the stream from the
|
|
// callback.
|
|
function onEvent(event: PointerEvent) {
|
|
Queue.offerUnsafe(queue, event)
|
|
}
|
|
// register the event listener and add a finalizer to unregister it when the
|
|
// stream is finished.
|
|
yield* Effect.acquireRelease(
|
|
Effect.sync(() => button.addEventListener("click", onEvent)),
|
|
() => Effect.sync(() => button.removeEventListener("click", onEvent))
|
|
)
|
|
}))
|
|
|
|
export class NodeStreamError extends Schema.TaggedErrorClass<NodeStreamError>()("NodeStreamError", {
|
|
cause: Schema.Defect()
|
|
}) {}
|
|
|
|
// Create a stream from a Node.js readable stream.
|
|
//
|
|
// It takes options to convert any errors emitted by the stream into a typed
|
|
// error, and to evaluate the stream lazily.
|
|
export const nodeStream = NodeStream.fromReadable({
|
|
evaluate: () => Readable.from(["Hello", " ", "world", "!"]),
|
|
onError: (cause) => new NodeStreamError({ cause }),
|
|
closeOnDone: true // true by default
|
|
})
|