Queue

The universal flow { } computation expression.

An asynchronous FIFO queue for handing values between fibers, with an overflow strategy chosen at creation. Create one with Queue.make or a shorthand such as Queue.bounded, offer with Queue.offer, and take with the Dequeue functions. A queue is meant for one logical consumer; use Hub when zero or many consumers each need every value. The type of the queued values.

An abbreviation for the CLI type . Basic Types

Creates queues and offers values to them. Take values, inspect, and shut a queue down with the Dequeue functions.

Creates a queue with the given overflow strategy. A non-positive capacity fails the returned flow with a defect. Queue.make (QueueStrategy.Sliding 1)

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.

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

Adds values in order according to the queue's strategy. true when every value was accepted. A dropping queue accepts values while it has room and returns false if it discarded any. A bounded queue suspends as needed; values from other producers may be interleaved between the batch's values, but the batch's own order is kept. If the flow is interrupted part way, the values already accepted stay in the queue. flow { let! (samples: Queue<float>) = Queue.unbounded () do! samples |> Queue.offerAll [ 1.0; 2.0; 3.0 ] |> Flow.ignore return! samples |> Dequeue.takeUpTo 500 }

The standard overloaded range operator, e.g. [n..m] for lists, seq {n..m} for sequences The start value of the range. The end value of the range. The sequence spanning the range. [1..4] // Evaluates to [1; 2; 3; 4] [1.5..4.4] // Evaluates to [1.5; 2.5; 3.5] ['a'..'d'] // Evaluates to ['a'; 'b'; 'c'; 'd'] [|1..4|] // Evaluates to an array [|1; 2; 3; 4|] { 1..4 } // Evaluates to a sequence [1; 2; 3; 4])

Maps the successful value of a synchronous flow to unit. The source flow. A flow that succeeds with unit instead of the original value. let flow = Flow.succeed 42 |> Flow.ignore

Takes values from a , inspects it, and shuts it down. Every function here also accepts a .

Removes every available value in FIFO order, without suspending.

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

Represents an error channel that cannot occur.

The type of immutable singly-linked lists. See the module for further operations related to lists. Use the constructors [] and :: (infix) to create values of this type, or the notation [1; 2; 3]. Use the values in the List module to manipulate values of this type, or pattern match against the values directly. See also F# Language Guide - Lists.

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

The universal flow { } computation expression.

An asynchronous FIFO queue for handing values between fibers, with an overflow strategy chosen at creation. Create one with Queue.make or a shorthand such as Queue.bounded, offer with Queue.offer, and take with the Dequeue functions. A queue is meant for one logical consumer; use Hub when zero or many consumers each need every value. The type of the queued values.

An abbreviation for the CLI type . Basic Types

Creates queues and offers values to them. Take values, inspect, and shut a queue down with the Dequeue functions.

Creates a lossless queue whose offer suspends while it holds values. Shorthand for Queue.make (QueueStrategy.BackPressure capacity). Queue.bounded 64

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 that takes values from a queue or hub subscription until it is shut down and drained. Shutdown is the normal end of the stream, not a failure: after Dequeue.shutdown or Hub.shutdown the stream emits the remaining backlog and then completes. Each pull suspends while the queue is empty. Interrupting a pull leaves any value it would have received in the queue. flow { let! (jobs: Queue<string>) = Queue.bounded 8 do! jobs |> Queue.offerAll [ "a"; "b" ] |> Flow.ignore do! Dequeue.shutdown jobs return! jobs |> FlowStream.fromDequeue |> FlowStream.runCollect }

Groups consecutive values into non-empty lists of at most elements. The operator pulls and retains at most upstream values for each emitted list.

Collects all emitted values into a list. stream |> FlowStream.runCollect

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 fiber runs in its own scope and releases its scoped resources (anything registered with a scope, such as scopeResource, scopeFinalizer, or a hub subscription) when it settles, whether or not anyone joins it. A value the fiber returns therefore outlives the scoped resources the fiber acquired: a resource returned through Fiber.join has already been released. Use forkResource to acquire a resource in the background and keep it, or zipPar and sequencePar to acquire several in parallel. The flow to fork. A flow that produces a handle.

