Repository F# setup

open System
open System.IO
open System.Threading
open System.Threading.Tasks
open Axial
open Axial.Layers
open Axial.Console
open Axial.FileSystem
open Axial.Hosting
open Axial.Hosting.Browser
open Axial.Hosting.Node
open Axial.PlatformService
open Axial.State
open Axial.Telemetry
open Axial.Telemetry.JavaScript

Constructing Streams

An array or seq is the right source when all values already exist and producing the next value is synchronous. Wrap one in a FlowStream to compose it with effectful sources and incremental consumers under one failure, cancellation, backpressure, and cleanup model.

Use FlowStream.fromSeq, FlowStream.singleton, and FlowStream.empty for those existing values:

Verified

> (FlowStream.fromSeq [ 1..100 ] : FlowStream<int>)
- |> FlowStream.take 3
- |> FlowStream.append (FlowStream.singleton 42)
- |> FlowStream.runCollect
- |> Flow.run ();;

val it: Exit<int list,Never> = Success [1; 2; 3; 42]

Axial.FlowStreamModule

fromSeq: 'value seq -> FlowStream<'env,'error,'value>

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 ()

(..): ^T -> ^T -> ^T seq

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])

FlowStream

A stream with no environment requirement and no typed failure.

int

An abbreviation for the CLI type . Basic Types

(|>): 'T1 -> ('T1 -> 'U) -> 'U

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

take: int -> FlowStream<'a,'b,'c> -> FlowStream<'a,'b,'c>

Emits at most values. stream |> FlowStream.take 10

append: FlowStream<'a,'b,'c> -> FlowStream<'a,'b,'c> -> FlowStream<'a,'b,'c>

Concatenates two streams, evaluating the second only after the first ends. first |> FlowStream.append second

singleton: 'value -> FlowStream<'env,'error,'value>

Creates a stream containing one value. FlowStream.singleton 42

runCollect: FlowStream<'a,'b,'c> -> Flow<'a,'b,'c list>

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

Axial.Flow

run: 'env -> Flow<'env,'error,'value> -> Exit<'value,'error>

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

FlowStream.empty emits nothing. FlowStream.fromSeq obtains its enumerator only when consumption starts. The enumerator is registered with the stream's Flow scope and disposed after completion or early termination.

Bridge callbacks

FlowStream.fromCallback capacity subscribe creates a bounded stream from a synchronous callback. The subscription returns an IDisposable in a Flow; its disposal runs when consumption completes, fails, stops early, or is interrupted. If subscription attaches a handler before a later step can fail, register its cleanup with Flow.scopeFinalizer. Emit.Value returns false when the buffer is full (dropping the newest value) or the stream has ended. Emit.Fail ends with a typed error; Emit.End completes normally. Later callbacks are ignored. Use FlowStream.fromCallbackWith capacity (CallbackOverflow.Fail error) subscribe to fail rather than drop on overflow: previously queued values are delivered before the failure.

Verified

> open System;;
> open System.Threading.Tasks;;
> open Axial;;
> let notifications () : FlowStream<unit, string, int> = FlowStream.fromCallback 3 (fun emit -> Flow.delay (fun () -> Task.Run(fun () -> [1; 2; 3] |> List.iter (fun i -> emit.Value i |> ignore); emit.End()) |> ignore; Flow.ok ({ new IDisposable with member _.Dispose() = () })));;
> notifications () |> FlowStream.runCollect |> Flow.run ();;

val it: Exit<int list,string> = Success [1; 2; 3]

System

Threading

Tasks

Axial

notifications: unit -> FlowStream<unit,string,int>

