Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .github/copilot-instructions.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
3 changes: 2 additions & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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

Expand Down
82 changes: 66 additions & 16 deletions src/FSharp.Control.R3/AsyncObservable.fs
Original file line number Diff line number Diff line change
@@ -1,3 +1,7 @@
/// <summary>
/// The Async flavour: functions that consume observable sequences as <see cref="T:Microsoft.FSharp.Control.FSharpAsync`1"/> computations
/// and asynchronous operators built on Async computations.
/// </summary>
module FSharp.Control.R3.Async

open System
Expand Down Expand Up @@ -39,36 +43,55 @@ module internal Interop =
| Choice2Of2 error -> return! failedOutcome task error
}

/// <remarks>Caution! All functions returning <see cref="Async`1"/> are blocking and may never return if awaited</remarks>
/// <summary>
/// Functions that consume an observable sequence as an <see cref="T:Microsoft.FSharp.Control.FSharpAsync`1"/>.
/// <para>
/// 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 <see cref="M:Microsoft.FSharp.Control.FSharpAsync.AwaitTask``1(System.Threading.Tasks.Task{``0})"/> they resume on the
/// synchronization context that was current when they started to wait.
/// <see cref="M:FSharp.Control.R3.Async.Observable.mapAsync``2(FSharp.Control.R3.ProcessingOptions,Microsoft.FSharp.Core.FSharpFunc{``0,Microsoft.FSharp.Control.FSharpAsync{``1}},R3.Observable{``0})"/>
/// and <see cref="M:FSharp.Control.R3.Async.Observable.ofAsync``1(Microsoft.FSharp.Control.FSharpAsync{``0})"/> are operators instead:
/// they return an observable sequence.
/// </para>
/// <para>
/// A failure of the source, including an error reported through <see cref="M:R3.Observer`1.OnErrorResume(System.Exception)"/>,
/// raises the original exception. Cancelling the computation disposes the subscription and cancels the computation.
/// </para>
/// </summary>
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!
ObservableExtensions.AggregateAsync (source, seed, f, ct)
|> 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!
ObservableExtensions.AllAsync (source, f, ct)
|> 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!
ObservableExtensions.AnyAsync (source, ct)
|> Interop.awaitTask
}

/// Returns the first element of an observable sequence
/// <summary>
/// Returns the first element of the source.
/// Raises <see cref="T:System.InvalidOperationException"/> when the source completes without an element.
/// </summary>
let firstAsync source = async {
let! ct = Async.CancellationToken
return!
Expand All @@ -77,29 +100,38 @@ module Observable =
}

/// <summary>
/// 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.
/// <para>
/// 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.
/// </para>
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
let iter (action : 't -> unit) source = async {
let! ct = Async.CancellationToken
return!
ObservableExtensions.ForEachAsync (source, action, ct)
|> 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!
ObservableExtensions.CountAsync (source, ct)
|> Interop.awaitTask
}

/// <summary>Maps the given observable with the given asynchronous function</summary>
/// <summary>
/// Projects every element with the asynchronous function, processing elements that arrive while a previous invocation
/// is running as defined by <paramref name="options"/>.
/// <para>
/// The computation runs with a cancellation token that is cancelled when the subscription is disposed,
/// when the source fails, with <see cref="P:FSharp.Control.R3.AwaitOperationConfiguration.AwaitSwitch"/> when the next element arrives,
/// or, with <see cref="P:FSharp.Control.R3.ProcessingOptions.CancelOnCompleted"/>, when the source completes.
/// An exception raised by the computation is reported through <see cref="M:R3.Observer`1.OnErrorResume(System.Exception)"/>
/// and the sequence continues.
Comment on lines +131 to +132
/// </para>
/// </summary>
/// <exception cref="T:System.ArgumentOutOfRangeException">Thrown when the concurrency limit of the options is 0 or below -1.</exception>
let mapAsync (options : ProcessingOptions) (f : 't -> Async<'r>) source =
options.Validate (nameof options)
Expand All @@ -113,20 +145,27 @@ module Observable =
options.MaxConcurrent
)

/// Creates observable sequence from a single element returned by asynchronous computation
/// <summary>
/// Creates an observable sequence that starts the computation on every subscription, emits its result and completes.
/// <para>
/// Disposing the subscription cancels the computation. A failure of the computation completes the sequence with that failure.
/// </para>
/// </summary>
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!
ObservableExtensions.ToArrayAsync (source, ct)
|> Interop.awaitTask
}

/// Collects the elements of the source into a list once it completes.
let toList source = async {
let! ct = Async.CancellationToken
let! array =
Expand Down Expand Up @@ -183,37 +222,48 @@ module Observable =
guard.ThrowIfFailed ()
}

/// <summary>
/// Overloaded functions of the Async flavour, reachable through the type name <c>Observable</c>, such as
/// <see cref="M:FSharp.Control.R3.Async.Extensions.Observable.toLookup``2(R3.Observable{``0},Microsoft.FSharp.Core.FSharpFunc{``0,``1})"/>,
/// after opening <see cref="T:FSharp.Control.R3.Async"/>.
/// </summary>
[<AutoOpen>]
module Extensions =