Adds values in order according to the queue's strategy. true when every value was accepted. A dropping queue accepts values while it has room and returns false if it discarded any. A bounded queue suspends as needed; values from other producers may be interleaved between the batch's values, but the batch's own order is kept. If the flow is interrupted part way, the values already accepted stay in the queue. flow { let! (samples: Queue<float>) = Queue.unbounded () do! samples |> Queue.offerAll [ 1.0; 2.0; 3.0 ] |> Flow.ignore return! samples |> Dequeue.takeUpTo 500 }

The standard overloaded range operator, e.g. [n..m] for lists, seq {n..m} for sequences The start value of the range. The end value of the range. The sequence spanning the range. [1..4] // Evaluates to [1; 2; 3; 4] [1.5..4.4] // Evaluates to [1.5; 2.5; 3.5] ['a'..'d'] // Evaluates to ['a'; 'b'; 'c'; 'd'] [|1..4|] // Evaluates to an array [|1; 2; 3; 4|] { 1..4 } // Evaluates to a sequence [1; 2; 3; 4])

Maps the successful value of a synchronous flow to unit. The source flow. A flow that succeeds with unit instead of the original value. let flow = Flow.succeed 42 |> Flow.ignore

Takes values from a , inspects it, and shuts it down. Every function here also accepts a .

Shuts the queue down. Calling it again has no effect. Either side may shut a queue down. On a hub subscription this unsubscribes: the hub stops delivering to it, and a publisher waiting on it moves on.

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 }

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

Represents an error channel that cannot occur.

The type of immutable singly-linked lists. See the module for further operations related to lists. Use the constructors [] and :: (infix) to create values of this type, or the notation [1; 2; 3]. Use the values in the List module to manipulate values of this type, or pattern match against the values directly. See also F# Language Guide - Lists.

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.

The universal flow { } computation expression.

An asynchronous FIFO queue for handing values between fibers, with an overflow strategy chosen at creation. Create one with Queue.make or a shorthand such as Queue.bounded, offer with Queue.offer, and take with the Dequeue functions. A queue is meant for one logical consumer; use Hub when zero or many consumers each need every value. The type of the queued values.

An abbreviation for the CLI type . Basic Types

Creates queues and offers values to them. Take values, inspect, and shut a queue down with the Dequeue functions.

Creates a lossless queue whose offer suspends while it holds values. Shorthand for Queue.make (QueueStrategy.BackPressure capacity). Queue.bounded 64

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 that takes values from a queue or hub subscription until it is shut down and drained. Shutdown is the normal end of the stream, not a failure: after Dequeue.shutdown or Hub.shutdown the stream emits the remaining backlog and then completes. Each pull suspends while the queue is empty. Interrupting a pull leaves any value it would have received in the queue. flow { let! (jobs: Queue<string>) = Queue.bounded 8 do! jobs |> Queue.offerAll [ "a"; "b" ] |> Flow.ignore do! Dequeue.shutdown jobs return! jobs |> FlowStream.fromDequeue |> FlowStream.runCollect }

Collects all emitted values into a list. stream |> FlowStream.runCollect

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 fiber runs in its own scope and releases its scoped resources (anything registered with a scope, such as scopeResource, scopeFinalizer, or a hub subscription) when it settles, whether or not anyone joins it. A value the fiber returns therefore outlives the scoped resources the fiber acquired: a resource returned through Fiber.join has already been released. Use forkResource to acquire a resource in the background and keep it, or zipPar and sequencePar to acquire several in parallel. The flow to fork. A flow that produces a handle.

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 several streams concurrently and emits their values as they arrive. Each stream keeps its own order; values from different streams interleave in arrival order. The merged stream ends when every stream has ended. The first failure is delivered after the values already buffered, and the other streams are interrupted. When the consumer stops early, every stream is interrupted. [ readerA; readerB; readerC ] |> FlowStream.mergePar |> FlowStream.runIntoQueue controlInputs

Runs a stream and offers every value to a queue, waiting while a bounded queue is full. The queue is not shut down when the stream ends, so several streams can feed one queue; shut it down with Dequeue.shutdown once every producer has finished. If the queue is shut down first, the flow is interrupted. deviceReadings |> FlowStream.runIntoQueue controlInputs

Takes values from a , inspects it, and shuts it down. Every function here also accepts a .

Shuts the queue down. Calling it again has no effect. Either side may shut a queue down. On a hub subscription this unsubscribes: the hub stops delivering to it, and a publisher waiting on it moves on.

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 }

