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
1 change: 1 addition & 0 deletions .github/copilot-instructions.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
│ ├── AssemblyInfo.fs – assembly metadata
│ ├── ProcessingOptions.fs – processing configuration helpers
│ ├── Observable.fs – observable wrappers/extensions
│ ├── ObservableFactories.fs – factories reachable as `Observable.xxx` (`ofSeq`)
│ ├── AsyncObservable.fs – async observable helpers
│ └── TaskObservable.fs – task-based observable helpers
├── tests/FSharp.Control.R3.Tests/ – MSTest test project
Expand Down
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,11 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

### Added

- `Observable.ofSeq (items)` and `Observable.ofSeq (items, cancellationToken)`, reachable after `open FSharp.Control.R3`
- `rxqueryWith cancellationToken` and `RxQueryBuilder (cancellationToken)` to cancel the query operators that return a task

### Changed

- **Breaking:** `AwaitOperationConfiguration` cases are prefixed with `Await` (`AwaitSequential`, `AwaitParallel 4`, ...) so they no longer collide with `System.Threading.Tasks.Parallel`
- **Breaking:** the `Observable.Extensions` type is replaced by `Observable.ofSeq`
- **Breaking:** after `open FSharp.Control.R3`, the bare type name `Observable` refers to `FSharp.Control.R3.ObservableFactories.Observable`; write `open type R3.Observable` and extend `R3.Observable` by its full name
- **Breaking:** `rxquery` `sumBy` adds values of the projected type: it requires `(+) : 'Value * 'Value -> 'Value`
- `mapAsync` and `iterAsync` validate the concurrency limit of the options when they are called

Expand Down
1 change: 1 addition & 0 deletions src/FSharp.Control.R3/FSharp.Control.R3.fsproj
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
<Compile Include="AssemblyInfo.fs" />
<Compile Include="ProcessingOptions.fs" />
<Compile Include="Observable.fs" />
<Compile Include="ObservableFactories.fs" />
<Compile Include="AsyncObservable.fs" />
<Compile Include="TaskObservable.fs" />
</ItemGroup>
Expand Down
9 changes: 0 additions & 9 deletions src/FSharp.Control.R3/Observable.fs
Original file line number Diff line number Diff line change
Expand Up @@ -75,15 +75,6 @@ let inline take (count : int) (source) = ObservableExtensions.Take (source, coun
let inline where ([<InlineIfLambda>] f : 'T -> bool) source = ObservableExtensions.Where (source, f)

open System.Runtime.CompilerServices
open System.Runtime.InteropServices

[<AutoOpen>]
module Extensions =

[<AbstractClass; Sealed; Extension>]
type Observable private () =

static member ofSeq (items : _ seq, [<Optional>] cancellationToken) = Observable.ToObservable (items, cancellationToken)

[<AutoOpen>]
module OptionExtensions =
Expand Down
46 changes: 46 additions & 0 deletions src/FSharp.Control.R3/ObservableFactories.fs
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
namespace FSharp.Control.R3

open System.Threading
open R3

/// <summary>
/// Factories of observable sequences that are reachable through the type name <c>Observable</c>
/// as soon as <see cref="N:FSharp.Control.R3"/> is opened.
/// <para>
/// This module sits directly in the namespace, so opening the namespace opens it too.
/// Name resolution of <see cref="M:FSharp.Control.R3.ObservableFactories.Observable.ofSeq``1(System.Collections.Generic.IEnumerable{``0})"/>
/// finds no such function in the modules named <c>Observable</c> and falls through to the type
/// <see cref="T:FSharp.Control.R3.ObservableFactories.Observable"/>, the same way
/// <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)"/>
/// resolves after opening <see cref="T:FSharp.Control.R3.Task"/>. R3's own factories, such as
/// <see cref="M:R3.Observable.Range(System.Int32,System.Int32)"/>, still resolve through the same name.
/// </para>
/// <para>
/// Where the bare type name matters, it now refers to this type: write <c>open type R3.Observable</c> rather than
/// <c>open type Observable</c>, and extend <see cref="T:R3.Observable"/> by its full name.
/// </para>
/// </summary>
[<AutoOpen>]
module ObservableFactories =

/// Factories of observable sequences.
[<AbstractClass; Sealed>]
type Observable private () =

/// <summary>
/// Creates an observable sequence that emits the elements of <paramref name="items"/> synchronously on every subscription
/// and then completes.
/// <para>The sequence is enumerated to the end even when the subscriber unsubscribes earlier, so it must be finite.</para>
/// </summary>
static member ofSeq (items : 'T seq) : Observable<'T> = R3.Observable.ToObservable items

/// <summary>
/// Creates an observable sequence that emits the elements of <paramref name="items"/> synchronously on every subscription
/// and then completes.
/// <para>
/// Once <paramref name="cancellationToken"/> is cancelled, no further element is emitted and the sequence completes successfully.
/// The token is checked after the enumerator moved to the next element, so the side effects of producing that element still run.
/// </para>
/// </summary>
static member ofSeq (items : 'T seq, cancellationToken : CancellationToken) : Observable<'T> =
R3.Observable.ToObservable (items, cancellationToken)