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

Resources and Results

Own a resource for the life of a stream

A stream over a file, a socket, or a database cursor has to close it when consumption ends, however it ends. FlowStream.using acquires a Resource on the first pull and releases it when the consuming Flow finishes: when the stream is exhausted, when it fails, when it is interrupted, or when take stops it early.

Show page setup
// Setup for the checked examples on this page.
open System
open System.IO
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 readersClosed = ref 0

let openReader (path: string) : Resource<unit, Never, StreamReader> =
    Resource.create
        (Flow.delay (fun () -> Flow.ok (File.OpenText path)))
        (fun reader ->
            Flow.delay (fun () ->
                reader.Dispose()
                readersClosed.Value <- readersClosed.Value + 1
                Flow.ok ()))

let lines (path: string) : FlowStream<string> =
    FlowStream.using (openReader path) (fun reader ->
        FlowStream.repeatFlow (Flow.fromBlocking (fun _ -> reader.ReadLine()))
        |> FlowStream.takeWhile (isNull >> not))
Requires & types
Requires: open Axial; open System.IO (packages: Axial)
val readersClosed: int ref
val openReader: string -> Resource<unit,Never,StreamReader>
val lines: string -> FlowStream<string>
Verified
let path = Path.Combine(Path.GetTempPath(), $"axial-docs-lines-{Guid.NewGuid():N}.txt")
File.WriteAllLines(path, [ "first"; "second"; "third" ])

lines path |> FlowStream.runCollect |> Flow.run () |> shouldEqual (Exit.Success [ "first"; "second"; "third" ])
lines path |> FlowStream.take 1 |> FlowStream.runCollect |> Flow.run () |> shouldEqual (Exit.Success [ "first" ])
readersClosed.Value |> shouldEqual 2
File.Delete path
Requires & types
Requires: open Axial; open System; open System.IO (packages: Axial)
val path: string

The reader was closed after the full read and after take 1 stopped the stream early. This example is about using's release behavior only: it opens File.OpenText directly, so a missing file or failed read is a defect rather than a typed error. For a recommended read pattern, model the filesystem with IFileSystem and a typed FileSystemError; see FileSystem.

FlowStream.repeatFlow runs its flow once per pull, forever, so it never runs ahead of the consumer. Bound it with take or takeWhile, or pace it with throttle.

Take one result

Some consumers only need part of a stream:

  • FlowStream.runTryHead pulls a single value and returns it, or None for an empty stream. It stops the stream as soon as the value arrives, so resources and producer fibers are released right away.
  • FlowStream.runTryLast consumes the whole stream and returns its last value, or None.
  • FlowStream.runCount consumes the whole stream and returns how many values it emitted.
Verified
let firstLongLine (path: string) : Flow<string option> =
    lines path |> FlowStream.filter (fun line -> line.Length > 5) |> FlowStream.runTryHead
Requires & types
Requires: open Axial (packages: Axial)
val firstLongLine: string -> Flow<string option>
Verified
let stream = FlowStream.fromSeq [ 1; 2; 3 ]

(stream |> FlowStream.runTryHead |> Flow.run () : Exit<int option, Never>)
|> shouldEqual (Exit.Success (Some 1))

(stream |> FlowStream.runTryLast |> Flow.run () : Exit<int option, Never>)
|> shouldEqual (Exit.Success (Some 3))

(stream |> FlowStream.runCount |> Flow.run () : Exit<int64, Never>)
|> shouldEqual (Exit.Success 3L)

(FlowStream.fromSeq ([] : int list) |> FlowStream.runTryHead |> Flow.run () : Exit<int option, Never>)
|> shouldEqual (Exit.Success None)

(FlowStream.fromSeq ([] : int list) |> FlowStream.runTryLast |> Flow.run () : Exit<int option, Never>)
|> shouldEqual (Exit.Success None)

(FlowStream.fromSeq ([] : int list) |> FlowStream.runCount |> Flow.run () : Exit<int64, Never>)
|> shouldEqual (Exit.Success 0L)
Requires & types
Requires: open Axial (packages: Axial)
val stream: FlowStream<unit,Never,int>