Contains operations for working with values of type . Operations for collections such as lists, arrays, sets, maps and sequences. See also F# Collection Types in the F# Language Guide.

Sorts the given list using . This is a stable sort, i.e. the original order of equal elements is preserved. This is an O(n log n) operation, where n is the length of the list. The input list. The sorted list. let input = [8; 4; 3; 1; 6; 1] List.sort input Evaluates to [1; 1; 3; 4; 6; 8].

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

Represents an error channel that cannot occur.

The type of immutable singly-linked lists. See the module for further operations related to lists. Use the constructors [] and :: (infix) to create values of this type, or the notation [1; 2; 3]. Use the values in the List module to manipulate values of this type, or pattern match against the values directly. See also F# Language Guide - Lists.

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.

The universal flow { } computation expression.

An asynchronous FIFO queue for handing values between fibers, with an overflow strategy chosen at creation. Create one with Queue.make or a shorthand such as Queue.bounded, offer with Queue.offer, and take with the Dequeue functions. A queue is meant for one logical consumer; use Hub when zero or many consumers each need every value. The type of the queued values.

An abbreviation for the CLI type . Basic Types

Creates queues and offers values to them. Take values, inspect, and shut a queue down with the Dequeue functions.

Creates a queue that never rejects or suspends an offer. Memory grows with the backlog; prefer a bounded queue unless producers are otherwise limited.

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

Adds values in order according to the queue's strategy. true when every value was accepted. A dropping queue accepts values while it has room and returns false if it discarded any. A bounded queue suspends as needed; values from other producers may be interleaved between the batch's values, but the batch's own order is kept. If the flow is interrupted part way, the values already accepted stay in the queue. flow { let! (samples: Queue<float>) = Queue.unbounded () do! samples |> Queue.offerAll [ 1.0; 2.0; 3.0 ] |> Flow.ignore return! samples |> Dequeue.takeUpTo 500 }

The standard overloaded range operator, e.g. [n..m] for lists, seq {n..m} for sequences The start value of the range. The end value of the range. The sequence spanning the range. [1..4] // Evaluates to [1; 2; 3; 4] [1.5..4.4] // Evaluates to [1.5; 2.5; 3.5] ['a'..'d'] // Evaluates to ['a'; 'b'; 'c'; 'd'] [|1..4|] // Evaluates to an array [|1; 2; 3; 4|] { 1..4 } // Evaluates to a sequence [1; 2; 3; 4])

Maps the successful value of a synchronous flow to unit. The source flow. A flow that succeeds with unit instead of the original value. let flow = Flow.succeed 42 |> Flow.ignore

Takes values from a , inspects it, and shuts it down. Every function here also accepts a .

Removes between and values in FIFO order, suspending until at least are available. This is the batching consumer's take: it waits for work, then takes everything that has accumulated up to a batch limit. It takes nothing until min values are available and then takes the whole batch at once, so an interruption while it waits leaves every value in the queue. Takers are served in the order they started waiting, a single take and a takeBetween alike. After shutdown it returns the remaining values even if there are fewer than min, and is interrupted once nothing remains. min must be at least 1, no greater than max, and no greater than a bounded queue's capacity; otherwise the flow fails with a defect. // Wait for at least one sample, then write up to 500 in one batch. samples |> Dequeue.takeBetween 1 500

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

Represents an error channel that cannot occur.

The type of immutable singly-linked lists. See the module for further operations related to lists. Use the constructors [] and :: (infix) to create values of this type, or the notation [1; 2; 3]. Use the values in the List module to manipulate values of this type, or pattern match against the values directly. See also F# Language Guide - Lists.

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

The universal flow { } computation expression.

An asynchronous FIFO queue for handing values between fibers, with an overflow strategy chosen at creation. Create one with Queue.make or a shorthand such as Queue.bounded, offer with Queue.offer, and take with the Dequeue functions. A queue is meant for one logical consumer; use Hub when zero or many consumers each need every value. The type of the queued values.

An abbreviation for the CLI type . Basic Types

Creates queues and offers values to them. Take values, inspect, and shut a queue down with the Dequeue functions.

Creates a queue whose offer evicts the oldest value when full and returns true. Shorthand for Queue.make (QueueStrategy.Sliding capacity).

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

