Hub

Initializes a new instance of the class that is empty and has the default initial capacity.

An abbreviation for the CLI type . Basic Types

Represents a cold workflow that reads an environment, returns a typed result, and is executed explicitly through one of its execution members such as ToTask, ToAsync, or RunSynchronously. The type of the environment dependency. The type of the failure value. The type of the success value.

An environment containing only a clock, for timed flows with no other services.

Represents an error channel that cannot occur.

The type 'unit', which has only one value "()". This value is special and always uses the representation 'null'. Basic Types

Defers flow construction until execution time. A function that returns the flow to execute. A flow that lazily evaluates the factory when executed. let flow = Flow.delay (fun () -> Flow.succeed 42)

Adds an object to the end of the . The object to be added to the end of the . The value can be for reference types.

Creates a successful synchronous flow. The value to wrap in a successful flow. A flow that always succeeds with the provided value.

Broadcasts every published value to every current subscription. Create one with Hub.make. Use a hub when there can be zero or many consumers. The type of the published values.

Apply a function to a value, the value being on the left, the function on the right The argument. The function. The function result. let doubleIt x = x * 2 3 |> doubleIt // Evaluates to 6

Creates a stream of the values published to a hub, subscribing for the life of the stream. The stream subscribes with when it starts and unsubscribes when it ends, so the subscription can never outlive its consumer. It ends normally once the hub is shut down and the backlog is drained. Values published before the stream starts are not delivered. When a consumer is forked and must see values published straight afterwards, subscribe first with Hub.subscribe and consume the subscription with FlowStream.fromDequeue. readings |> FlowStream.fromHub (QueueStrategy.Sliding 1) |> FlowStream.runForEach display

What a queue, or a hub subscription, does with a value offered while it is full. Lossless delivery needs either back-pressure or unbounded memory. BackPressure keeps every value by making the producer wait; Unbounded keeps every value by growing without limit; Dropping and Sliding never make the producer wait and lose values instead. A non-positive capacity fails the flow that uses the strategy with a defect.

A full buffer evicts its oldest value to make room for the new one.

Runs an effectful action for every stream value. stream |> FlowStream.runForEachFlow save

Creates a stream from a synchronous sequence of values. The sequence of values to be emitted by the stream. A that yields each value from the sequence. FlowStream.fromSeq [1..10] |> FlowStream.runCollect |> Flow.run ()

Runs a stream and publishes every value to a hub. Each value is published with Hub.publish, so a full BackPressure subscription slows the stream down. If the hub is shut down first, the flow is interrupted. sensor |> FlowStream.runIntoHub readings

The universal flow { } computation expression.

Broadcasts every published value to every current subscription. Create one with Hub.make. Use a hub when there can be zero or many consumers. The type of the published values.

An abbreviation for the CLI type . Basic Types

Creates hubs, publishes to them, and subscribes to them. A publish reaches every subscription that exists when it starts; a subscriber sees only values published after its subscribe completes, so a late joiner that needs current state must get it elsewhere, for example from a SubscriptionRef. Publishes are serialized, so every subscriber observes values in the same order even with concurrent publishers. Dropping, Sliding, and Unbounded subscriptions never delay the publisher. A BackPressure subscription with a full buffer suspends publish until it has room or its subscription ends. While the publisher waits, no later value reaches any subscriber, lossy ones included: keeping one order for every subscriber means a stalled lossless subscriber stalls the feed. Size a lossless subscriber's buffer for the bursts you expect, watch Dequeue.size to raise an alarm before it fills, and use Hub.tryPublish where the publisher must never wait. Shutting a hub down shuts every subscription down with Dequeue.shutdown semantics: subscribers drain their backlog, streams over subscriptions end normally, and later publishes are interrupted.

Creates a hub with no subscriptions.

Apply a function to a value, the value being on the left, the function on the right The argument. The function. The function result. let doubleIt x = x * 2 3 |> doubleIt // Evaluates to 6

Starts a flow in a new fiber without waiting for it to complete. Forking turns a cold flow description into hot child work and returns a handle that can later be joined or interrupted. Prefer zipPar or race when the caller only needs a simple parallel composition. Wait for the handle with Fiber.join or Fiber.await, and stop it with Fiber.interrupt. The flow to fork. A flow that produces a handle.

Structural equality The first parameter. The second parameter. The result of the comparison. 5 = 5 // Evaluates to true 5 = 6 // Evaluates to false [1; 2] = [1; 2] // Evaluates to true (1, 5) = (1, 6) // Evaluates to false

Returns the number of current subscriptions.

Suspends the flow for the specified duration, observing cancellation. The duration to sleep. A flow that completes after the specified delay, or is interrupted if cancelled first.

Represents a time interval.

Returns a that represents a specified number of milliseconds. A number of milliseconds. An object that represents . is less than or greater than . -or- is . -or- is . is equal to .

Shuts the hub and all of its subscriptions down. Calling it again has no effect.

Operations on a running , the handle returned by Flow.fork. Every operation returns a flow; nothing waits or interrupts until that flow runs. Reading a fiber's outcome through join, await, or interrupt marks it observed, so a defect it died with is not also reported as unobserved.

Waits for a fiber and returns its value, failing the same way the fiber failed. Joining preserves the child's error channel: a Cause.Fail becomes the same typed error, and interruption and defects remain interruption and defects. Use await to inspect the outcome instead. The fiber to join. A flow that completes with the fiber's value. let loadProfile : Flow<unit, string, string> = Flow.ok "profile" let loadOrders : Flow<unit, string, int list> = Flow.ok [ 1; 2 ] let page = flow { let! fiber = Flow.fork loadProfile let! orders = loadOrders let! profile = Fiber.join fiber return profile, orders }

Runs the workflow and blocks until the final exit is available. The environment used by the workflow. The workflow to run. The final workflow exit. let exit = workflow |> Flow.run environment

Helpers for the clock service.

Creates a live clock backed by and a monotonic timer.

Represents the final outcome of a workflow execution. The type of the success value. The type of the domain-specific failure value.

The workflow completed successfully.

Contains operations for working with values of type .

Returns the last element of the sequence. The input sequence. The last element of the sequence. Thrown when the input sequence is null. Thrown when the input does not have any elements. ["pear"; "banana"] |> Seq.last Evaluates to banana [] |> Seq.last Throws ArgumentException This is an O(n) operation, where n is the length of the sequence.