FlowModule

Flow.scopeAsyncFinalizer (fun _ -> async { resource.Close() })

let commits = Flow.fromBlocking (fun _ -> repository.Commits |> Seq.truncate 50 |> List.ofSeq)

let flow = Flow.fromTask (fun token -> client.GetStringAsync(url, token))

let result = Flow.succeed 42 |> Flow.run ()
// result = Success 42

let result = Flow.fail "error" |> Flow.run ()
// result = Failure (Cause.Fail "error")

Flow.fromResult (Ok "success") |> Flow.run ()

let sumSquares (values: int list) : Flow<unit, string, int> =
    flow {
        let mutable total = 0
        for value in values do
            do! Flow.ensureNotCanceled
            total <- total + value * value
        return total
    }

fetch |> Flow.retry (Schedule.recurs 3)

fetch |> Flow.retry (Retry.schedule { Retry.defaults with When = HttpError.isTransient })

heartbeat |> Flow.repeat (Schedule.spaced (TimeSpan.FromSeconds 5.0))

worker |> Flow.supervise (Retry.schedule { Retry.defaults with Retries = 5 })

flow {
    let! (samples: Queue<float>) = Queue.bounded 1000
    let! _ =
        samples
        |> FlowStream.fromDequeue
        |> FlowStream.runForEachFlow writeSample
        |> Flow.forkGraceful (Dequeue.shutdown samples) (TimeSpan.FromSeconds 5.0)
    do! samples |> Queue.offer 21.5 |> Flow.ignore
}
|> Flow.scoped // closing the scope flushes every offered sample before returning

let readConfig : Flow<unit, string, string> = Flow.delay (fun () -> Flow.ok "config")

let sameConfig : Flow<unit, string, bool> =
    flow {
        let! loadConfig = Flow.memoize readConfig
        let! a = loadConfig
        let! b = loadConfig // readConfig ran once
        return a = b
    }

let onQueryChanged slot query =
    search query |> Flow.forkReplacing slot |> Flow.ignore

loadPreview document |> Flow.forkReplacingKey document.Id previews

let combined = Flow.zipPar (Flow.succeed 1) (Flow.succeed 2)
combined |> Flow.run ()

let fastOrSlow = Flow.race (Flow.succeed "cached") (Flow.succeed "loaded")
fastOrSlow |> Flow.run ()

let opt = Some "value"
Flow.fromOption "missing" opt |> Flow.run ()

let result = Result.Error ()
let flow = Flow.orElseFlow (Flow.envWith (fun env -> "error")) result

let myFlow = Flow.env |> Flow.map (fun env -> env)

let currentTime () =
    Flow.envWith (fun (environment: BaseRuntime) -> environment.Clock.UtcNow())

let flow = Flow.succeed 1 |> Flow.map (fun x -> x + 1)

let flow = Flow.succeed 42 |> Flow.ignore

let flow = Flow.succeed 1 |> Flow.bind (fun x -> Flow.succeed (x + 1))

let flow = Flow.succeed 42 |> Flow.tap (fun x -> Flow.succeed ())

let flow = Flow.fail "error" |> Flow.tapError (fun err -> Flow.succeed ())

let flow = Flow.fail "error" |> Flow.mapError (fun err -> err + "!")

let ready : Flow<unit, string, int> = Flow.succeed 1 |> Flow.widenError

let flow = loadUser |> Flow.tracedError "billing.load-user"

transfer
|> Flow.onExit (fun exit ->
    match exit with
    | Exit.Success _ -> audit "transfer committed"
    | Exit.Failure _ -> audit "transfer rolled back")

render frame |> Flow.ensuring (Flow.delay (fun () -> canvas.Unlock(); Flow.ok ()))

upload |> Flow.onInterrupt (deletePartialFile path)

// Runs the worker until the service is interrupted, then lets it drain its queue.
let serve (jobs: Queue<string>) (worker: Flow<unit, Never, unit>) : Flow<unit, Never, unit> =
    flow {
        let! _ = worker |> Flow.forkGraceful (Dequeue.shutdown jobs) (TimeSpan.FromSeconds 5.0)
        return! Flow.never
    }

let flow = Flow.die (System.Exception("boom")) |> Flow.catch (fun ex -> "caught: " + ex.Message)

let flow = Flow.fail "error" |> Flow.orElseWith (fun err -> Flow.succeed "recovered")

let flow = Flow.fail "error" |> Flow.orElse (Flow.succeed "recovered")

Flow.zip (Flow.succeed 1) (Flow.succeed 2) |> Flow.run ()

let flow = Flow.map2 (fun x y -> x + y) (Flow.succeed 1) (Flow.succeed 2)

let flow = Flow.apply (Flow.succeed (fun x -> x + 1)) (Flow.succeed 1)

let flow = Flow.map3 (fun x y z -> x + y + z) (Flow.succeed 1) (Flow.succeed 2) (Flow.succeed 3)

let flow = Flow.succeed 1 |> Flow.localEnv (fun outer -> outer)

let flow = Flow.delay (fun () -> Flow.succeed 42)

let flows = [1; 2; 3] |> Flow.traverse (fun x -> Flow.succeed (x * 2))

let! pages = urls |> Flow.traversePar (Parallelism.bounded 8) fetchPage

let! matches = commits |> Flow.traverseParUsing (Parallelism.ofProcessors id) openReader (fun reader commit -> search reader commit)

do! files |> Flow.forEachPar (Parallelism.ofProcessors id) indexFile

Flow.sequence [Flow.succeed 1; Flow.succeed 2] |> Flow.run ()

let handle = workflow |> Flow.toAsync environment

let running = workflow |> Flow.startTask environment

let exit = workflow |> Flow.run environment