@uniflowed/effect
type
QueueStrategy
export type QueueStrategy = "bounded" | "dropping" | "sliding";
What a full queue does with one more value.
bounded is the only one of the three that is back pressure: the producer waits, and a producer that waits is a producer that cannot outrun its consumer. The other two keep the producer running and lose a value instead — dropping loses the value being offered, sliding loses the oldest one already waiting — which is what a metrics feed or a live cursor position wants, where the newest reading is the only interesting one.
opaque-type
Effect
export opaque type Effect<out A, out E = empty, out R = empty>: $Iterable<
Effect<mixed, E, R>,
A,
mixed,
> = EffectCarrier<A, E, R>;
Work that produces an A, may fail with an E, and needs an R.
E defaults to empty, Flow's bottom type, so Effect<number> reads as cannot fail. R defaults to empty for an effect that needs no services.
The $Iterable bound is what makes yield* someEffect legal inside effect(function* () { … }) *and* typed: the delegate's Return type is A, so the checker gives the yielded expression the effect's success type rather than the mixed a bare yield produces.
opaque-type
Fiber
export opaque type Fiber<out A, out E = empty> = FiberCarrier<A, E>;
A running effect, addressable so it can be awaited or cancelled.
opaque-type
Tag
export opaque type Tag<out Service>: Effect<Service, empty, Service> = TagCarrier<Service>;
Identifies one service inside a context, and is the effect that reads it.
The supertype bound is the whole point: a tag can be handed to any combinator that takes an effect, and yield* Clock inside a generator produces the service, so there is no second accessor function to learn.
opaque-type
Layer
export opaque type Layer<out Out, out E = empty, out In = empty> = LayerCarrier<Out, E, In>;
A recipe for building services, itself possibly failing.
opaque-type
Runtime
export opaque type Runtime<R> = RuntimeCarrier<R>;
Services built once, for an application rather than for one effect.
provide builds a layer, runs one effect against it and closes the scope when that effect ends, which is right for what provide is and wrong for what a server is: a request handler is an Effect per request against a connection pool opened once at boot. Building the layer per request opens two pools a second; assembling a provideService chain by hand at the entry point is the thing Layer exists to replace.
Invariant in R, unlike Effect and Layer, and that is what makes the requirement check mean something here. Effect is covariant in R, so with a covariant Runtime the checker could satisfy runtimeRunPromise by widening *both* sides to a union containing a service the runtime does not have. Pinned to exactly what the layer produced, the effect has to be a subtype of it, which is the ordinary subtyping the issue this came from expected and not another instance of the requirement-subtraction caveat.
opaque-type
Ref
export opaque type Ref<A> = RefCarrier<A>;
A place two fibers can both read and write.
Invariant in A, unlike Effect and Fiber: a Ref is written as well as read, so a Ref<string> is not a Ref<mixed> — writing a number through the second would break the first.
The operations are flat monomorphic functions — refGet, refUpdate — and not methods or a Ref namespace object. That is this package's convention rather than a preference expressed once more: there is no pipe with inference to lose, nothing to dispatch at run time, and a bundler can drop the eleven of these a program does not call. It is stated here because Ref is the first type where a namespace would have looked natural.
opaque-type
Deferred
export opaque type Deferred<A, E = empty> = DeferredCarrier<A, E>;
A value one fiber will produce and others are waiting for.
Fiber covers "wait for the thing this fiber returns". This covers the other one: waiting for a value nobody has promised yet — a one-shot handshake, a lazy singleton, "the first caller does the work and the rest wait".
opaque-type
Semaphore
export opaque type Semaphore = SemaphoreCarrier;
Permission to run, in a fixed number of copies.
all's concurrency bounds one call. This bounds a budget shared across call sites, which is what a rate-limited API needs: two independent forEaches against the same host can hold one of these between them.
opaque-type
Queue
export opaque type Queue<A> = QueueCarrier<A>;
A bounded place to hand work from one fiber to another.
The type the rest of this package was waiting on. all collects into an array, so a producer that outruns its consumer grows one until the process dies and there is no seam to put a limit in; a Queue is that seam. A bounded queue makes the producer wait, which is the only one of the three strategies that is back pressure rather than a way of losing values politely.
Invariant in A for the reason Ref is: a queue is written as well as read, so a Queue<string> is not a Queue<mixed>.
Every operation but queueShutdown and queueIsShutdown reports an *interruption* on a queue that has been shut down, waiting or not. That is one rule rather than two, and it is the one Cause already has a word for: a shutdown is a decision somebody took, which is exactly what interrupt means and exactly what fail does not.
opaque-type
PubSub
export opaque type PubSub<A> = PubSubCarrier<A>;
One value, delivered to everybody who is listening when it is published.
Queue is one-to-one — a value taken by one consumer is gone. This is the many-to-many case: every subscriber gets its own buffer, so a slow one falls behind rather than stealing from a fast one, and with bounded a slow one makes the publisher wait.
A subscription is an ordinary Queue, which is what makes this a small type rather than a second family of readers: queueTake, queueTakeAll and queueSize are already the operations a subscriber needs.
opaque-type
Scope
export opaque type Scope = { readonly __kind: "Scope" };
The lifetime a resource is released at.
Never constructed: it exists to appear in an effect's R so that a value built with acquireRelease cannot be run until something has said where its release belongs.
type
Cause
export type Cause<out E> =
| { readonly kind: "empty" }
| { readonly kind: "fail", readonly error: E }
| { readonly kind: "die", readonly defect: string }
| { readonly kind: "interrupt" }
| { readonly kind: "sequential", readonly causes: $ReadOnlyArray<Cause<E>> }
| { readonly kind: "parallel", readonly causes: $ReadOnlyArray<Cause<E>> };
Why an effect did not produce a value.
Three leaves, and the distinction between them is the point of the package: fail is the typed error the signature promised and the only one recovery and retry act on, die is a bug in the program, and interrupt is a decision somebody already took. sequential and parallel hold the causes of several effects that failed together — race produces a parallel when every entrant failed.
type
Exit
export type Exit<out A, out E> =
| { readonly kind: "success", readonly value: A }
| { readonly kind: "failure", readonly cause: Cause<E> };
How an effect ended: a value, or the reason there is none.
type
TimeoutError
export type TimeoutError = { readonly kind: "timeout", readonly millis: number };
What timeout adds to an effect's error channel.
type
Concurrency
export type Concurrency = number | "unbounded";
How many effects all and forEach may have in flight.
Effect's "inherit" is deliberately absent: there is no enclosing limit recorded to inherit, and an option that silently means something else is worse than one that is not offered.
type
EffectGenerator
export type EffectGenerator<A, E, R> = Generator<Effect<mixed, E, R>, A, mixed>;
The generator an effect body is.
Yield is Effect<mixed, E, R> and not Effect<A, E, R> because the steps of a pipeline produce different types; E and R accumulate as the union of everything yielded, which is where a pipeline's failure type comes from. Next is mixed for the same reason, so a bare yield produces mixed and yield* produces the effect's own A.
function
succeed
export function succeed<A>(value: A): Effect<A> { ... }
An effect that has already produced value.
function
fail
export function fail<E>(error: E): Effect<empty, E> { ... }
An effect that has already failed with the typed error error.
function
die
export function die(defectValue: mixed): Effect<empty> { ... }
An effect that has already failed with a defect.
Not the same thing as fail: a defect is outside the error channel, so catchAll will not see it, retry will not repeat it, and either will not reify it. That is the whole distinction the package exists to keep.
function
never
export function never(): Effect<empty> { ... }
An effect that never produces anything — until it is interrupted.
Interruptible on purpose. A never that ignored the flag would make interrupt(runFork(never())) hang for the life of the process, which is the one thing a caller reaching for never is most likely to do next.
function
sync
export function sync<A>(body: () => A): Effect<A> { ... }
Run body when the effect runs. A throw becomes a defect, not a failure.
function
trySync
export function trySync<A, E>(options: {
readonly try: () => A,
readonly catch: (error: mixed) => E,
}): Effect<A, E> { ... }
Run a synchronous operation, translating a thrown value into a typed failure.
Like tryPromise, but retains a synchronous kernel for runSync and SQLite transactions. The mapper receives the original value, including custom error fields; sync deliberately records a defect instead. Both the operation and mapper run lazily, once per execution. A mapper that throws is itself a defect.
function
suspend
export function suspend<A, E, R>(body: () => Effect<A, E, R>): Effect<A, E, R> { ... }
Build the effect when it runs, not when it is described.
function
promise
export function promise<A>(body: () => Promise<A>): Effect<A> { ... }
Adopt a promise. A rejection is a defect: nothing said what it could be.
function
tryPromise
export function tryPromise<A, E>(options: {
readonly try: () => Promise<A>,
readonly catch: (error: mixed) => E,
}): Effect<A, E> { ... }
Adopt a promise, naming what a rejection means as a typed failure.
function
call
export function call<A>(body: () => A | Promise<A>): Effect<A> { ... }
Run body, which may or may not be asynchronous.
function
effect
export function effect<A, E, R>(body: () => EffectGenerator<A, E, R>): Effect<A, E, R> { ... }
Write a pipeline as a generator.
yield* effect is the typed form: the yielded expression has the effect's success type, and the body's failure type is the union of every step's. A bare yield effect works too and produces mixed, which is occasionally what a caller wants and never what it should reach for first.
Both a synchronous and an asynchronous driver, because a pipeline of synchronous steps has a synchronous answer and runSync should be able to ask for it. The synchronous driver refuses at the first step that has no synchronous kernel, with the same defect any other such effect gives.
function
map
export function map<A, B, E, R>(
self: Effect<A, E, R>,
transform: (value: A) => B,
): Effect<B, E, R> { ... }
Change a success, leaving the failure channel alone.
function
mapError
export function mapError<A, E, F, R>(
self: Effect<A, E, R>,
transform: (error: E) => F,
): Effect<A, F, R> { ... }
Change a typed failure, leaving a defect and an interruption alone.
function
flatMap
export function flatMap<A, B, E1, E2, R1, R2>(
self: Effect<A, E1, R1>,
next: (value: A) => Effect<B, E2, R2>,
): Effect<B, E1 | E2, R1 | R2> { ... }
Run next on the success of self.
The failure channel is the union of both, which is what makes a pipeline's error type the sum of what its steps can fail with rather than whatever the last step happened to declare.
function
andThen
export function andThen<A, B, E1, E2, R1, R2>(
self: Effect<A, E1, R1>,
next: Effect<B, E2, R2>,
): Effect<B, E1 | E2, R1 | R2> { ... }
Run next after self, discarding self's value.
function
zip
export function zip<A, B, E1, E2, R1, R2>(
self: Effect<A, E1, R1>,
other: Effect<B, E2, R2>,
): Effect<[A, B], E1 | E2, R1 | R2> { ... }
Both values as a pair, self first and sequentially.
function
all
export function all<A, E, R>(
effects: $ReadOnlyArray<Effect<A, E, R>>,
options?: { readonly concurrency?: Concurrency },
): Effect<$ReadOnlyArray<A>, E, R> { ... }
Every effect, with at most concurrency of them in flight.
Fails fast, and states exactly what that costs: the first failure — first in *completion* order, not in index order — is the result. Its siblings are interrupted and then awaited, so all does not return until nothing it started is still running. That is the difference between a combinator and a leak: a sibling left in flight would go on writing to the world after the caller had already handled the failure.
Values come back in the order of the input, whatever order they finished in.
function
forEach
export function forEach<A, B, E, R>(
items: $ReadOnlyArray<A>,
body: (item: A, index: number) => Effect<B, E, R>,
options?: { readonly concurrency?: Concurrency },
): Effect<$ReadOnlyArray<B>, E, R> { ... }
all over the effects body builds from items.
function
race
export function race<A, E, R>(effects: $ReadOnlyArray<Effect<A, E, R>>): Effect<A, E, R> { ... }
The first entrant to *succeed*.
A failure removes that entrant from the race rather than ending it, because "whichever settles first" is not what a caller wants when the fastest answer is an error. If every entrant fails, the result is a parallel cause holding all of them, so nothing is hidden and catchAll still finds the first typed error among them. An empty list behaves as never.
The losers are interrupted and *not* awaited. Waiting for the slower entrant to reach its next checkpoint would hand back the latency the race exists to avoid; the cost is that a loser may run a little past the point race returned, so a loser with side effects has to be written to survive that.
function
catchAll
export function catchAll<A, B, E, F, R1, R2>(
self: Effect<A, E, R1>,
recover: (error: E) => Effect<B, F, R2>,
): Effect<A | B, F, R1 | R2> { ... }
Recover from a typed failure.
A defect and an interruption pass straight through: neither is in the error channel recover was written against, and catching them here is how a bug ends up reported as a handled condition.
function
catchTag
export function catchTag<
A,
B,
E extends { readonly kind?: string, readonly tag?: string, ... },
F,
R1,
R2,
>(
self: Effect<A, E, R1>,
tagName: string,
recover: (error: E) => Effect<B, F, R2>,
): Effect<A | B, E | F, R1 | R2> { ... }
Recover from one kind of tagged failure, re-failing the rest.
The bound is what makes this typed rather than a probe: an error has to say which kind it is before it can be caught by kind. Both kind — the discriminant Flow's match and the rest of uf use — and tag, which is what an error ported from Effect-TS will be carrying, are read.
recover still sees the whole E. Flow cannot narrow a type variable by a string compared at run time, so the narrowing Effect-TS gets from a literal _tag is not available; see Readiness.
function
orElse
export function orElse<A, B, E, F, R1, R2>(
self: Effect<A, E, R1>,
fallback: () => Effect<B, F, R2>,
): Effect<A | B, F, R1 | R2> { ... }
Fall back to another effect on any typed failure.
function
either
export function either<A, E, R>(
self: Effect<A, E, R>,
): Effect<
{ readonly ok: true, readonly value: A } | { readonly ok: false, readonly error: E },
empty,
R,
> { ... }
Reify a typed failure as a value, so the effect itself cannot fail.
Only the typed failure. A defect and an interruption stay failures, which is why the error channel of the result is empty and not a lie: what comes back as { ok: false } is exactly what E said could happen.
type
RetryOptions
export type RetryOptions<in E> = {
readonly while?: (error: E) => boolean,
readonly until?: (error: E) => boolean,
};
Which typed failures another attempt is worth making for.
A separate parameter rather than a { schedule, while, until } union in retry's second position, which is what Effect takes. Flow's objects are exact, so a union of a Schedule and an options object cannot be refined by reading a property one of them does not have, and the trick that makes it possible would be a worse thing to explain than a third parameter. Effect's times is not here either: it is intersect with recurs, which the schedule union already says, and two ways to say one thing is the cost of copying an API rather than reading it.
A Schedule<E> can now read the error itself, so whileInput says the same thing inside a policy. That is not a duplicate: this is the condition the *call site* knows, and a policy is the one a config file could name. Both have to allow an attempt for one to happen.
The predicates see the first typed failure in the cause, which is the same error catchAll would hand a recovery function and the same one the schedule is stepped with.
type
RepeatOptions
export type RepeatOptions<in A> = {
readonly while?: (value: A) => boolean,
readonly until?: (value: A) => boolean,
};
The same, over the value a repeat produced.
function
retry
export function retry<A, E, R>(
self: Effect<A, E, R>,
schedule: Schedule<E>,
options?: RetryOptions<E>,
): Effect<A, E, R> { ... }
Try again on a typed failure, on the schedule's timetable.
The first run is not a retry, so { kind: "recurs", times: 2 } runs the effect three times. Only a typed failure is worth another attempt: a defect is a bug, so running it again runs the bug again, and an interruption is a decision already taken.
The schedule's input is the error, so a policy can decide on it — { kind: "whileInput", schedule: backoff, predicate: (e) => e.kind !== "forbidden" } is a policy that gives up on a rejection. options says the same thing at the call site; both have to allow an attempt.
The clock is read once per decision and passed to the schedule, which is what lets fixed subtract the time the attempt itself took, and what keeps scheduleStep a pure function of its arguments. The random factor a jittered schedule needs is drawn in the same place and for the same reason.
function
repeat
export function repeat<A, E, R>(
self: Effect<A, E, R>,
schedule: Schedule<A>,
options?: RepeatOptions<A>,
): Effect<number, E, R> { ... }
Run again on *success*, on the schedule's timetable: a poll, a heartbeat, a cache refresh.
The other half of what a schedule is for. The first run is not a repetition, so { kind: "recurs", times: 2 } runs the effect three times, and { kind: "spaced", millis: 1000 } runs it until something stops it.
The value is the schedule's *input*, so "poll until the job reports finished" is a schedule rather than a loop: { kind: "untilInput", schedule: everySecond, predicate: (job) => job.done }. options says the same thing at the call site, the way retry's does.
The result is the schedule's output — the number of repetitions, the elapsed time, whichever number the schedule's last decision reached — which is what Effect's repeat gives back and what this could not give back while a schedule was arithmetic over an attempt count. The effect's own last value is one tap or one Ref away and was never the interesting half: a poll that ran eleven times wants to say eleven.
A failure ends the repetition and is the result. That is not a policy choice: an effect that failed produced no value to repeat *from*, and swallowing the failure to keep polling would hide the outage the poll exists to notice. retry is what wraps an unreliable step, and the two compose — repeat(retry(poll, backoff), everySecond) is a poll that tolerates a blip and stops on a real failure.
Interruption is checked before each wait and after it, so a fiber polling once a minute stops when it is cancelled rather than at the top of the next minute, and reports an interruption rather than the last number it happened to have.
function
timeout
export function timeout<A, E, R>(
self: Effect<A, E, R>,
millis: number,
): Effect<A, E | TimeoutError, R> { ... }
Give up on self after millis.
The effect runs in a child fiber, so the timer expiring cancels it rather than leaving it running with nobody waiting, and timeout does not return until it has stopped. An interruption of the *calling* fiber is reported as an interruption and not as a timeout: reporting it as a timeout would make it a typed failure, and retry would then run the effect again after somebody asked for it to stop.
function
acquireRelease
export function acquireRelease<A, E, R>(
acquire: Effect<A, E, R>,
release: (resource: A) => Effect<void>,
): Effect<A, E, R | Scope> { ... }
Acquire something, and register how to give it back.
The Scope in the requirement channel is what stops this being run without somebody saying where the release belongs; scoped is what discharges it. Reaching this without a scope is a defect rather than a silent skip, because a release that never runs is the failure this combinator exists to prevent.
There is no interruption window between acquiring and registering: the flag is only read at the checkpoints this file writes, and there is none between the await below and the push after it.
function
scoped
export function scoped<A, E, R>(self: Effect<A, E, R | Scope>): Effect<A, E, R> { ... }
Give the effect a scope, and close it however the effect ends.
Finalizers run in reverse order of acquisition, and in a detached fiber: a scope closing because its fiber was interrupted still has to release what it took, and a finalizer running under the interrupted fiber would stop at its own first checkpoint.
A finalizer that fails turns into a defect and only replaces a success — a body that already failed keeps its own failure, which is the more useful half of the news.
function
ensuring
export function ensuring<A, E, R>(
self: Effect<A, E, R>,
finalizer: () => Effect<mixed, mixed, empty>,
): Effect<A, E, R> { ... }
Run finalizer however self ends: success, failure, defect, interruption.
acquireRelease needs a Scope; this does not, which makes it the right tool for the ordinary "close this when the block is done" case that would otherwise force a scope onto the caller's type to run one cleanup.
function
tag
export function tag<Service>(identifier: string): Tag<Service> { ... }
Name a service, in a value that is also the effect reading it.
function
provideService
export function provideService<A, E, R, Service>(
self: Effect<A, E, R | Service>,
serviceTag: Tag<Service>,
service: Service,
): Effect<A, E, R> { ... }
Satisfy one requirement with a value already in hand.
function
provide
export function provide<A, E, R, Out, LayerError, In>(
self: Effect<A, E, R | Out>,
layer: Layer<Out, LayerError, In>,
): Effect<A, E | LayerError, R | In> { ... }
Satisfy requirements by building them.
The layer's own failure joins the effect's error channel, because a service that could not be built is a way the whole thing can fail.
One build pass, with one memo, so a layer reached twice in the graph is built once. The pass also gets a scope of its own — separate from the body's — which is what a layerScoped resource is released at: the layer that opened a connection pool has somewhere to close it, and it closes after the body that was using it has finished, however it finished.
That scope is deliberately not the body's. Handing the body a scope here would silently discharge the Scope an acquireRelease inside it requires, and the requirement channel would go on saying otherwise; scoped is still the only thing that answers for an effect's own resources.
function
layerSucceed
export function layerSucceed<Service>(serviceTag: Tag<Service>, service: Service): Layer<Service> { ... }
A layer holding one service that is already built.
function
layerEffect
export function layerEffect<Service, E, R>(
serviceTag: Tag<Service>,
build: Effect<Service, E, R>,
): Layer<Service, E, R> { ... }
A layer that builds its service with an effect, which may itself fail.
function
layerScoped
export function layerScoped<Service, E, R>(
serviceTag: Tag<Service>,
build: Effect<Service, E, R | Scope>,
): Layer<Service, E, R> { ... }
A layer that acquires something, released when the provide using it ends.
The difference from layerEffect is entirely in the type, and that is the point rather than an admission: provide gives every build a scope, so a layerEffect over an acquireRelease would already release — but its Scope requirement would sit in the layer's In for ever, and a Layer has no scoped of its own to discharge it with. This is that discharge, in the same shape and with the same caveat scoped carries: the service being removed is stated and Flow solves for the rest.
The resource outlives the build and dies with the provide, which is the whole reason a pool belongs in a layer rather than in the body.
function
layerMerge
export function layerMerge<Out1, Out2, E1, E2, In1, In2>(
left: Layer<Out1, E1, In1>,
right: Layer<Out2, E2, In2>,
): Layer<Out1 | Out2, E1 | E2, In1 | In2> { ... }
Both layers' services, left built first so the right may fail after it.
function
layerProvide
export function layerProvide<Out, E1, In1, Out2, E2, In2>(
inner: Layer<Out, E1, In1 | Out2>,
outer: Layer<Out2, E2, In2>,
): Layer<Out, E1 | E2, In1 | In2> { ... }
Build inner with outer's services in scope, discharging what it needed.
This is what makes a layer's In mean something. Before it, In was carried through every signature and could never be satisfied: a Layer<Database, ConfigError, Config> could only be used by an effect that still required Config, so the requirement was a label rather than a debt anything could pay.
The same requirement-subtraction caveat as provide, in the same words: Flow has no type-level set difference, so the discharged service is stated — inner is typed as needing In1 | Out2 — and the checker solves for In1. That works when Out2 is a distinct member of the union and silently leaves it in In1 when it is not. See Readiness.
outer is built first and through the pass's memo, so a Config that two layers both provide into is built once.
function
layerProvideMerge
export function layerProvideMerge<Out, E1, In1, Out2, E2, In2>(
inner: Layer<Out, E1, In1 | Out2>,
outer: Layer<Out2, E2, In2>,
): Layer<Out | Out2, E1 | E2, In1 | In2> { ... }
layerProvide, keeping the outer layer's services in the result.
For the ordinary case where Config is wanted by the application as well as by the Database it was built for. Free, because the merge is the one line that differs.
function
managedRuntime
export function managedRuntime<Out, E, In>(layer: Layer<Out, E, In>): Effect<Runtime<Out>, E, In> { ... }
Build a layer once, and hand back something many effects can be run against.
Effect-TS calls this ManagedRuntime.make. The build is one pass with one memo, exactly as provide's is; the difference is entirely in who closes the scope and when. provide closes it when its one effect ends, and this does not close it at all — runtimeDispose does, whenever the application is over.
That makes this the one effect in the package that deliberately leaves something open when it returns, which is why dispose is not optional and why it is a named function rather than a finalizer somebody might not have registered. A runtime built and never disposed holds whatever its layers acquired for the life of the process, which for a process-lifetime pool is the point and for anything shorter is a leak.
The runtime keeps the services in scope where it was built as well as the ones the layer produced, so a runtime built inside a provideService sees both. A build that fails releases whatever the build had already acquired and reports the failure, as provide's does.
function
runtimeRunPromise
export function runtimeRunPromise<A, E, R>(self: Runtime<R>, body: Effect<A, E, R>): Promise<A> { ... }
Run an effect against a built runtime, raising whatever it failed with.
runPromise with the runtime's services in scope. Each run gets a root fiber of its own, so two effects running against one runtime do not own each other: one of them ending stops what *it* forked and nothing of the other's.
The runtime's own scope is not the effect's scope, for the reason provide does not hand the body one either — doing so would silently discharge the Scope an acquireRelease in the body requires while the requirement channel went on saying otherwise.
function
runtimeRunPromiseExit
export function runtimeRunPromiseExit<A, E, R>(
self: Runtime<R>,
body: Effect<A, E, R>,
): Promise<Exit<A, E>> { ... }
The same, returning the outcome rather than raising.
function
runtimeRunSync
export function runtimeRunSync<A, E, R>(self: Runtime<R>, body: Effect<A, E, R>): A { ... }
Run a synchronous effect against a built runtime, raising its failure.
function
runtimeRunSyncExit
export function runtimeRunSyncExit<A, E, R>(self: Runtime<R>, body: Effect<A, E, R>): Exit<A, E> { ... }
The same, returning the outcome rather than raising.
function
runtimeRunFork
export function runtimeRunFork<A, E, R>(self: Runtime<R>, body: Effect<A, E, R>): Fiber<A, E> { ... }
Start an effect against a built runtime and keep a handle on it.
function
runtimeDispose
export function runtimeDispose<R>(self: Runtime<R>): Effect<void> { ... }
Close what the runtime's layers acquired.
Idempotent, so a shutdown path that runs twice is not a second release of a connection pool. Every run against a disposed runtime is a defect rather than a typed failure: the services are gone, so an effect that asks for one is reaching for a resource somebody has already closed, and that is a bug in the program's shutdown order rather than a condition it should be recovering from — the same line readService draws for a service that was never provided.
function
fork
export function fork<A, E, R>(self: Effect<A, E, R>): Effect<Fiber<A, E>, empty, R> { ... }
Start self beside the current fiber and hand back a handle to it.
The child gets its own interruption state, so cancelling it does not cancel the fiber that forked it. The link runs the other way: the child is registered as the caller's, so the caller's interruption reaches it, and the caller ending reaches it too. See *What a fiber owns* in the header — this is where the rule stated there is entered.
The handle is the fiber's own promise with the bookkeeping attached in front, so a caller that has joined or interrupted a child is looking at a tree the child has already left. The promise has no rejection arm because no kernel in this file has one: runKernel turns a throw into a defect and every asynchronous kernel below settles its own errors into an Exit.
This used to be detachedContext, which is forkDaemon under this name. The bug that made was not that a cancelled child kept running — nobody cancels a fiber they cannot see — but that cancelling a *request* left the work it had started writing to a connection that was already closed, with no handle anywhere that could have stopped it.
function
forkDaemon
export function forkDaemon<A, E, R>(self: Effect<A, E, R>): Effect<Fiber<A, E>, empty, R> { ... }
Start self in a fiber that outlives the one that forked it.
The escape from the rule fork keeps, for work whose lifetime is genuinely not the caller's: a cache warmer, a metrics flush, a supervisor started from a request that has no business owning it. Nothing but the returned handle can stop a daemon, so dropping that handle is dropping the work — which is why this is a name a reader can look up rather than an option on fork.
A daemon is detached from its parent, not from its own children: it still ends the fibers it started, or the escape would be inherited by everything below it.
function
forkScoped
export function forkScoped<A, E, R>(self: Effect<A, E, R>): Effect<Fiber<A, E>, empty, R | Scope> { ... }
Start self beside the current fiber, tied to the enclosing Scope.
The third lifetime, and the one a request handler usually means. fork stops the child when the fiber that forked it returns, which is too early for a background refresh that should finish its own work; forkDaemon never stops it, which is a leak with one more step. The lifetime the caller means is neither fiber's — it is the request's, and a request is a Scope.
The child is registered on the scope and deliberately *not* on the forking fiber's children. Being on both would let endFiber stop it the moment the handler returned, which is the behaviour this exists to avoid.
The child gets a scope of its own, and that is an ordering rather than a detail: finalizers run newest first, so a resource the child acquired after the fork would be released *before* the finalizer that stops the child, out from under a fiber still using it. Its own scope closes when it settles, however it settles.
The scope's finalizer interrupts the child and waits for it to stop, so a scoped block does not return until what it started has actually finished and released — the same promise interrupt makes, made by the scope.
function
join
export function join<A, E>(fiber: Fiber<A, E>): Effect<A, E> { ... }
Wait for a fiber and take its result as this effect's result.
function
interrupt
export function interrupt<A, E>(fiber: Fiber<A, E>): Effect<Exit<A, E>> { ... }
Cancel a fiber and wait for it to actually stop.
Interruption is cooperative — the flag is read between steps, and a sleep is woken — so this resolves once the fiber has reached its next checkpoint, not once the request was filed. Returning the Exit rather than void is what makes that observable: a fiber that had already finished reports the value it produced, and one that stopped reports interrupt, so a caller can tell "cancelled in time" from "too late, it was done".
function
ref
export function ref<A>(initial: A): Effect<Ref<A>> { ... }
A place two fibers can both read and write, starting at initial.
An Effect rather than a value, so that making one is part of the program: a Ref built at module scope is shared by every run of that program, which is rarely what anybody wants and never what they meant to write.
Every operation has a synchronous kernel, so a program that uses a Ref can still be answered by runSync.
function
refGet
export function refGet<A>(self: Ref<A>): Effect<A> { ... }
What the ref holds now.
function
refSet
export function refSet<A>(self: Ref<A>, value: A): Effect<void> { ... }
Replace what the ref holds.
function
refModify
export function refModify<A, B>(self: Ref<A>, transform: (value: A) => [B, A]): Effect<B> { ... }
Read, compute a new value and an answer, and write, without yielding.
The one place a Ref is read and written, and the reason the rest of these are one line each. It needs no lock: transform runs between two property accesses in one step, and nothing in this runtime interleaves fibers except at an await. That is also the guarantee's boundary — a transform that returned an Effect would yield, and serialising *that* is what a semaphore is for. See refUpdateEffect.
A transform that throws leaves the ref alone and becomes a defect, because half of a read-modify-write is worse than none of it.
function
refUpdate
export function refUpdate<A>(self: Ref<A>, transform: (value: A) => A): Effect<void> { ... }
Apply a function to what the ref holds.
function
refUpdateAndGet
export function refUpdateAndGet<A>(self: Ref<A>, transform: (value: A) => A): Effect<A> { ... }
Apply a function, and give back what it produced.
function
refGetAndUpdate
export function refGetAndUpdate<A>(self: Ref<A>, transform: (value: A) => A): Effect<A> { ... }
Apply a function, and give back what was there before it.
function
refGetAndSet
export function refGetAndSet<A>(self: Ref<A>, value: A): Effect<A> { ... }
Replace what the ref holds, and give back what was there before.
function
refUpdateEffect
export function refUpdateEffect<A, E, R>(
self: Ref<A>,
lock: Semaphore,
transform: (value: A) => Effect<A, E, R>,
): Effect<A, E, R> { ... }
Update a ref with an effect, one fiber at a time.
This is Effect's SynchronizedRef, and it is a Ref and a Semaphore held together rather than a third opaque type. Passing the lock in is what makes the serialisation visible at the call site, and it lets two refs that must move together share one — which a bundled lock could not express.
refModify needs no lock because it cannot yield. This can, so it must have one: without it, two fibers read the same value, both compute from it, and the second write silently discards the first.
function
deferred
export function deferred<A, E = empty>(): Effect<Deferred<A, E>> { ... }
A value that has not been produced yet, and can be waited for.
Completed at most once: the first deferredSucceed or deferredFail wins and says so by returning true, and every later one returns false rather than overwriting an answer somebody may already have acted on.
function
deferredAwait
export function deferredAwait<A, E>(self: Deferred<A, E>): Effect<A, E> { ... }
Wait for the value, and take its outcome as this effect's outcome.
Interruptible, by the protocol pause uses for a sleep: a waker goes on the fiber's list, and cancelling the fiber ends the wait now. Getting this wrong is how a handshake becomes a fiber interrupt cannot stop, which is the whole reason a hand-written Promise and a let are not good enough for this.
No synchronous kernel: waiting for a value nobody has produced is what this is, and an effect that pretended otherwise would have to answer for a value that does not exist. deferredIsDone is the question with a synchronous answer.
function
deferredSucceed
export function deferredSucceed<A, E>(self: Deferred<A, E>, value: A): Effect<boolean> { ... }
Complete it with a value. true if this call was the one that did.
function
deferredFail
export function deferredFail<A, E>(self: Deferred<A, E>, error: E): Effect<boolean> { ... }
Complete it with a typed failure. true if this call was the one that did.
function
deferredIsDone
export function deferredIsDone<A, E>(self: Deferred<A, E>): Effect<boolean> { ... }
Whether it has been completed, without waiting to find out.
function
semaphore
export function semaphore(permits: number): Effect<Semaphore> { ... }
A budget of permits, shared by whoever holds this.
permits is the capacity and the starting count. Asking for more than the capacity later is a defect rather than a wait, because nothing will ever release enough and a permanent hang is the least debuggable way to say so.
function
withPermit
export function withPermit<A, E, R>(self: Semaphore, body: Effect<A, E, R>): Effect<A, E, R> { ... }
Run body holding one permit.
function
withPermits
export function withPermits<A, E, R>(
self: Semaphore,
permits: number,
body: Effect<A, E, R>,
): Effect<A, E, R> { ... }
Run body holding permits of them, and give them back however it ends.
The same guarantee ensuring gives, and for the same reason: a permit that is not returned when the fiber holding it is interrupted is a budget that shrinks every time somebody cancels a request, until nothing can run at all. finally rather than a finalizer effect, because releasing is a counter and an array splice — it cannot fail, and it must not be interruptible.
A fiber interrupted while *queued* never took a permit, so it returns without releasing one it does not hold.
function
queue
export function queue<A>(capacity: number, strategy?: QueueStrategy): Effect<Queue<A>> { ... }
A place to hand values from one fiber to another, capacity deep.
capacity is at least one: a queue of nothing has no buffer for a sliding strategy to slide and no room for a dropping one to drop into, so the two would silently mean "hand over directly or lose it", which is a rendezvous and not a queue. A Deferred is the rendezvous.
strategy defaults to bounded, which is the one that is back pressure: the default should be the answer that cannot lose a value.
function
queueOffer
export function queueOffer<A>(self: Queue<A>, value: A): Effect<boolean> { ... }
Put a value in, waiting for room if the queue is bounded and full.
true when the queue took the value, false when a dropping queue that was full threw it away. A sliding queue always answers true, because it took the value — what it lost was an older one.
A fiber waiting for room can be interrupted out of the wait, by the protocol deferredAwait and acquirePermits use, and a fiber interrupted while waiting never enqueues the value it was holding: it was never in the queue, and putting it there on the way out would deliver work from a request that had already been cancelled.
function
queueTake
export function queueTake<A>(self: Queue<A>): Effect<A> { ... }
Take the next value, waiting for one if there is none.
Interruptible while waiting, and a taker woken by an interruption leaves the queue exactly as it found it: it is removed from the queue of takers before anything can hand it a value, so the value it did not receive is still there for the next one. Getting that wrong is how a cancelled request eats a piece of work that nobody then does.
A taker that has already been handed a value keeps it even if its fiber is interrupted a moment later. The alternative is dropping a value that has left the queue, and interruption is checked at the caller's next step anyway — a cancelled fiber stops there rather than one step earlier, and the work does not vanish in between.
Unlike deferredAwait this has a synchronous kernel, and the difference is real rather than an inconsistency: taking from a queue that has something in it is reading a buffer, not waiting for a value nobody has produced.
function
queueTakeAll
export function queueTakeAll<A>(self: Queue<A>): Effect<$ReadOnlyArray<A>> { ... }
Everything waiting, without waiting. An empty queue answers with [].
function
queueTakeUpTo
export function queueTakeUpTo<A>(self: Queue<A>, max: number): Effect<$ReadOnlyArray<A>> { ... }
At most max of what is waiting, without waiting for more.
function
queueSize
export function queueSize<A>(self: Queue<A>): Effect<number> { ... }
How many values are waiting to be taken.
The buffer's length, and not Effect's signed count. A fiber blocked in queueOffer is holding a value that is not in the queue, and a fiber blocked in queueTake is not a negative value; a number that means three different things depending on its sign is a worse answer than one that means the one thing its name says.
function
queueShutdown
export function queueShutdown<A>(self: Queue<A>): Effect<void> { ... }
Stop the queue, and stop everybody waiting on it.
Waiting takers and waiting offerers are all interrupted, and what was in the buffer is dropped: a shutdown is a decision that the work is over, and handing out three more values on the way down would be that decision half taken.
Idempotent, and the one operation a shut-down queue still answers normally — along with queueIsShutdown, which has to stay answerable for anything to be able to tell a shutdown from an interruption of its own fiber.
function
queueIsShutdown
export function queueIsShutdown<A>(self: Queue<A>): Effect<boolean> { ... }
Whether the queue has been shut down, which a shut-down queue still answers.
function
pubSub
export function pubSub<A>(capacity: number, strategy?: QueueStrategy): Effect<PubSub<A>> { ... }
A place to publish values every subscriber sees, capacity deep each.
The capacity and the strategy describe one *subscriber's* buffer, because that is where the choice bites: with bounded, the slowest subscriber is what a publisher waits for, and with sliding a subscriber that falls behind loses its oldest values rather than holding the publisher up.
function
pubSubSubscribe
export function pubSubSubscribe<A>(self: PubSub<A>): Effect<Queue<A>, empty, Scope> { ... }
Listen, until the enclosing scope closes.
The Scope is not decoration: a subscription that nothing unsubscribes is a buffer that a publisher keeps filling and nobody keeps draining, which with bounded stops the publisher for ever and with the other two is a leak. So the lifetime is stated in the type, in the same shape acquireRelease states it, and scoped discharges both.
A subscriber sees what is published after it subscribes and nothing that came before, because the buffer it is handed is its own and starts empty.
function
pubSubPublish
export function pubSubPublish<A>(self: PubSub<A>, value: A): Effect<boolean> { ... }
Hand a value to every subscriber, waiting for the slowest one that has back pressure.
true when every subscriber took it, and false when one of them did not — a dropping subscriber that was full, or one whose scope closed while the publish was waiting for it. A pub-sub with no subscribers answers true, because nobody failed to take it.
No synchronous kernel, unlike queueOffer. A publish to a bounded subscriber waits by design, and one that reported "would wait" for the fast subscribers and not the slow one would be an answer about which subscribers happened to be behind.
function
pubSubShutdown
export function pubSubShutdown<A>(self: PubSub<A>): Effect<void> { ... }
Stop the pub-sub, and every subscription with it.
Every subscriber is interrupted where it waits, by the rule a shut-down queue already has. A subscription's scope still runs its own finalizer afterwards, which finds the queue already down and says so by doing nothing.
function
sleep
export function sleep(millis: number): Effect<void> { ... }
Wait, doing nothing.
Interruptible: cancelling the fiber ends the wait now, rather than leaving a timer holding the process open until it fires.
function
delay
export function delay<A, E, R>(self: Effect<A, E, R>, millis: number): Effect<A, E, R> { ... }
Run self after waiting, keeping its result.
function
tap
export function tap<A, E1, E2, R1, R2>(
self: Effect<A, E1, R1>,
body: (value: A) => Effect<mixed, E2, R2>,
): Effect<A, E1 | E2, R1 | R2> { ... }
Look at a success without changing it.
The point of a tap is that it cannot alter the value by accident: logging a result inside a map means one careless edit turns the log's return value into the pipeline's value, and this shape makes that impossible.
function
tapError
export function tapError<A, E, R1, R2>(
self: Effect<A, E, R1>,
body: (error: E) => Effect<mixed, mixed, R2>,
): Effect<A, E, R1 | R2> { ... }
Look at a typed failure without recovering from it.
function
orDie
export function orDie<A, E, R>(self: Effect<A, E, R>): Effect<A, empty, R> { ... }
Turn any typed failure into a defect, so the error channel becomes empty.
For the boundary where a failure is no longer a condition anybody is going to handle — a logging tap, a fire-and-forget notification — and letting it stay in the error channel would only invite a catchAll that pretends to.
function
exit
export function exit<A, E, R>(self: Effect<A, E, R>): Effect<Exit<A, E>, empty, R> { ... }
Reify the outcome, so a failure is a value instead of a short circuit.
either narrows to the typed error and leaves defects and interruption as failures; this keeps the whole Exit, which is what a supervisor or a test that asserts on *how* something failed actually needs.
function
filterOrFail
export function filterOrFail<A, E1, E2, R>(
self: Effect<A, E1, R>,
predicate: (value: A) => boolean,
error: (value: A) => E2,
): Effect<A, E1 | E2, R> { ... }
Keep a success only when it passes predicate, failing with error if not.
The alternative is a flatMap whose body is an if returning succeed or fail, written out at every call site.
function
as
export function as<A, B, E, R>(self: Effect<A, E, R>, value: B): Effect<B, E, R> { ... }
Replace a success with a constant, keeping the failure channel.
function
runPromise
export function runPromise<A, E>(self: Effect<A, E>): Promise<A> { ... }
Run an effect, raising whatever it failed with.
The root fiber ends when this returns, which is what stops a program that forked and did not wait from leaving the fork behind. forkDaemon is how a caller says the work should outlive the run.
function
runPromiseExit
export function runPromiseExit<A, E>(self: Effect<A, E>): Promise<Exit<A, E>> { ... }
Run an effect, returning its outcome rather than raising.
.then rather than async, because an extra async frame costs a microtask on a function whose whole body is one call, and the root fiber has to be ended after the run either way.
function
runSyncExit
export function runSyncExit<A, E>(self: Effect<A, E>): Exit<A, E> { ... }
Run a synchronous effect, returning its outcome rather than raising.
function
runSync
export function runSync<A, E>(self: Effect<A, E>): A { ... }
Run a synchronous effect, raising whatever it failed with.
The counterpart of runPromise for effects with no asynchronous step. An effect that turns out to have one fails as a defect rather than returning a promise nobody is awaiting, because a silently-unawaited promise is how a synchronous-looking call ends up returning undefined.
function
runFork
export function runFork<A, E>(self: Effect<A, E>): Fiber<A, E> { ... }
Start an effect from outside the runtime and keep a handle on it.
The handle is a root fiber, so it owns what it forks in the same way any other fiber does: when it settles, the children it still has are stopped.
Without a doc comment
@uniflowed/effect/stream
opaque-type
Stream
export opaque type Stream<out A, out E = empty, out R = empty> = StreamCarrier<A, E, R>;
Many As, which may fail with an E, and need an R.
The same three channels an Effect has, and they compose the same way: a streamMapEffect over a step that can fail widens E to the union, and a traversal ends where the rest of the package begins — every runner produces an ordinary Effect.
function
streamFromArray
export function streamFromArray<A>(
items: $ReadOnlyArray<A>,
options?: { readonly chunkSize?: number },
): Stream<A> { ... }
Every element of an array, in batches.
The batch size is the throughput knob and nothing else depends on it: a streamTake(3) takes three elements whatever the batches were, because the transforms below truncate batches rather than assuming them.
function
streamFromIterator
export function streamFromIterator<A>(
open: () => Iterator<A>,
options?: { readonly chunkSize?: number },
): Stream<A> { ... }
Every element an iterator produces, in batches.
open rather than an Iterator value, because a stream is traversable more than once and an iterator is not: handing the same exhausted iterator to a second traversal would make the second one empty, which is the kind of bug that shows up a long way from here.
The source may be infinite. streamTake and a failing step both end a traversal without draining it, and the batch size bounds how far past the last wanted element the source is asked to go.
function
streamFromEffect
export function streamFromEffect<A, E, R>(self: Effect<A, E, R>): Stream<A, E, R> { ... }
One element: the value the effect produces, or its failure.
function
streamPaginate
export function streamPaginate<A, S, E, R>(
initial: S,
page: (cursor: S) => Effect<{ readonly items: $ReadOnlyArray<A>, readonly next: ?S }, E, R>,
): Stream<A, E, R> { ... }
A page at a time, until a page says there is no next one.
The shape a paginated API actually has, and the reason a stream earns its place here rather than a forEach over a list of page numbers: the number of pages is not known until the last one says so.
function
streamFromReadableStream
export function streamFromReadableStream<A>(open: () => ReadableChunks<A>): Stream<A> { ... }
A web ReadableStream, as an effect stream.
Response.body on every host uf targets, which is the concrete reason this module exists rather than the checklist one. A batch here is one chunk from the reader, because that is the batching the host already chose.
Cancelling the traversal cancels the reader, so a streamTake over a response body closes the connection rather than reading it to the end.
function
streamMap
export function streamMap<A, B, E, R>(
self: Stream<A, E, R>,
transform: (value: A) => B,
): Stream<B, E, R> { ... }
Change every element.
A transform that throws is a defect, because map over an Effect already says so and a stream should not quietly disagree with the package it is built on.
function
streamFilter
export function streamFilter<A, E, R>(
self: Stream<A, E, R>,
keep: (value: A) => boolean,
): Stream<A, E, R> { ... }
Keep the elements that pass. An emptied batch is not the end.
function
streamTake
export function streamTake<A, E, R>(self: Stream<A, E, R>, count: number): Stream<A, E, R> { ... }
The first count elements, and then stop pulling.
The traversal ends without draining the source, and the runner still closes it — which is what makes streamTake(3) of an infinite source release what the source opened rather than leaking it.
function
streamTap
export function streamTap<A, E1, E2, R1, R2>(
self: Stream<A, E1, R1>,
body: (value: A) => Effect<mixed, E2, R2>,
): Stream<A, E1 | E2, R1 | R2> { ... }
Look at every element without changing it.
function
streamMapEffect
export function streamMapEffect<A, B, E1, E2, R1, R2>(
self: Stream<A, E1, R1>,
body: (value: A) => Effect<B, E2, R2>,
options?: { readonly concurrency?: number },
): Stream<B, E1 | E2, R1 | R2> { ... }
Change every element with an effect, up to concurrency at a time.
Order is preserved: the results come back in the order the elements arrived, whatever order the effects finished in, because forEach already promises that and this is a forEach over a window.
The window is filled across batch boundaries, so the concurrency a caller asks for is the concurrency they get whatever the source's batching was. Without that, a source handing out one element at a time would silently make every concurrency limit mean one.
There is no "unbounded", unlike all and forEach. A stream has no length, so unbounded here would mean buffering the whole of it — which is the thing a stream exists to avoid, and an option that silently means something else is worse than one that is not offered.
function
streamEnsuring
export function streamEnsuring<A, E, R>(
self: Stream<A, E, R>,
finalizer: () => Effect<mixed, mixed, empty>,
): Stream<A, E, R> { ... }
Run finalizer when a traversal ends, however it ends.
After the stream's own close, so finalizers run outermost last — the order ensuring gives an effect, said again for a traversal.
function
streamRunFold
export function streamRunFold<A, B, E, R>(
self: Stream<A, E, R>,
initial: B,
step: (state: B, value: A) => B,
): Effect<B, E, R> { ... }
Consume the whole stream into one value.
The runner every other runner is written in terms of, and the one place a traversal is opened and closed. ensuring is what closes it: the traversal ends on success, on a failure from a pull, on a defect from step, and on the fiber being interrupted, and the source is closed in all four — the same guarantee acquireRelease gives, reached the same way.
The interruption checkpoint is effect's own, between steps, so a fiber draining a stream stops between pulls rather than in the middle of one.
function
streamRunCollect
export function streamRunCollect<A, E, R>(self: Stream<A, E, R>): Effect<$ReadOnlyArray<A>, E, R> { ... }
Every element, in order, as an array.
function
streamRunDrain
export function streamRunDrain<A, E, R>(self: Stream<A, E, R>): Effect<void, E, R> { ... }
Run the stream for its effects and discard its elements.
function
streamRunForEach
export function streamRunForEach<A, E1, E2, R1, R2>(
self: Stream<A, E1, R1>,
body: (value: A) => Effect<mixed, E2, R2>,
): Effect<void, E1 | E2, R1 | R2> { ... }
Run body for every element, in order.
function
streamRunHead
export function streamRunHead<A, E, R>(self: Stream<A, E, R>): Effect<?A, E, R> { ... }
The first element, or null for a stream that had none.
Stops after one: the source is closed without being drained, which is the whole difference between this and streamRunCollect(...)[0].
function
streamFromQueue
export function streamFromQueue<A>(
source: Queue<A>,
options?: { readonly chunkSize?: number },
): Stream<A> { ... }
Everything a queue is handed, until it is shut down.
A queue has no end of its own, so one has to be agreed: queueShutdown is it, and a traversal that finds the queue shut down ends rather than failing. That is the same decision Queue already took for a shutdown — it is an interruption and not a typed failure — read from the consumer's side.
A batch is up to chunkSize of whatever is waiting, so a fast producer is consumed in batches rather than one value at a time, and a slow one does not make the traversal wait for a batch to fill.
function
streamBuffer
export function streamBuffer<A, E, R>(self: Stream<A, E, R>, capacity: number): Stream<A, E, R> { ... }
Let the source run ahead of the consumer, up to capacity batches.
The first combinator here that needs a fiber of its own: the source is drained into a bounded queue by a fiber the traversal owns, and the pull takes from the queue. A consumer that is slower than the source stops the source at the queue's bound rather than at its own speed, which is the difference between a pipeline that overlaps and one that alternates.
bounded and not dropping: a buffer that silently lost elements would make streamBuffer change what a stream contains rather than when it arrives.
function
streamMerge
export function streamMerge<A, E1, E2, R1, R2>(
left: Stream<A, E1, R1>,
right: Stream<A, E2, R2>,
options?: { readonly capacity?: number },
): Stream<A, E1 | E2, R1 | R2> { ... }
Both streams' elements, in whatever order they arrive.
Two fibers filling one bounded queue, which is what makes this a merge rather than a concatenation: neither side waits for the other, and the queue's bound is what stops the faster one from running away. The traversal ends when both sides have, and a failure on either side ends it with that failure.
There is no ordering promise between the sides, which is what "merge" means. streamZip is the combinator with one.
function
streamZip
export function streamZip<A, B, E1, E2, R1, R2>(
left: Stream<A, E1, R1>,
right: Stream<B, E2, R2>,
): Stream<[A, B], E1 | E2, R1 | R2> { ... }
Pairs, until either side runs out.
The one combinator here that needs no queue and no fiber: both sides are pulled in step and the leftovers of the longer batch are kept until the shorter one catches up. A queue would buy nothing, because a zip cannot get ahead of its slower side by definition.
The traversal ends with the shorter stream, and the longer one is closed without being drained — which is what makes zipping an infinite source with a finite one terminate.
function
streamToReadableStream
export function streamToReadableStream<A, E, Made>(
self: Stream<A, E>,
make: (source: ChunkSource<A>) => Made,
): Made { ... }
A stream, as something a web consumer can read.
The other end of streamFromReadableStream, and the one @uniflowed/server needs: streaming SSR and RSC produce a ReadableStream, and until this existed an effect program had no way to be on that end of one. Three decisions, none of which is about the function's body:
**Which fiber the pull runs on.** One per host pull, started with runFork and kept in a closure that cancel can reach. A ReadableStream's pull is a callback the *host* calls, so there is no fiber to inherit and no interruption to propagate — the handle is the only thing that can stop the work, which is exactly the situation runFork exists for. The traversal itself spans many pulls and is opened on the first of them, so a ReadableStream nobody reads never opens what the source would have.
**What a typed failure becomes.** A ReadableStream has one error(reason) and no channels, so the three ways an effect can fail cannot stay three. A typed failure crosses as its own value, because E is the type the consumer named and an Error wrapped round it would lose it. A defect and an interruption cross as an Error, because neither has a value anybody named — a defect has a message and an interruption has only the fact.
**How the host's constructor is reached.** By being handed it: streamToReadableStream(stream, (source) => new ReadableStream(source)). That is the same avoidance streamFromReadableStream makes by taking open, for the same reason — this module cannot name a global that four hosts declare in four places — and it is why the return type is whatever the caller's constructor produced rather than a type this module invented.
The guarantees are the runners': a consumer that cancels stops the pull it interrupted and closes the traversal, so a ReadableStream abandoned halfway releases what its source opened, exactly as streamTake(3) of an infinite source does. Nothing is enqueued after a cancel, because the controller is closed by then and touching it would raise inside the host.
R is empty, as it is for every runner in this package: an effect that still needs a service has nowhere to get one from outside the runtime.
uf