diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index d4afe38..dab8a2b 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -15,7 +15,7 @@ ├── src/FSharp.Control.R3/ – main library │ ├── AssemblyInfo.fs – assembly metadata │ ├── ProcessingOptions.fs – processing and chunk configuration -│ ├── Observable.fs – observable wrappers/extensions +│ ├── Observable.fs – observable operators and the `rxquery` builder │ ├── ObservableOption.fs – `option` variants of the `Observable` functions (`choose`) │ ├── ObservableFactories.fs – factories reachable as `Observable.xxx` (`ofSeq`) │ ├── AsyncObservable.fs – async observable helpers diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e2d3e6..f34a9d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `rxquery` emitted its elements on the thread pool, out of order and after the source had moved on; `yield` and `zero` are now synchronous - `rxquery` `sumBy` passed `null` to the `(+)` of reference types - `chunkBy` accepted a non-positive window length of `ChunkTimeSpanCount` and `ChunkMillisecondsCount`, which failed every element +- Misleading XML docs of `catch` ## [0.3.1] - 2026-01-28 diff --git a/src/FSharp.Control.R3/Observable.fs b/src/FSharp.Control.R3/Observable.fs index c332b57..f26dbab 100644 --- a/src/FSharp.Control.R3/Observable.fs +++ b/src/FSharp.Control.R3/Observable.fs @@ -1,25 +1,56 @@ +/// +/// Curried, pipeline-friendly F# functions over the operators of , +/// and the query builder. +/// module FSharp.Control.R3.Observable open System open R3 -/// Hides the identy of an observable sequence +/// Hides the identity of the source, so that consumers cannot cast it back to its original type, such as a subject. let inline asObservable source : Observable<'Source> = ObservableExtensions.AsObservable source -/// Binds an observable to generate a subsequent observable. +/// +/// Projects every element to an observable sequence and merges the inner sequences into one. +/// +/// The inner sequences run concurrently and their elements are emitted as they arrive. +/// The result completes when the source and every inner sequence have completed. It fails when the source fails, +/// or when an inner sequence fails while the source is still running. +/// +/// let inline bind ([] f : 'T -> Observable<'TNext>) source = ObservableExtensions.SelectMany (source, f) -/// Converts the elements of the sequence to the specified type +/// +/// Casts every element to . +/// An element that cannot be cast is reported through +/// as an , or as a when +/// a element is cast to a value type, and the sequence continues. +/// let inline cast<'T, 'CastType> (source) = ObservableExtensions.Cast<'T, 'CastType>(source) /// -/// Adds an error handler to an observable sequence. +/// Continues with the observable sequence returned by when the source fails. +/// +/// Only a terminal failure, a completion with whose exception is +/// of type , is handled; other failures are forwarded unchanged. The failed source is not resumed. +/// +/// +/// Non-terminal errors reported through , such as exceptions +/// thrown by the functions passed to +/// or , +/// pass through without reaching the handler. +/// A handler whose argument type is not annotated handles every exception. +/// /// -/// Exception does not stop further processing let inline catch ([] f : 'Exn -> Observable<'T>) o = ObservableExtensions.Catch (o, f) -/// Concatenates the second observable sequence to the first observable sequence -/// upn the successful termination of the first +/// +/// Emits the elements of and, after it completes successfully, the elements of . +/// +/// A failure of terminates the result without subscribing to . +/// The parameters follow the reading order, so in a pipeline a |> Observable.concat b emits b first. +/// +/// let inline concat first second = ObservableExtensions.Concat (first, second) // R3 validates the length of Chunk(count) with the message as the parameter name and without the value, and does not validate @@ -76,36 +107,57 @@ let chunkBy (configuration : ChunkConfiguration<'T>) (source : Observable<'T>) = let inline chunkByBoundaries (windowBoundaries : Observable<'Boundary>) (source : Observable<'T>) = ObservableExtensions.Chunk (source, windowBoundaries = windowBoundaries) -/// Returns an observable sequence that only contains distinct elements +/// Emits every element that has not been emitted before, compared with the default equality comparer. let inline distinct source = ObservableExtensions.Distinct source -/// Returns an observable sequence that contains no elements +/// An observable sequence that completes successfully on subscription without emitting any element. let inline empty () = Observable.Empty () -/// Filters the observable elements of a sequence based on a predicate +/// +/// Emits the elements that satisfy the predicate. +/// An exception thrown by the predicate is reported through +/// and the sequence continues. +/// let inline filter ([] f : 'T -> bool) source = ObservableExtensions.Where (source, f) -/// Maps the given observable with the given function +/// +/// Projects every element with the function. +/// An exception thrown by the function is reported through +/// and the sequence continues. +/// let inline map ([] f : 'T -> 'R) source = ObservableExtensions.Select (source, f) -/// Maps the given observable with the given function and the index of the element -let inline mapi ([] f : int -> 'T -> 'R) source = ObservableExtensions.Select (source, (fun i x -> f x i)) +/// Projects every element with the function, which receives the zero-based index of the element first and the element second. +/// The index restarts for every subscription. +let inline mapi ([] f : int -> 'T -> 'R) source = ObservableExtensions.Select (source, (fun x i -> f i x)) -/// Merges two observable sequences into one observable sequence +/// Emits the elements of both sequences as they arrive. +/// The result completes when both sequences have completed, and fails as soon as either of them fails. let inline merge (source1, source2) = ObservableExtensions.Merge (source1, source2) +/// Emits the elements that are of type 'R, cast to it, and silently drops the others. let inline ofType<'T, 'R> (source) = ObservableExtensions.OfType<'T, 'R>(source) -/// Returns an observable sequence that contains only a single element +/// An observable sequence that emits the item synchronously on subscription and then completes. let inline singleton item = Observable.Return<'T> item -/// Bypasses a specified number of elements in an observable sequence and then returns the remaining elements +/// Bypasses the first elements and emits the remaining ones. +/// Thrown when is negative. let inline skip (count : int) (source) = ObservableExtensions.Skip (source, count) -/// Takes n elements (from the beginning of an observable sequence?) +/// +/// Emits the first elements and then completes, disposing the subscription to the source. +/// A count of 0 completes at once without subscribing to the source. +/// +/// Thrown when is negative. let inline take (count : int) (source) = ObservableExtensions.Take (source, count) -/// Filters the observable elements of a sequence based on a predicate +/// +/// Emits the elements that satisfy the predicate; the same as +/// . +/// An exception thrown by the predicate is reported through +/// and the sequence continues. +/// let inline where ([] f : 'T -> bool) source = ObservableExtensions.Where (source, f) // choose is a module function rather than a static member of an extension type: a module function here shadows