Axial.FlowStream`3

Represents a cold stream of values that requires an environment, can fail with a typed error, and supports backpressure. The type of the environment dependency. The type of the failure value. The type of the success values in the stream.

unit

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

string

An abbreviation for the CLI type . Basic Types

int

An abbreviation for the CLI type . Basic Types

Axial.FlowStreamModule

fromCallback: int -> (Emit<'error,'value> -> Flow<'env,'error,IDisposable>) -> FlowStream<'env,'error,'value>

Bridges a synchronous callback into a bounded, scope-owned stream. When full, the newest value is dropped; Value returns false. The subscription is disposed on completion, failure, interruption, or early termination. Subscribe should register cleanup with Flow.scopeFinalizer for anything it attaches before a later step can fail or be interrupted. Callbacks after completion are ignored.

emit: Emit<string,int>

Axial.Flow

delay: (unit -> Flow<'env,'error,'value>) -> Flow<'env,'error,'value>

Defers flow construction until execution time. A function that returns the flow to execute. Recursive calls deferred here are consumed by the executor without growing a chain of parent tasks. Work that must run after the recursive call still needs storage until that call finishes. A flow that lazily evaluates the factory when executed. let flow = Flow.delay (fun () -> Flow.succeed 42)

System.Threading.Tasks.Task

Represents an asynchronous operation.

Run: Action -> Task

Queues the specified work to run on the thread pool and returns a object that represents that work. The work to execute asynchronously A task that represents the work queued to execute in the ThreadPool. The parameter was .

(|>): 'T1 -> ('T1 -> 'U) -> 'U

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

Microsoft.FSharp.Collections.ListModule

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.

iter: ('T -> unit) -> 'T list -> unit

Applies the given function to each element of the collection. The function to apply to elements from the input list. The input list. let inputs = [ "a"; "b"; "c" ] inputs |> List.iter (printfn "%s") Evaluates to unit and prints a b c in the console. This is an O(n) operation, where n is the length of the list.

i: int

Value: int -> bool

Offers a value; false means it was dropped or the stream has ended.

ignore: 'T -> unit

Ignore the passed value. This is often used to throw away results of a computation. The value to ignore. ignore 55555 // Evaluates to ()

End: unit -> unit

Ends normally.

ok: 'value -> Flow<'env,'error,'value>

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

System.IDisposable

Provides a mechanism for releasing unmanaged resources.

_: IDisposable

Dispose: unit -> unit

Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.

runCollect: FlowStream<'a,'b,'c> -> Flow<'a,'b,'c list>

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

run: 'env -> Flow<'env,'error,'value> -> Exit<'value,'error>

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

A producer that emits synchronously, before the consumer starts pulling, makes overflow deterministic. With the default Dropping policy, the newest values are dropped once the buffer is full; emit.Value returns false for each one:

Show page setup

// Setup for the checked examples on this page.
open System
open Axial
/// Fails the docs test when an example's result differs from the value shown.
let shouldEqual expected actual =
    if actual <> expected then failwithf "Expected %A but got %A" expected actual

Verified

let droppedFlags = ref []

let overflowing : FlowStream<unit, string, int> =
    FlowStream.fromCallback 3 (fun emit ->
        Flow.delay (fun () ->
            droppedFlags.Value <- [ 1; 2; 3; 4; 5 ] |> List.map (fun i -> emit.Value i)
            emit.End()
            Flow.ok ({ new IDisposable with member _.Dispose() = () })))

droppedFlags: bool list ref

ref: 'T -> 'T ref

Create a mutable reference cell The value to contain in the cell. The created reference cell. let count = ref 0 // Creates a reference cell object with a mutable Value property count.Value // Evaluates to 0 count.Value <- 1 // Updates the value count.Value // Evaluates to 1

overflowing: FlowStream<unit,string,int>

Axial.FlowStream`3

unit

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

string

An abbreviation for the CLI type . Basic Types

int

An abbreviation for the CLI type . Basic Types

Axial.FlowStreamModule

fromCallback: int -> (Emit<'error,'value> -> Flow<'env,'error,IDisposable>) -> FlowStream<'env,'error,'value>

emit: Emit<string,int>

Axial.Flow