open System.Collections.Generic

// The overloads take no cancellation token: like every other Async function they observe the token of the computation.
// An unused generic [<Optional>] token parameter used to swallow a positional keyComparer or elementSelector.

/// Overloaded functions of the Async flavour.
[<AbstractClass; Sealed>]
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!
ObservableExtensions.ToLookupAsync (source, keySelector, ct)
|> 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!
ObservableExtensions.ToLookupAsync (source, keySelector, keyComparer = keyComparer, cancellationToken = ct)
|> 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!
ObservableExtensions.ToLookupAsync (source, keySelector, elementSelector = elementSelector, cancellationToken = ct)
|> 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 {
Expand Down
69 changes: 54 additions & 15 deletions src/FSharp.Control.R3/TaskObservable.fs
Original file line number Diff line number Diff line change
@@ -1,3 +1,7 @@
/// <summary>
/// The Task flavour: functions that consume observable sequences as <see cref="T:System.Threading.Tasks.Task`1"/> values
/// and asynchronous operators built on tasks.
Comment on lines +2 to +3
/// </summary>
module FSharp.Control.R3.Task

open System
Expand All @@ -6,39 +10,64 @@ open System.Threading.Tasks
open R3
open FSharp.Control.R3

/// <remarks>Caution! All functions returning <see cref="Task"/>/<see cref="Task`1"/> are blocking and may never return if awaited</remarks>
/// <summary>
/// Functions that consume an observable sequence as a <see cref="T:System.Threading.Tasks.Task`1"/>.
/// <para>
/// 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.
/// <see cref="M:FSharp.Control.R3.Task.Observable.mapAsync``2(FSharp.Control.R3.ProcessingOptions,Microsoft.FSharp.Core.FSharpFunc{System.Threading.CancellationToken,Microsoft.FSharp.Core.FSharpFunc{``0,System.Threading.Tasks.Task{``1}}},R3.Observable{``0})"/>
/// is an operator instead: it returns an observable sequence and subscribes only when that sequence is subscribed.
/// </para>
/// <para>
/// A failure of the source, including an error reported through <see cref="M:R3.Observer`1.OnErrorResume(System.Exception)"/>,
/// faults the task with the original exception. Cancelling the token disposes the subscription and cancels the task.
/// </para>
/// </summary>
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 ([<InlineIfLambda>] 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 ([<InlineIfLambda>] f : 'T -> bool) source = ObservableExtensions.AllAsync (source, f, cancellationToken)

/// <summary>
/// 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.
/// <para>
/// 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.
/// </para>
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
let inline iter cancellationToken ([<InlineIfLambda>] 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
/// <summary>
/// Returns the first element of the source.
/// Faults with <see cref="T:System.InvalidOperationException"/> when the source completes without an element.
/// </summary>
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)

/// <summary>Maps the given observable with the given asynchronous function</summary>
/// <summary>
/// Projects every element with the asynchronous function, processing elements that arrive while a previous invocation
/// is running as defined by <paramref name="options"/>.
/// <para>
/// The function receives a cancellation token that is cancelled when the subscription is disposed,
/// when the source fails, with <see cref="P:FSharp.Control.R3.AwaitOperationConfiguration.AwaitSwitch"/> when the next element arrives,
/// or, with <see cref="P:FSharp.Control.R3.ProcessingOptions.CancelOnCompleted"/>, when the source completes.
/// An exception of the function is reported through <see cref="M:R3.Observer`1.OnErrorResume(System.Exception)"/>
/// and the sequence continues.
Comment on lines +67 to +68
/// </para>
/// </summary>
/// <exception cref="T:System.ArgumentOutOfRangeException">Thrown when the concurrency limit of the options is 0 or below -1.</exception>
let mapAsync (options : ProcessingOptions) (f : CancellationToken -> 'T -> Task<'R>) source =
options.Validate (nameof options)
Expand Down Expand Up @@ -116,12 +145,18 @@ module Observable =
return List.ofArray array
}

/// <summary>
/// Overloaded functions of the Task flavour, reachable through the type name <c>Observable</c>, such as
/// <see cref="M:FSharp.Control.R3.Task.Extensions.Observable.ofTask``1(Microsoft.FSharp.Core.FSharpFunc{System.Threading.CancellationToken,System.Threading.Tasks.ValueTask{``0}},System.Boolean)"/>,
/// after opening <see cref="T:FSharp.Control.R3.Task"/>.
/// </summary>
[<AutoOpen>]
module Extensions =

open System.Collections.Generic
open System.Runtime.InteropServices

/// Overloaded functions of the Task flavour.
[<AbstractClass; Sealed>]
type Observable private () =

Expand Down Expand Up @@ -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, [<Optional>] 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>,
Expand All @@ -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, [<Optional>] 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>,
Expand Down