Adds values in order according to the queue's strategy. true when every value was accepted. A dropping queue accepts values while it has room and returns false if it discarded any. A bounded queue suspends as needed; values from other producers may be interleaved between the batch's values, but the batch's own order is kept. If the flow is interrupted part way, the values already accepted stay in the queue. flow { let! (samples: Queue<float>) = Queue.unbounded () do! samples |> Queue.offerAll [ 1.0; 2.0; 3.0 ] |> Flow.ignore return! samples |> Dequeue.takeUpTo 500 }

The standard overloaded range operator, e.g. [n..m] for lists, seq {n..m} for sequences The start value of the range. The end value of the range. The sequence spanning the range. [1..4] // Evaluates to [1; 2; 3; 4] [1.5..4.4] // Evaluates to [1.5; 2.5; 3.5] ['a'..'d'] // Evaluates to ['a'; 'b'; 'c'; 'd'] [|1..4|] // Evaluates to an array [|1; 2; 3; 4|] { 1..4 } // Evaluates to a sequence [1; 2; 3; 4])

Maps the successful value of a synchronous flow to unit. The source flow. A flow that succeeds with unit instead of the original value. let flow = Flow.succeed 42 |> Flow.ignore

Takes values from a , inspects it, and shuts it down. Every function here also accepts a .

Reads the queue's backlog and counters in one consistent snapshot. flow { let! stats = Dequeue.stats historian if stats.Capacity |> Option.exists (fun limit -> stats.Size * 10 >= limit * 8) then do! raiseAlarm "historian buffer 80% full" }

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

Represents an error channel that cannot occur.

An abbreviation for the CLI type . Basic Types

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

The universal flow { } computation expression.

An asynchronous FIFO queue for handing values between fibers, with an overflow strategy chosen at creation. Create one with Queue.make or a shorthand such as Queue.bounded, offer with Queue.offer, and take with the Dequeue functions. A queue is meant for one logical consumer; use Hub when zero or many consumers each need every value. The type of the queued values.

An abbreviation for the CLI type . Basic Types

Creates queues and offers values to them. Take values, inspect, and shut a queue down with the Dequeue functions.

Creates a queue that is shut down when the current scope closes. Use it inside Flow.scoped, a forked fiber, or an application root. Closing the scope interrupts fibers suspended on the queue and ends any FlowStream.fromDequeue consumer after it drains the backlog. flow { let! (jobs: Queue<string>) = Queue.makeScoped (QueueStrategy.BackPressure 64) do! jobs |> Queue.offer "first job" |> Flow.ignore } |> Flow.scoped

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.

Lossless. A full buffer suspends the producer until a value is taken.

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

Adds values in order according to the queue's strategy. true when every value was accepted. A dropping queue accepts values while it has room and returns false if it discarded any. A bounded queue suspends as needed; values from other producers may be interleaved between the batch's values, but the batch's own order is kept. If the flow is interrupted part way, the values already accepted stay in the queue. flow { let! (samples: Queue<float>) = Queue.unbounded () do! samples |> Queue.offerAll [ 1.0; 2.0; 3.0 ] |> Flow.ignore return! samples |> Dequeue.takeUpTo 500 }

Maps the successful value of a synchronous flow to unit. The source flow. A flow that succeeds with unit instead of the original value. let flow = Flow.succeed 42 |> Flow.ignore

Runs a flow in a child scope and closes that scope before returning. Resources and fibers acquired inside the flow are released after success, typed failure, defect, or interruption without waiting for the surrounding application scope to close.

Creates a stream that takes values from a queue or hub subscription until it is shut down and drained. Shutdown is the normal end of the stream, not a failure: after Dequeue.shutdown or Hub.shutdown the stream emits the remaining backlog and then completes. Each pull suspends while the queue is empty. Interrupting a pull leaves any value it would have received in the queue. flow { let! (jobs: Queue<string>) = Queue.bounded 8 do! jobs |> Queue.offerAll [ "a"; "b" ] |> Flow.ignore do! Dequeue.shutdown jobs return! jobs |> FlowStream.fromDequeue |> FlowStream.runCollect }

Collects all emitted values into a list. stream |> FlowStream.runCollect

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

Represents an error channel that cannot occur.

The type of immutable singly-linked lists. See the module for further operations related to lists. Use the constructors [] and :: (infix) to create values of this type, or the notation [1; 2; 3]. Use the values in the List module to manipulate values of this type, or pattern match against the values directly. See also F# Language Guide - Lists.

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