diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index 33b8855..4253cc3 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -19,8 +19,8 @@ │ ├── ObservableOption.fs – `option` variants of the `Observable` functions (`choose`) │ ├── ObservableFactories.fs – factories reachable as `Observable.xxx` (`ofSeq`) │ ├── IterationGuard.fs – internal guard that stops `iterAsync` and reports its failures -│ ├── AsyncObservable.fs – async observable helpers -│ └── TaskObservable.fs – task-based observable helpers +│ ├── AsyncObservable.fs – Async flavour (cold `Async` terminals, `mapAsync`, `ofAsync`) +│ └── TaskObservable.fs – Task flavour (hot `Task` terminals, `mapAsync`, `ofTask`) ├── tests/FSharp.Control.R3.Tests/ – MSTest test project │ ├── BuilderTests.fs – builder behavior tests │ └── ObservableTests.fs – observable behavior tests diff --git a/CHANGELOG.md b/CHANGELOG.md index 0327dc1..b7b1625 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `Observable.ofSeq (items)` and `Observable.ofSeq (items, cancellationToken)`, reachable after `open FSharp.Control.R3` - `Observable.chunkByBoundaries` accepting window boundaries of any element type - `rxqueryWith cancellationToken` and `RxQueryBuilder (cancellationToken)` to cancel the query operators that return a task +- XML documentation for every public API ### Changed @@ -36,7 +37,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `iterAsync` kept invoking the action after it failed when the source emitted synchronously, invoked it although the token was already cancelled, completed successfully when the action failed after the source completed, and never completed when the action threw an `OperationCanceledException` of its own, such as a timeout - `iterAsync` could fault with `OverflowException` on sources with more than `Int32.MaxValue` elements - `chunkBy` accepted a non-positive window length of `ChunkTimeSpanCount` and `ChunkMillisecondsCount`, which failed every element -- Misleading XML docs of `catch` and `iterAsync` +- Misleading XML docs of `catch`, `length`, `existsAsync`, `iter`, `iterAsync`, `aggregate` and the "blocking" module remarks ## [0.3.1] - 2026-01-28 diff --git a/src/FSharp.Control.R3/AsyncObservable.fs b/src/FSharp.Control.R3/AsyncObservable.fs index d119a22..d187fb6 100644 --- a/src/FSharp.Control.R3/AsyncObservable.fs +++ b/src/FSharp.Control.R3/AsyncObservable.fs @@ -1,3 +1,7 @@ +/// +/// The Async flavour: functions that consume observable sequences as computations +/// and asynchronous operators built on Async computations. +/// module FSharp.Control.R3.Async open System @@ -39,11 +43,26 @@ module internal Interop = | Choice2Of2 error -> return! failedOutcome task error } -/// Caution! All functions returning are blocking and may never return if awaited +/// +/// Functions that consume an observable sequence as an . +/// +/// The computations are cold: they subscribe when they start, so elements that a hot source emits before the start are missed, +/// and they complete when the deciding element arrives or the source terminates, which for an endless source may be never. +/// Like they resume on the +/// synchronization context that was current when they started to wait. +/// +/// and are operators instead: +/// they return an observable sequence. +/// +/// +/// A failure of the source, including an error reported through , +/// raises the original exception. Cancelling the computation disposes the subscription and cancels the computation. +/// +/// module Observable = - /// Applies an accumulator function over an observable sequence, returning the - /// result of the aggregation as a single element in the result sequence + /// Applies the accumulator to every element, starting from the seed, and returns the final accumulated value; + /// the seed when the source has no element. let aggregate seed (f : 'r -> 't -> 'r) source = async { let! ct = Async.CancellationToken return! @@ -51,7 +70,8 @@ module Observable = |> Interop.awaitTask } - /// Determines whether all elements of an observable satisfy a predicate + /// Determines whether every element satisfies the predicate; returns false at the first element that does not, + /// true when the source completes. let all (f : 't -> bool) source = async { let! ct = Async.CancellationToken return! @@ -59,8 +79,8 @@ module Observable = |> Interop.awaitTask } - /// Determines whether an observable sequence contains a specified value - /// which satisfies the given predicate + /// Determines whether the source has any element; returns true at the first element, + /// false when the source completes without one. let existsAsync source = async { let! ct = Async.CancellationToken return! @@ -68,7 +88,10 @@ module Observable = |> Interop.awaitTask } - /// Returns the first element of an observable sequence + /// + /// Returns the first element of the source. + /// Raises when the source completes without an element. + /// let firstAsync source = async { let! ct = Async.CancellationToken return! @@ -77,13 +100,12 @@ module Observable = } /// - /// Invokes an action for each element in the observable sequence, and propagates all observer - /// messages through the result sequence. + /// Subscribes to the source and invokes the action for every element. + /// + /// The computation completes when the source completes. An exception thrown by the action, or an error of the source, + /// stops the processing and is raised by the computation. + /// /// - /// - /// This method can be used for debugging, logging, etc. of query behavior - /// by intercepting the message stream to run arbitrary actions for messages on the pipeline. - /// let iter (action : 't -> unit) source = async { let! ct = Async.CancellationToken return! @@ -91,7 +113,7 @@ module Observable = |> Interop.awaitUnitTask } - /// Returns the last element of an observable sequence till its completion or cancellation + /// Returns the number of elements of the source once it completes. let length source = async { let! ct = Async.CancellationToken return! @@ -99,7 +121,17 @@ module Observable = |> Interop.awaitTask } - /// Maps the given observable with the given asynchronous function + /// + /// Projects every element with the asynchronous function, processing elements that arrive while a previous invocation + /// is running as defined by . + /// + /// The computation runs with a cancellation token that is cancelled when the subscription is disposed, + /// when the source fails, with when the next element arrives, + /// or, with , when the source completes. + /// An exception raised by the computation is reported through + /// and the sequence continues. + /// + /// /// Thrown when the concurrency limit of the options is 0 or below -1. let mapAsync (options : ProcessingOptions) (f : 't -> Async<'r>) source = options.Validate (nameof options) @@ -113,13 +145,19 @@ module Observable = options.MaxConcurrent ) - /// Creates observable sequence from a single element returned by asynchronous computation + /// + /// Creates an observable sequence that starts the computation on every subscription, emits its result and completes. + /// + /// Disposing the subscription cancels the computation. A failure of the computation completes the sequence with that failure. + /// + /// let ofAsync (computation : Async<'T>) = Observable.FromAsync (fun ct -> Async.StartImmediateAsTask (computation, cancellationToken = ct) |> ValueTask<'T> ) + /// Collects the elements of the source into an array once it completes. let toArray source = async { let! ct = Async.CancellationToken return! @@ -127,6 +165,7 @@ module Observable = |> Interop.awaitTask } + /// Collects the elements of the source into a list once it completes. let toList source = async { let! ct = Async.CancellationToken let! array = @@ -183,6 +222,11 @@ module Observable = guard.ThrowIfFailed () } +/// +/// Overloaded functions of the Async flavour, reachable through the type name Observable, such as +/// , +/// after opening . +/// [] module Extensions = @@ -190,9 +234,12 @@ module Extensions = // The overloads take no cancellation token: like every other Async function they observe the token of the computation. // An unused generic [] token parameter used to swallow a positional keyComparer or elementSelector. + + /// Overloaded functions of the Async flavour. [] type Observable private () = + /// Groups the elements of the source by key once it completes. static member toLookup (source : Observable<'T>, keySelector : 'T -> 'Key) = async { let! ct = Async.CancellationToken return! @@ -200,6 +247,7 @@ module Extensions = |> Interop.awaitTask } + /// Groups the elements of the source by key, compared with the comparer, once it completes. static member toLookup (source : Observable<'T>, keySelector : 'T -> 'Key, keyComparer : IEqualityComparer<'Key>) = async { let! ct = Async.CancellationToken return! @@ -207,6 +255,7 @@ module Extensions = |> Interop.awaitTask } + /// Groups the projected elements of the source by key once it completes. static member toLookup (source : Observable<'T>, keySelector : 'T -> 'Key, elementSelector : 'T -> 'Element) = async { let! ct = Async.CancellationToken return! @@ -214,6 +263,7 @@ module Extensions = |> Interop.awaitTask } + /// Groups the projected elements of the source by key, compared with the comparer, once it completes. static member toLookup (source : Observable<'T>, keySelector : 'T -> 'Key, elementSelector : 'T -> 'Element, keyComparer : IEqualityComparer<'Key>) = async { diff --git a/src/FSharp.Control.R3/TaskObservable.fs b/src/FSharp.Control.R3/TaskObservable.fs index 10990f7..dd608ac 100644 --- a/src/FSharp.Control.R3/TaskObservable.fs +++ b/src/FSharp.Control.R3/TaskObservable.fs @@ -1,3 +1,7 @@ +/// +/// The Task flavour: functions that consume observable sequences as values +/// and asynchronous operators built on tasks. +/// module FSharp.Control.R3.Task open System @@ -6,39 +10,64 @@ open System.Threading.Tasks open R3 open FSharp.Control.R3 -/// Caution! All functions returning / are blocking and may never return if awaited +/// +/// Functions that consume an observable sequence as a . +/// +/// The functions that return a task subscribe when they are called, so elements that a hot source emitted before the call are missed, +/// and their tasks complete when the deciding element arrives or the source terminates, which for an endless source may be never. +/// +/// is an operator instead: it returns an observable sequence and subscribes only when that sequence is subscribed. +/// +/// +/// A failure of the source, including an error reported through , +/// faults the task with the original exception. Cancelling the token disposes the subscription and cancels the task. +/// +/// module Observable = - /// Applies an accumulator function over an observable sequence, returning the - /// result of the aggregation as a single element in the result sequence + /// Applies the accumulator to every element, starting from the seed, and returns the final accumulated value; + /// the seed when the source has no element. let inline aggregate cancellationToken seed ([] f : 'R -> 'T -> 'R) source = ObservableExtensions.AggregateAsync (source, seed, f, cancellationToken) - /// Determines whether all elements of an observable satisfy a predicate + /// Determines whether every element satisfies the predicate; returns false at the first element that does not, + /// true when the source completes. let inline all cancellationToken ([] f : 'T -> bool) source = ObservableExtensions.AllAsync (source, f, cancellationToken) /// - /// Invokes an action for each element in the observable sequence, and propagates all observer - /// messages through the result sequence. + /// Subscribes to the source and invokes the action for every element. + /// + /// The task completes when the source completes. An exception thrown by the action, or an error of the source, + /// stops the processing and faults the task. + /// /// - /// - /// This method can be used for debugging, logging, etc. of query behavior - /// by intercepting the message stream to run arbitrary actions for messages on the pipeline. - /// let inline iter cancellationToken ([] action : 'T -> unit) source = ObservableExtensions.ForEachAsync (source, action, cancellationToken) - /// Determines whether an observable sequence contains a specified value - /// which satisfies the given predicate + /// Determines whether the source has any element; returns true at the first element, + /// false when the source completes without one. let inline existsAsync cancellationToken source = ObservableExtensions.AnyAsync (source, cancellationToken) - /// Returns the first element of an observable sequence + /// + /// Returns the first element of the source. + /// Faults with when the source completes without an element. + /// let inline firstAsync cancellationToken source = ObservableExtensions.FirstAsync (source, cancellationToken) - /// Returns the length of the observable sequence till its completion or cancellation + /// Returns the number of elements of the source once it completes. let length cancellationToken source = ObservableExtensions.CountAsync (source, cancellationToken) - /// Maps the given observable with the given asynchronous function + /// + /// Projects every element with the asynchronous function, processing elements that arrive while a previous invocation + /// is running as defined by . + /// + /// The function receives a cancellation token that is cancelled when the subscription is disposed, + /// when the source fails, with when the next element arrives, + /// or, with , when the source completes. + /// An exception of the function is reported through + /// and the sequence continues. + /// + /// /// Thrown when the concurrency limit of the options is 0 or below -1. let mapAsync (options : ProcessingOptions) (f : CancellationToken -> 'T -> Task<'R>) source = options.Validate (nameof options) @@ -116,12 +145,18 @@ module Observable = return List.ofArray array } +/// +/// Overloaded functions of the Task flavour, reachable through the type name Observable, such as +/// , +/// after opening . +/// [] module Extensions = open System.Collections.Generic open System.Runtime.InteropServices + /// Overloaded functions of the Task flavour. [] type Observable private () = @@ -151,9 +186,11 @@ module Extensions = = Observable.FromAsync (asyncFactory, configureAwait) + /// Groups the elements of the source by key once it completes. static member toLookup (source : Observable<'T>, keySelector : 'T -> 'Key, [] cancellationToken : CancellationToken) = ObservableExtensions.ToLookupAsync (source, keySelector, cancellationToken) + /// Groups the elements of the source by key, compared with the comparer, once it completes. static member toLookup ( source : Observable<'T>, @@ -164,11 +201,13 @@ module Extensions = = ObservableExtensions.ToLookupAsync (source, keySelector, keyComparer = keyComparer, cancellationToken = cancellationToken) + /// Groups the projected elements of the source by key once it completes. static member toLookup (source : Observable<'T>, keySelector : 'T -> 'Key, elementSelector : 'T -> 'Element, [] cancellationToken : CancellationToken) = ObservableExtensions.ToLookupAsync (source, keySelector, elementSelector = elementSelector, cancellationToken = cancellationToken) + /// Groups the projected elements of the source by key, compared with the comparer, once it completes. static member toLookup ( source : Observable<'T>,