From c6ee6291f75c3aff549b548e2735f26fedf23854 Mon Sep 17 00:00:00 2001 From: Dag Brattli Date: Sun, 18 Oct 2026 23:43:19 +0200 Subject: [PATCH 1/2] fix: bound retry attempts per subscription --- src/AsyncObservable.fs | 2 +- src/Transform.fs | 42 +++++++++------- test/Main.fs | 1 + test/RetryTest.fs | 106 +++++++++++++++++++++++++++++++++++++++++ test/Tests.fsproj | 1 + 5 files changed, 135 insertions(+), 17 deletions(-) create mode 100644 test/RetryTest.fs diff --git a/src/AsyncObservable.fs b/src/AsyncObservable.fs index f154aca..4188532 100644 --- a/src/AsyncObservable.fs +++ b/src/AsyncObservable.fs @@ -240,7 +240,7 @@ module Reactive = let catch (handler: exn -> IAsyncObservable<'a>) (source: IAsyncObservable<'a>) : IAsyncObservable<'a> = Transform.catch handler source - /// Retries the given Observable retryCount number of times. + /// Resubscribes after an error up to retryCount times for each subscription. let retry (retryCount: int) (source: IAsyncObservable<'a>) : IAsyncObservable<'a> = Transform.retry retryCount source diff --git a/src/Transform.fs b/src/Transform.fs index 4db7c3b..9fe34b3 100644 --- a/src/Transform.fs +++ b/src/Transform.fs @@ -199,8 +199,8 @@ module internal Transform = /// /// decision: serializes fallback subscription and disposal so eager sources cannot overwrite the current subscription /// invariant: disposal acknowledges cleanup of the current subscription and rejects later notifications - let catch - (handler: exn -> IAsyncObservable<'TSource>) + let private catchWith + (handler: int -> exn -> IAsyncObservable<'TSource> option) (source: IAsyncObservable<'TSource>) : IAsyncObservable<'TSource> = let subscribeAsync (aobv: IAsyncObserver<'TSource>) = @@ -237,10 +237,15 @@ module internal Transform = let! next = async { try - let! subscription = - (handler err).SubscribeAsync(observer (generation + 1)) - - return Some subscription + match handler generation err with + | Some fallback -> + let! subscription = + fallback.SubscribeAsync(observer (generation + 1)) + + return Some subscription + | None -> + do! safeObv.OnErrorAsync err + return None with failure -> do! safeObv.OnErrorAsync failure return None @@ -273,17 +278,22 @@ module internal Transform = member _.SubscribeAsync o = subscribeAsync o } - let retry (retryCount: int) (source: IAsyncObservable<'TSource>) = - let mutable count = retryCount - - let factory exn = - match count with - | 0 -> Create.fail exn - | _ -> - count <- count - 1 - source + let catch + (handler: exn -> IAsyncObservable<'TSource>) + (source: IAsyncObservable<'TSource>) + : IAsyncObservable<'TSource> = + catchWith (fun _ err -> Some(handler err)) source - catch factory source + // decision: derives the retry budget from the catch actor's generation so BEAM never shares mutable state + // invariant: an exhausted nonnegative budget forwards the source error without subscribing again + let retry (retryCount: int) (source: IAsyncObservable<'TSource>) = + catchWith + (fun generation _ -> + if retryCount < 0 || generation < retryCount then + Some source + else + None) + source type Cmd = | Connect diff --git a/test/Main.fs b/test/Main.fs index 960b51c..6fbd168 100644 --- a/test/Main.fs +++ b/test/Main.fs @@ -32,6 +32,7 @@ let main _argv = Tests.Bind.tests Tests.Query.tests Tests.Catch.tests + Tests.Retry.tests Tests.Scan.tests Tests.Stateful.tests Tests.SubjectTest.tests diff --git a/test/RetryTest.fs b/test/RetryTest.fs new file mode 100644 index 0000000..b4531d7 --- /dev/null +++ b/test/RetryTest.fs @@ -0,0 +1,106 @@ +module Tests.Retry + +open Scriptorium.Quill +open Scriptorium.Nib.Assertion +open type Scriptorium.Quill.Test + +open Fable.Actor +open Fable.Actor.Types +open Fable.Reactive +open Tests.Utils + +let tests = + testList ( + "Retry", + [ + testAsync ( + "zero retries forwards the first error", + async { + let error = System.Exception "failed" + + let source = + fromNotification [ OnNext 1; OnError error ] + |> Reactive.retry 0 + + let observer = TestObserver() + let! _sub = source.SubscribeAsync observer + do! observer.AwaitIgnore() + assertThat (List.ofSeq observer.Notifications) (isEqualTo [ OnNext 1; OnError error ]) + } + ) + testAsync ( + "positive count allows exactly that many retries", + async { + let error = System.Exception "failed" + + let source = + fromNotification [ OnNext 1; OnError error ] + |> Reactive.retry 2 + + let observer = TestObserver() + let! _sub = source.SubscribeAsync observer + do! observer.AwaitIgnore() + + assertThat + (List.ofSeq observer.Notifications) + (isEqualTo [ OnNext 1; OnNext 1; OnNext 1; OnError error ]) + } + ) + testAsync ( + "completion after a retry stops further subscriptions", + async { + let error = System.Exception "transient" + + // decision: owns attempt count in an actor so BEAM subscriptions observe one shared count + let attempts = + Actor.spawn (fun inbox -> + let rec loop count = + actor { + let! (_, reply: ReplyChannel) = inbox.Receive() + let next = count + 1 + reply.Reply next + return! loop next + } + + loop 0) + + let source = + Reactive.create (fun observer -> + async { + let! attempt = Actor.asyncCall attempts 0 + + if attempt = 1 then + do! observer.OnNextAsync 1 + do! observer.OnErrorAsync error + else + do! observer.OnNextAsync 2 + do! observer.OnCompletedAsync() + + return AsyncDisposable.Empty + }) + |> Reactive.retry 2 + + let observer = TestObserver() + let! _sub = source.SubscribeAsync observer + do! observer.AwaitIgnore() + assertThat (List.ofSeq observer.Notifications) (isEqualTo [ OnNext 1; OnNext 2; OnCompleted ]) + } + ) + testAsync ( + "each subscriber starts with a fresh retry budget", + async { + let error = System.Exception "failed" + + let source = + fromNotification [ OnNext 1; OnError error ] + |> Reactive.retry 1 + + for _ in 1..2 do + let observer = TestObserver() + let! _sub = source.SubscribeAsync observer + do! observer.AwaitIgnore() + assertThat (List.ofSeq observer.Notifications) (isEqualTo [ OnNext 1; OnNext 1; OnError error ]) + } + ) + ] + ) diff --git a/test/Tests.fsproj b/test/Tests.fsproj index d54b0a2..661a0e2 100644 --- a/test/Tests.fsproj +++ b/test/Tests.fsproj @@ -18,6 +18,7 @@ + From 998a7b90244b705c5e45bcfe9fbc6bb847db13de Mon Sep 17 00:00:00 2001 From: Dag Brattli Date: Sun, 18 Oct 2026 23:43:26 +0200 Subject: [PATCH 2/2] docs: add Reactive API guides --- README.md | 23 ++++--- docs/actor-interop.md | 140 ++++++++++++++++++++++++++++++++++++++++ docs/aggregating.md | 60 +++++++++++++++++ docs/builder.md | 73 +++++++++++++++++++++ docs/combining.md | 44 +++++++++++++ docs/creating.md | 66 +++++++++++++++++++ docs/error-handling.md | 51 +++++++++++++++ docs/filtering.md | 55 ++++++++++++++++ docs/getting-started.md | 72 +++++++++++++++++++++ docs/time-shifting.md | 42 ++++++++++++ docs/transforming.md | 42 ++++++++++++ 11 files changed, 656 insertions(+), 12 deletions(-) create mode 100644 docs/actor-interop.md create mode 100644 docs/aggregating.md create mode 100644 docs/builder.md create mode 100644 docs/combining.md create mode 100644 docs/creating.md create mode 100644 docs/error-handling.md create mode 100644 docs/filtering.md create mode 100644 docs/getting-started.md create mode 100644 docs/time-shifting.md create mode 100644 docs/transforming.md diff --git a/README.md b/README.md index bd3e4b0..3040cca 100644 --- a/README.md +++ b/README.md @@ -45,18 +45,17 @@ All sub-namespaces and modules have been renamed accordingly: ## Documentation -Documentation is currently being updated. - -See [Interoperability with .NET Reactive Extensions](docs/INTEROPERABILITY.md) for existing -observable, observer, and subscription adapters and their asynchronous execution boundaries. - -### takeUntil ordering - -`Reactive.takeUntil notifier source` stops forwarding source values when it observes a notifier value. -The two subscriptions deliver independently, so source values still in flight may be dropped when the -notifier stops the stream. Posting source values before a notifier value, even with sequential -`do! sourceObserver.OnNextAsync ...` calls, does not guarantee that downstream has received them. -When that ordering matters, arrange downstream acknowledgement before triggering the notifier. +- [Getting started](docs/getting-started.md): subscribe, handle notifications, and dispose +- [Creating streams](docs/creating.md): values, workflows, and subjects +- [Filtering](docs/filtering.md): select, limit, and stop values +- [Transforming](docs/transforming.md): map values and switch or flatten streams +- [Combining](docs/combining.md): concatenate, merge, and pair streams +- [Aggregating](docs/aggregating.md): running state, final results, and groups +- [Error handling](docs/error-handling.md): failures, fallback streams, and retries +- [Time shifting](docs/time-shifting.md): timers, delays, debounce, and sampling +- [Reactive builder](docs/builder.md): write streams with `reactive { ... }` +- [Actor interoperability](docs/actor-interop.md): subscribe actors and build actor-backed streams +- [Interoperability with .NET Reactive Extensions](docs/INTEROPERABILITY.md): adapters and their execution boundaries ## Install diff --git a/docs/actor-interop.md b/docs/actor-interop.md new file mode 100644 index 0000000..f767a8e --- /dev/null +++ b/docs/actor-interop.md @@ -0,0 +1,140 @@ +# Actor interoperability + +Fable.Reactive works with [Fable.Actor](https://github.com/fable-hub/Fable.Actor). Open +`Fable.Reactive`, `Fable.Actor`, and `Fable.Actor.Types` for the examples below. An `Actor<'Msg>` is +the send-only reference; an actor body's `ActorInbox<'Msg>` receives messages. Share +`inbox.Reference` when another component needs to post to an actor created inside a handler. + +```fsharp +open Fable.Reactive +open Fable.Actor +open Fable.Actor.Types +``` + +## Send stream notifications to an actor + +`Reactive.subscribeActor actor source` subscribes an actor that receives `Notification<'T>` values. +It returns `Async`. Disposal unsubscribes from the source; the caller remains +responsible for the actor's lifetime. + +```fsharp +let sink = + Actor.spawn (fun inbox -> + let rec loop () = + actor { + let! notification = inbox.Receive() + + match notification with + | OnNext value -> printfn "Value: %d" value + | OnError error -> printfn "Error: %s" error.Message + | OnCompleted -> printfn "Completed" + + return! loop () + } + + loop ()) + +let subscription = + Reactive.interval 0 1000 + |> Reactive.subscribeActor sink +``` + +Run `subscription` to obtain its disposable. Keep `sink` alive while it may receive notifications; +stop it separately after its work is finished. + +## Map through a stateful actor + +`mapActor handler initialState` creates an actor for each subscription. Its handler receives the +current state and one source value, then returns `(nextState, result)`. The operator waits for the +actor's reply before forwarding that result, so actor processing applies backpressure to the +source. + +```fsharp +let runningTotals = + Reactive.ofSeq [ 10; 20; 30 ] + |> Reactive.mapActor (fun total value -> + let next = total + value + next, next) 0 + +// Emits 10, 30, 60. +``` + +The returned subscription owns this actor as well as the source subscription. Handler failure is +reported as `OnError`. + +## Emit from an actor + +`flatMapActor` creates a receiving actor for each subscription. The handler gets an `emit` callback +and an `ActorInbox` for source values. It can emit zero, one, or many results per value. + +```fsharp +let doubled = + Reactive.interval 0 1000 + |> Reactive.flatMapActor (fun emit inbox -> + let rec loop () = + actor { + let! value = inbox.Receive() + emit (value * 2) + return! loop () + } + + loop ()) +``` + +`emit` posts output without waiting for downstream processing. Source completion and error bypass +the actor, so a terminal notification can overtake actor output still in flight. Use `mapActor` +when each input needs an acknowledged, one-to-one result. An actor crash in `flatMapActor` becomes +`OnError`. + +## Choose crash behavior + +`flatMapActorSupervised decide handler` adds a decision for an actor failure: + +| Directive | Behavior | +| --- | --- | +| `Escalate` | Send the failure downstream as `OnError` | +| `Stop` | Leave the child stopped; the crashing value and later values sent to it produce no output | +| `Restart` | Replace the child; the crashing value is lost, and later values go to the replacement | + +```fsharp +let restartOnFailure = + Reactive.interval 0 1000 + |> Reactive.flatMapActorSupervised + (fun _ -> Directive.Restart) + (fun emit inbox -> + let rec loop () = + actor { + let! value = inbox.Receive() + emit (value * 2) + return! loop () + } + + loop ()) +``` + +As with `flatMapActor`, source terminal notifications bypass the child. Dispose the subscription +with `do! subscription.DisposeAsync()` to await its owned source and actor cleanup. If the handler +needs to share its receiving actor, pass `inbox.Reference` to code that expects an `Actor<'Msg>`. + +## Create an actor-backed source + +`Reactive.ofActor body` starts an actor immediately and returns its send-only reference and a hot +stream of values emitted by the body. + +```fsharp +let publisher, events = + Reactive.ofActor (fun emit inbox -> + let rec loop () = + actor { + let! value = inbox.Receive() + emit (value * 2) + return! loop () + } + + loop ()) +``` + +Subscribe to `events` before posting to `publisher`. Values emitted before subscription are not +replayed. The caller owns the actor's lifetime; disposing a stream subscription removes that +subscriber but does not stop `publisher`. See [creating streams](creating.md#publish-values-to-subscribers) +for subjects and [the reactive builder](builder.md) for composition syntax. diff --git a/docs/aggregating.md b/docs/aggregating.md new file mode 100644 index 0000000..4c8d44f --- /dev/null +++ b/docs/aggregating.md @@ -0,0 +1,60 @@ +# Aggregating + +Aggregation keeps state across values in one subscription. Choose a running result when each +intermediate state matters, or a final result when only the end matters. + +## Running results + +`scanInit initial accumulator` starts with an explicit state. It emits the updated state after +each source value; it does not emit the initial state on its own. `scan` uses the first source value +as its initial state and emits that value. Both have async variants for an accumulator returning +`Async<'State>`. + +```fsharp +open Fable.Reactive + +let totals = + Reactive.ofSeq [ 1; 2; 3 ] + |> Reactive.scanInit 0 (+) + +// Emits 1, 3, 6, then completes. +``` + +Use `scanInitAsync initial accumulator` or `scanAsync accumulator` when updating the state takes +asynchronous work. As with other stateful operators, each subscription has its own accumulated +state. + +## Final results + +`reduceAsync accumulator` emits only the final accumulated value when the source completes. It +uses the first source value as its starting state. `max` and `min` are final-result operators built +on the same reduction pattern. An empty source has no final value to emit. + +```fsharp +let largest = + Reactive.ofSeq [ 8; 3; 13; 5 ] + |> Reactive.max + +let sum = + Reactive.ofSeq [ 1; 2; 3 ] + |> Reactive.reduceAsync (fun total value -> async { return total + value }) +``` + +`takeLast count` in the [filtering guide](filtering.md) is useful when you want the last values +themselves rather than a reduced result. Both kinds of final result require source completion. + +## Group values + +`groupBy keyMapper` emits one inner stream per distinct key. Subscribe to each inner stream as the +outer stream emits it: its first value waits for that subscription. Groups receive their source +values and terminal notification, and the outer stream completes after the groups complete. + +```fsharp +let perParityTotals = + Reactive.ofSeq [ 1; 2; 3; 4 ] + |> Reactive.groupBy (fun value -> value % 2) + |> Reactive.flatMap (Reactive.scanInit 0 (+)) +``` + +The groups contain values, not a public key property. Keep the key in each value if downstream +consumers need to identify which group produced a result. diff --git a/docs/builder.md b/docs/builder.md new file mode 100644 index 0000000..fb44ce2 --- /dev/null +++ b/docs/builder.md @@ -0,0 +1,73 @@ +# The reactive builder + +Open `Fable.Reactive` to use the `reactive { ... }` computation expression. It produces an +`IAsyncObservable<'T>`, which still needs a subscription to deliver values. Use `yield` for one +value and `yield!` to include an existing stream. + +```fsharp +open Fable.Reactive + +let values = + reactive { + yield 1 + yield! Reactive.ofSeq [ 2; 3 ] + } +``` + +This stream emits `1`, `2`, and `3`, then completes. Consecutive `yield` and `yield!` clauses are +concatenated: the first stream finishes before the next begins. `reactive { () }` makes an empty +stream. The builder supports `yield`, not `return`. + +## Bind a stream or async workflow + +`let!` binds each source value to the rest of the expression. For a stream source, it uses the same +merge behavior as `Reactive.flatMap`: inner results can interleave. + +```fsharp +let sums = + reactive { + let! left = Reactive.ofSeq [ 1; 2 ] + let! right = Reactive.ofSeq [ 10; 20 ] + yield left + right + } +``` + +`sums` produces the four combinations `11`, `21`, `12`, and `22`; their arrival order can vary. +`let!` also accepts `Async<'T>` and turns its result into a one-value stream: + +```fsharp +let loaded = + reactive { + let! value = async { return 42 } + yield value * 2 + } +``` + +`yield!` can include an `Async<'T>` directly when its one result is the value to emit. + +## Iterate in order + +`for` accepts an F# sequence or an `IAsyncObservable<'T>`. It uses `concatMap`, so each iteration's +stream completes before the next iteration begins. + +```fsharp +let ordered = + reactive { + for value in [ 1; 2; 3 ] do + yield value * 10 + } +``` + +Choose `for` for ordered expansion and `let!` when merging inner streams is acceptable. See +[transforming](transforming.md#map-each-value-to-a-stream) for the corresponding operators. + +## Manage the subscription outside the builder + +The builder creates a stream; it does not subscribe to it or own its lifetime. Use +`SubscribeAsync` and await `DisposeAsync()` when you are done, as shown in +[getting started](getting-started.md#your-first-subscription). + +The ordinary `async { ... }` builder also accepts `use!` for an `IReactiveDisposable`, including a +reactive subscription. Its cleanup starts `DisposeAsync()` without awaiting completion. When +cleanup must finish before continuing, bind the disposable with `let!` and explicitly +`do! subscription.DisposeAsync()`. diff --git a/docs/combining.md b/docs/combining.md new file mode 100644 index 0000000..765eb96 --- /dev/null +++ b/docs/combining.md @@ -0,0 +1,44 @@ +# Combining streams + +Use `concat` when one stream must finish before the next begins. Use `merge` when both can run +together and value interleaving is acceptable. + +```fsharp +open Fable.Reactive + +let first = Reactive.ofSeq [ 1; 2 ] +let second = Reactive.ofSeq [ 3; 4 ] + +let ordered = first |> Reactive.concat second +let concurrent = first |> Reactive.merge second +``` + +`ordered` emits values from `first`, then `second`. `concurrent` forwards values as they arrive; +do not rely on one source winning a race. `concatSeq` and `mergeSeq` take a sequence of streams. +`mergeInner` flattens a stream of streams with unbounded active inner subscriptions. `startWith` +prepends ordinary values before subscribing to the source. + +## Pair values + +| API | When it emits | +| --- | --- | +| `combineLatest other source` | After both have produced a value, then on each new value from either stream | +| `withLatestFrom other source` | When `source` emits and `other` already has a value | +| `zipSeq items source` | Pair each source value with the next item in a sequence | + +The tuple order is `(sourceValue, otherValue)`. `combineLatest` and `withLatestFrom` currently +complete when either input completes; plan for that when combining finite and long-lived sources. + +```fsharp +let readings = Reactive.interval 0 100 +let settings = Reactive.interval 0 250 + +let latestPairs = + readings |> Reactive.combineLatest settings + +let annotated = + readings |> Reactive.zipSeq [ "first"; "second"; "third" ] +``` + +`zipSeq` completes when the source completes or when it tries to pair a source value after the +sequence has run out. For derived values, continue with [aggregating](aggregating.md). diff --git a/docs/creating.md b/docs/creating.md new file mode 100644 index 0000000..345de81 --- /dev/null +++ b/docs/creating.md @@ -0,0 +1,66 @@ +# Creating streams + +The functions below return `IAsyncObservable<'T>`. Each subscription starts its own execution for +`ofSeq` and `ofAsync`; subscribing to a stream is what starts delivery. + +| API | What a subscriber receives | +| --- | --- | +| `Reactive.single value` | One value, then completion during subscription | +| `Reactive.ofSeq values` | Values from a sequence, then completion | +| `Reactive.ofAsync workflow` | The workflow result, then completion; workflow failure sends `OnError` | +| `Reactive.empty<'T> ()` | Completion with no values | +| `Reactive.never<'T> ()` | No values or terminal notification | +| `Reactive.fail<'T> error` | `OnError` with the supplied exception | + +```fsharp +open Fable.Reactive + +let numbers = Reactive.ofSeq [ 1; 2; 3 ] + +let fetched = + Reactive.ofAsync (async { return 42 }) + |> Reactive.map string +``` + +`ofSeq` and `ofAsync` run in cancellable workers. Disposing a subscription requests cancellation; +the underlying workflow must cooperate with cancellation for it to stop promptly. + +## Create per-subscription values + +`defer` invokes its factory separately for each subscriber. Use it when stream construction should +read fresh state or acquire a new source for each subscription. + +```fsharp +let mutable nextId = 0 + +let ids = + Reactive.defer (fun () -> + nextId <- nextId + 1 + Reactive.single nextId) +``` + +`Reactive.create` accepts a subscribe function of type +`IAsyncObserver<'T> -> Async`. It is the low-level entry point when wrapping +an existing producer; the returned disposable owns that producer's cleanup. For ordinary values +or async work, prefer the constructors above. + +## Publish values to subscribers + +`Reactive.subject<'T> ()` returns an observer to publish through and a stream to subscribe to. +Subscribers see values published while they are subscribed. `Reactive.replaySubject bufferSize` +retains up to `bufferSize` recent values for new subscribers; its buffer size must be at least one. +Late subscribers also receive its terminal notification after the retained values. + +```fsharp +let publisher, events = Reactive.replaySubject 2 + +let publish = + async { + do! publisher.OnNextAsync 1 + do! publisher.OnNextAsync 2 + do! publisher.OnNextAsync 3 + do! publisher.OnCompletedAsync() + } +``` + +The separate [time-shifting guide](time-shifting.md) covers `timer` and `interval`. diff --git a/docs/error-handling.md b/docs/error-handling.md new file mode 100644 index 0000000..dcf5157 --- /dev/null +++ b/docs/error-handling.md @@ -0,0 +1,51 @@ +# Error handling + +An observer receives values through `OnNext`, then at most one terminal notification: +`OnCompleted` or `OnError`. Handle both when subscribing to a stream; disposal stops a +subscription but is not a completion notification. + +```fsharp +open Fable.Reactive + +let observe (stream: IAsyncObservable<'T>) = + stream.SubscribeAsync(fun notification -> + async { + match notification with + | OnNext value -> printfn "Value: %A" value + | OnError error -> printfn "Failed: %s" error.Message + | OnCompleted -> printfn "Finished" + }) +``` + +## Recover with a fallback stream + +`Reactive.catch handler source` forwards the source's values. If the source sends `OnError`, it +disposes the failed subscription and subscribes to the stream returned by `handler`. + +```fsharp +let recovered = + Reactive.fail (System.Exception "unavailable") + |> Reactive.catch (fun _ -> Reactive.single 0) +``` + +If the fallback fails, `catch` invokes the handler again with that error. A handler that throws is +reported as `OnError`. `Reactive.fail<'T> error` constructs a stream that reports an error when +subscribed to. + +## Retry + +`Reactive.retry count source` resubscribes after an error up to `count` times per subscription. +`count = 0` forwards the first error. For example, `retry 2` allows the initial subscription and +two retries, then forwards the last error if all three attempts fail. Successful completion stops +retrying. Use `catch` when recovery should switch to a different stream. + +```fsharp +let fetchWithRetries (fetch: unit -> Async) = + Reactive.defer (fun () -> Reactive.ofAsync (fetch ())) + |> Reactive.retry 2 +``` + +The `defer` factory calls `fetch` for each attempt, including the retries. + +See [creating streams](creating.md) for `ofAsync` and `fail`, and [getting started](getting-started.md) +for subscription ownership. diff --git a/docs/filtering.md b/docs/filtering.md new file mode 100644 index 0000000..10203fc --- /dev/null +++ b/docs/filtering.md @@ -0,0 +1,55 @@ +# Filtering + +Filtering operators keep the source value type unless they explicitly select a new type. Put the +source last so filters compose with `|>`. + +```fsharp +open Fable.Reactive + +let evenValues = + Reactive.ofSeq [ 1; 2; 3; 4; 5 ] + |> Reactive.filter (fun value -> value % 2 = 0) + |> Reactive.take 2 +``` + +| API | Behavior | +| --- | --- | +| `filter predicate` | Forward values satisfying a predicate | +| `filterAsync predicate` | Await an `Async` predicate for each value | +| `choose chooser` | Forward `Some value`, discarding `None` | +| `chooseAsync chooser` | Async version of `choose` | +| `distinctUntilChanged` | Suppress adjacent equal values; an earlier value may appear again | +| `skip count` | Ignore the first `count` values | +| `take count` | Forward up to `count` values, then complete | +| `takeLast count` | Buffer the final `count` values and emit them when the source completes | + +For example, `Reactive.choose int.TryParse` is not directly suitable because `TryParse` uses an +out parameter. Return an option instead: + +```fsharp +let parsed = + Reactive.ofSeq [ "10"; "bad"; "20" ] + |> Reactive.choose (fun text -> + match System.Int32.TryParse text with + | true, value -> Some value + | false, _ -> None) +``` + +## Stop when another stream emits + +`Reactive.takeUntil notifier source` completes the result when the notifier sends a value. +Notifier completion alone does not stop the source; notifier error is forwarded. It subscribes to +the notifier first, so an immediate stop signal is observed. + +```fsharp +let stopped = + Reactive.interval 0 100 + |> Reactive.takeUntil (Reactive.timer 1000) +``` + +Source and notifier deliver independently. A source value still in flight when the notifier is +observed can be dropped. Even awaiting a source publication before publishing to the notifier does +not establish downstream delivery order across the two streams. If that order matters, wait for a +downstream acknowledgement before sending the stop signal. + +Continue with [transforming](transforming.md) or [aggregating](aggregating.md). diff --git a/docs/getting-started.md b/docs/getting-started.md new file mode 100644 index 0000000..6d0f255 --- /dev/null +++ b/docs/getting-started.md @@ -0,0 +1,72 @@ +# Getting started + +Fable.Reactive composes asynchronous streams with `IAsyncObservable<'T>`. A stream sends values to an +`IAsyncObserver<'T>` through `OnNextAsync`, then ends with either `OnCompletedAsync` or `OnErrorAsync`. +These callbacks return `Async`, so a producer can await an observer's work. + +## Install + +Add the package to your F# project: + +```sh +dotnet paket add Fable.Reactive --project +``` + +Use Fable 5.22.0 or newer when compiling to JavaScript, Python, or BEAM. This compiler requirement +does not apply to a .NET-only application. Open `Fable.Reactive` to use the stream types and the +qualified `Reactive` module. + +## Your first subscription + +```fsharp +open System.Threading.Tasks +open Fable.Reactive + +let program = + async { + let finished = TaskCompletionSource() + + let stream = + Reactive.single 21 + |> Reactive.map (fun value -> value * 2) + + let! subscription = + stream.SubscribeAsync(fun notification -> + async { + match notification with + | OnNext value -> printfn "Value: %d" value + | OnError error -> + printfn "Error: %s" error.Message + finished.TrySetResult() |> ignore + | OnCompleted -> + printfn "Completed" + finished.TrySetResult() |> ignore + }) + + do! finished.Task |> Async.AwaitTask + do! subscription.DisposeAsync() + } + +Async.RunSynchronously program +``` + +On .NET, this prints `Value: 42`, then `Completed`. The `TaskCompletionSource` lets the host wait for +a terminal callback before disposing. Even `Reactive.single` may finish delivering callbacks after +`SubscribeAsync` returns. Keep a subscription alive until the work you need has finished. + +`SubscribeAsync` returns `Async`: run the workflow to subscribe, then await +`DisposeAsync()` to stop owning that subscription. Disposing an active subscription is distinct from +receiving `OnCompleted` or `OnError`. The `RunAsync` convenience method discards the disposable; +prefer `SubscribeAsync` when you need to control lifetime. + +Operators are curried and put the source last, so pipelines read from left to right: + +```fsharp +let values = + Reactive.ofSeq [ 1; 2; 3; 4 ] + |> Reactive.filter (fun value -> value % 2 = 0) + |> Reactive.map (fun value -> value * 10) +``` + +`values` is a stream description. Subscribe to run it. Start with [creating streams](creating.md), +then choose operators from the [guide index](../README.md#documentation). diff --git a/docs/time-shifting.md b/docs/time-shifting.md new file mode 100644 index 0000000..9e56e77 --- /dev/null +++ b/docs/time-shifting.md @@ -0,0 +1,42 @@ +# Time shifting + +The time arguments in this guide are integer milliseconds. Keep the subscription returned by +`SubscribeAsync` so an active timer or repeating source can be disposed. + +## Start on a timer + +`Reactive.timer dueTime` emits `0` once after `dueTime`, then completes. +`Reactive.interval firstDelay period` starts at `0` after `firstDelay` and increments on each +`period`. A positive period repeats; a non-positive period produces one value and completes. + +```fsharp +open Fable.Reactive + +let once = Reactive.timer 500 +let ticks = Reactive.interval 500 1000 +``` + +## Shift or reduce activity + +| API | Behavior | +| --- | --- | +| `delay msecs` | Delay each notification while preserving their order | +| `debounce msecs` | Emit a value only after no newer value arrives during the quiet period | +| `sample msecs` | At periodic ticks, emit the most recently seen source value, if any | + +```fsharp +let slower = + Reactive.interval 0 100 + |> Reactive.sample 500 + +let settled = + Reactive.interval 0 100 + |> Reactive.debounce 250 +``` + +`settled` does not emit while values keep arriving every 100 ms. For a finite burst, it emits +only a value that survives the quiet period. If the source completes or errors while a debounce +timer is pending, that pending value is dropped and the terminal notification is sent. + +`delay` also delays completion and error after earlier values, maintaining their order. For +stopping a stream at a deadline, combine `timer` with [`takeUntil`](filtering.md#stop-when-another-stream-emits). diff --git a/docs/transforming.md b/docs/transforming.md new file mode 100644 index 0000000..94d148b --- /dev/null +++ b/docs/transforming.md @@ -0,0 +1,42 @@ +# Transforming + +Use `map` for a synchronous one-to-one projection and `mapAsync` when the projection returns an +`Async<'T>`. `mapi` and `mapiAsync` also pass a zero-based index as part of a `(value, index)` tuple. + +```fsharp +open Fable.Reactive + +let labels = + Reactive.ofSeq [ 10; 20; 30 ] + |> Reactive.mapi (fun (value, index) -> sprintf "%d: %d" index value) + +let loaded = + Reactive.ofSeq [ 1; 2; 3 ] + |> Reactive.mapAsync (fun id -> async { return sprintf "item-%d" id }) +``` + +## Map each value to a stream + +When each source value produces another `IAsyncObservable`, choose how to handle those inner +streams: + +| API | Inner-stream behavior | +| --- | --- | +| `flatMap mapper` | Merge all mapped streams; values may interleave | +| `concatMap mapper` | Subscribe to mapped streams one at a time, in source order | +| `flatMapLatest mapper` | Switch to the newest mapped stream, disposing the previous one | +| `switchLatest` | Apply the same switching behavior to a stream of streams | + +```fsharp +let ordered = + Reactive.ofSeq [ 1; 2; 3 ] + |> Reactive.concatMap (fun id -> Reactive.ofAsync (async { return id * 10 })) +``` + +`flatMapAsync` and `flatMapLatestAsync` await a mapper that returns +`Async>`. `flatMapi` and `flatMapiAsync` also pass an index. Switching disposes +the old inner subscription before subscribing to the new one; completion of the outer stream waits +for the current inner stream. + +Use [combining](combining.md) to join independent sources and [aggregating](aggregating.md) for +running state.