delay: (unit -> Flow<'env,'error,'value>) -> Flow<'env,'error,'value>

Value: bool list

The current value of the reference cell

(|>): 'T1 -> ('T1 -> 'U) -> 'U

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

Microsoft.FSharp.Collections.ListModule

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.

map: ('T -> 'U) -> 'T list -> 'U list

Builds a new collection whose elements are the results of applying the given function to each of the elements of the collection. The function to transform elements from the input list. The input list. The list of transformed elements. let inputs = [ "a"; "bbb"; "cc" ] inputs |> List.map (fun x -> x.Length) Evaluates to [ 1; 3; 2 ] This is an O(n) operation, where n is the length of the list.

i: int

Value: int -> bool

Offers a value; false means it was dropped or the stream has ended.

End: unit -> unit

Ends normally.

ok: 'value -> Flow<'env,'error,'value>

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

System.IDisposable

Provides a mechanism for releasing unmanaged resources.

_: IDisposable

Dispose: unit -> unit

Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.

Requires & types

Requires: open Axial; open System (packages: Axial)

val droppedFlags: bool list ref

val overflowing: FlowStream<unit,string,int>

Verified

overflowing |> FlowStream.runCollect |> Flow.run () |> shouldEqual (Exit.Success [ 1; 2; 3 ])
droppedFlags.Value |> shouldEqual [ true; true; true; false; false ]

overflowing: FlowStream<unit,string,int>

(|>): 'T1 -> ('T1 -> 'U) -> 'U

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

Axial.FlowStreamModule

runCollect: FlowStream<'a,'b,'c> -> Flow<'a,'b,'c list>

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

Axial.Flow

run: 'env -> Flow<'env,'error,'value> -> Exit<'value,'error>

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

shouldEqual: 'a -> 'a -> unit

Axial.Exit`2

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

Success

The workflow completed successfully.

droppedFlags: bool list ref

Value: bool list

The current value of the reference cell

Requires & types

Requires: open Axial (packages: Axial)

CallbackOverflow.Fail replaces the drop with a typed failure, and the values already queued are delivered before it:

Verified

let deliveredBeforeFailure = ref []

let failing : FlowStream<unit, string, int> =
    FlowStream.fromCallbackWith 3 (CallbackOverflow.Fail "buffer full") (fun emit ->
        Flow.delay (fun () ->
            [ 1; 2; 3; 4 ] |> List.iter (fun i -> emit.Value i |> ignore)
            emit.End()
            Flow.ok ({ new IDisposable with member _.Dispose() = () })))

deliveredBeforeFailure: int list ref

ref: 'T -> 'T ref

failing: FlowStream<unit,string,int>

Axial.FlowStream`3

unit

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

string

An abbreviation for the CLI type . Basic Types

int

An abbreviation for the CLI type . Basic Types

Axial.FlowStreamModule

fromCallbackWith: int -> CallbackOverflow<'error> -> (Emit<'error,'value> -> Flow<'env,'error,IDisposable>) -> FlowStream<'env,'error,'value>

Bridges a synchronous callback into a bounded, scope-owned stream with an overflow policy. On Fail overflow, queued values are delivered before the failure. Subscribe should register cleanup with Flow.scopeFinalizer for anything it attaches before a later step can fail or be interrupted.

Axial.CallbackOverflow`1

What a callback stream does when its buffer is full.

Fail

emit: Emit<string,int>

Axial.Flow

delay: (unit -> Flow<'env,'error,'value>) -> Flow<'env,'error,'value>

(|>): 'T1 -> ('T1 -> 'U) -> 'U

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

Microsoft.FSharp.Collections.ListModule

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.

iter: ('T -> unit) -> 'T list -> unit

i: int

Value: int -> bool

Offers a value; false means it was dropped or the stream has ended.

ignore: 'T -> unit

Ignore the passed value. This is often used to throw away results of a computation. The value to ignore. ignore 55555 // Evaluates to ()

End: unit -> unit

Ends normally.

ok: 'value -> Flow<'env,'error,'value>

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

System.IDisposable

Provides a mechanism for releasing unmanaged resources.

_: IDisposable

Dispose: unit -> unit

Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.

Requires & types

Requires: open Axial; open System (packages: Axial)

val deliveredBeforeFailure: int list ref

val failing: FlowStream<unit,string,int>

Verified

let tappedFailing =
    failing
    |> FlowStream.tapFlow (fun i ->
        Flow.delay (fun () ->
            deliveredBeforeFailure.Value <- deliveredBeforeFailure.Value @ [ i ]
            Flow.ok ()))
    |> FlowStream.runCollect

match tappedFailing |> Flow.run () with
| Exit.Failure(Cause.Fail "buffer full") -> ()
| other -> failwithf "Expected buffer full, got %A" other

deliveredBeforeFailure.Value |> shouldEqual [ 1; 2; 3 ]

tappedFailing: Flow<unit,string,int list>

failing: FlowStream<unit,string,int>

(|>): 'T1 -> ('T1 -> 'U) -> 'U

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

Axial.FlowStreamModule

tapFlow: ('a -> Flow<'b,'c,unit>) -> FlowStream<'b,'c,'a> -> FlowStream<'b,'c,'a>

Runs an effect for each value before emitting the original value. stream |> FlowStream.tapFlow logValue

i: int

Axial.Flow

delay: (unit -> Flow<'env,'error,'value>) -> Flow<'env,'error,'value>

deliveredBeforeFailure: int list ref

Value: int list

The current value of the reference cell

(@): 'T list -> 'T list -> 'T list

Concatenate two lists. The first list. The second list. The concatenation of the lists. let l1 = ['a'; 'b'; 'c'] let l2 = ['d'; 'e'; 'f'] l1 @ l2 // Evaluates to ['a'; 'b'; 'c'; 'd'; 'e'; 'f'] This is an O(n) operation, where n is the length of the first list.

ok: 'value -> Flow<'env,'error,'value>

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

runCollect: FlowStream<'a,'b,'c> -> Flow<'a,'b,'c list>

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

run: 'env -> Flow<'env,'error,'value> -> Exit<'value,'error>

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

Axial.Exit`2

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

Failure

The workflow failed due to a specific cause.

Axial.Cause`1

Represents the cause of a failed workflow. The type of the domain-specific failure value.

Fail

An expected domain-specific failure.

other: Exit<int list,string>

failwithf: Printf.StringFormat<'T,'Result> -> 'T

Print to a string buffer and raise an exception with the given result. Helper printers must return strings. The formatter. The formatted result. See Printf.failwithf (link: ) for examples.

shouldEqual: 'a -> 'a -> unit

Requires & types

Requires: open Axial (packages: Axial)

val tappedFailing: Flow<unit,string,int list>

Lift one effect

FlowStream.fromFlow creates a stream containing the successful result of one Flow:

Verified

> (Flow.succeed "Ada" : Flow<string>)
- |> FlowStream.fromFlow
- |> FlowStream.runCollect
- |> Flow.run ();;

val it: Exit<string list,Never> = Success ["Ada"]

Axial.Flow

succeed: 'value -> Flow<'env,'error,'value>

Same as ok. The value to wrap in a successful flow. A flow that always succeeds with the provided value. let result = Flow.succeed 42 |> Flow.run () // result = Success 42

Flow

A flow that requires no environment and cannot fail with a typed error.

string

An abbreviation for the CLI type . Basic Types

(|>): 'T1 -> ('T1 -> 'U) -> 'U

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

Axial.FlowStreamModule

fromFlow: Flow<'env,'error,'value> -> FlowStream<'env,'error,'value>

Creates a one-element stream from an effectful value. FlowStream.fromFlow (Flow.ok 42)

runCollect: FlowStream<'a,'b,'c> -> Flow<'a,'b,'c list>

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

run: 'env -> Flow<'env,'error,'value> -> Exit<'value,'error>

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

A failed Flow fails the stream before producing a value.

Unfold effectful state

FlowStream.unfoldFlow repeatedly runs an effectful state transition. Return Some(value, nextState) to emit a value and remember the state for the next pull, or return None to finish.

The value goes downstream; the state stays inside the source and tells it how to continue. A paginated source might emit a page of messages while retaining an opaque continuation token. A socket parser might emit a decoded frame while retaining unread bytes. Requiring only a next state would either emit implementation state to consumers or restrict unfolding to sources whose state happens to be their output.

Verified

> (1
-  |> FlowStream.unfoldFlow (fun number ->
-      Flow.succeed (
-          if number > 3 then None
-          else Some(number, number + 1)))
-  : FlowStream<int>)
- |> FlowStream.runCollect
- |> Flow.run ();;

val it: Exit<int list,Never> = Success [1; 2; 3]

(|>): 'T1 -> ('T1 -> 'U) -> 'U

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

Axial.FlowStreamModule

unfoldFlow: ('state -> Flow<'env,'error,('value * 'state) option>) -> 'state -> FlowStream<'env,'error,'value>

Creates a cold stream by repeatedly running an effectful state transition. Returns Some(value, nextState) or None when the stream is complete. The state used for the first pull. FlowStream.unfoldFlow (fun n -> Flow.ok (if n < 3 then Some(n, n + 1) else None)) 0

number: int

Axial.Flow

succeed: 'value -> Flow<'env,'error,'value>

Same as ok. The value to wrap in a successful flow. A flow that always succeeds with the provided value. let result = Flow.succeed 42 |> Flow.run () // result = Success 42

(>): 'T -> 'T -> bool

Structural greater-than The first parameter. The second parameter. The result of the comparison. 5 > 1 // Evaluates to true 5 > 5 // Evaluates to false (1, "a") > (1, "z") // Evaluates to false

None

The representation of "No value"

Some

The representation of "Value of type 'T" The input value. An option representing the value.

(+): ^T1 -> ^T2 -> ^T3

Overloaded addition operator The first parameter. The second parameter. The result of the operation. 2 + 2 // Evaluates to 4 "Hello " + "World" // Evaluates to "Hello World"

FlowStream

A stream with no environment requirement and no typed failure.

int

An abbreviation for the CLI type . Basic Types

runCollect: FlowStream<'a,'b,'c> -> Flow<'a,'b,'c list>

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

run: 'env -> Flow<'env,'error,'value> -> Exit<'value,'error>

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

Only one step runs per downstream pull. In this small numeric example the emitted value and next state look similar, but they are independent parts of the transition.

Build a concurrent crawler

For a source that discovers its own work — a link crawler or a paginated API — combine unfoldFlow with mapFlowPar and collect. The crawler recipe builds one end to end: a private frontier held in the unfold state, overlapping fetches, and backpressure between batches.

Constructing Streams

Bridge callbacks

Lift one effect

Unfold effectful state

Build a concurrent crawler

Overview

Docs

API Reference