Streams
A stream is a description of values over time: game events, net events, state bag changes, security events, ticks. Like an effect, nothing runs until you run it, and stopping it (take, an interrupted fiber, a closed scope) removes the event handlers it registered.
-- Post bad payloads to a webhook in batches: a post goes out once 20 have
-- piled up, or 10 s after the first of a batch, whichever comes first.
Fx.runFork(Fx.security.stream()
:filter(function(ev) return ev.kind ~= 'rate_limited' end)
:groupedWithin(20, 10000)
:runForEach(function(batch) return Webhook.post(batch) end))
-- A net event as a stream, with the same schema and rate limit as Fx.net.handle.
Fx.runFork(Fx.net.stream('radio:join', S.struct { channel = S.int:between(1, 999) })
:runForEach(function(m) return Radio.join(m.source, m.input.channel) end))
-- Client: react to a job change, but not to every flicker.
Fx.runFork(Job:changes({ bag = Fx.state.player() })
:debounce(250)
:runForEach(function(change) return Hud.setJob(change.value) end))
| Sources | Stream.make(...), fromTable, range, iterate, fromEffect, repeatEffect, interval(ms), fromSchedule, fromQueue, fromPubSub, async(register, { buffer }), event(name, shape?), merge(...); FiveM: Fx.net.stream, Key:changes, Fx.security.stream |
| Operators | map, mapEffect, filter, filterMap, tap, take, takeWhile, takeUntil, drop, dropWhile, scan, changes, grouped, groupedWithin, debounce, throttle, flatMap, concat, merge, catchAll, ensuring, interruptAfter |
| Running | runForEach, runForEachWhile, runCollect, runDrain, runFold, runCount, runHead, runLast, runFork |
Values pushed from callbacks wait in a buffer while the consumer is busy (1024 by default). When the buffer is full, the oldest value is dropped. Pass { buffer = n, strategy = 'dropping' } to keep the oldest instead.
Each value moves through the stream one at a time and in order. A slow runForEach slows the stream down; it never runs two values at once.