diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md
index dbc7e09..81dec63 100644
--- a/.github/copilot-instructions.md
+++ b/.github/copilot-instructions.md
@@ -21,10 +21,17 @@
│ ├── IterationGuard.fs – internal guard that stops `iterAsync` and reports its failures
│ ├── 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
+├── tests/FSharp.Control.R3.Tests/ – MSTest integration test project
│ ├── TestCategories.fs – test category attributes for `--filter TestCategory=...`
-│ ├── BuilderTests.fs – builder behavior tests
-│ └── ObservableTests.fs – observable behavior tests
+│ ├── TestHelpers.fs – deterministic sources, recorder, probes, gated selectors
+│ ├── ProcessingOptionsTests.fs – processing options
+│ ├── ObservableTests.fs – observable operators and factories
+│ ├── ChunkTests.fs – chunking with `FakeTimeProvider`
+│ ├── BuilderTests.fs – `rxquery` builder
+│ ├── AsyncObservableTests.fs – Async flavour
+│ ├── TaskObservableTests.fs – Task flavour
+│ ├── MapAsyncTests.fs – `mapAsync` per await operation, both flavours
+│ └── IntegrationTests.fs – end-to-end scenarios across the library
├── build/ – FAKE build scripts and release automation
└── docsSrc/ – FSharp.Formatting documentation source
```
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 84ad714..129b082 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
+- Integration tests covering every public function against real R3 pipelines ([#11](https://github.com/fsprojects/FSharp.Control.R3/issues/11))
- `Observable.choose` as a function of the `Observable` module; like every function of the library that works with optional values it takes a `voption` chooser, and the `option` variant is `ObservableOption.choose`
- `Observable.ofSeq (items)` and `Observable.ofSeq (items, cancellationToken)`, reachable after `open FSharp.Control.R3`
- `Observable.chunkByBoundaries` accepting window boundaries of any element type
diff --git a/Directory.Packages.props b/Directory.Packages.props
index f9f86a2..a2d90c6 100644
--- a/Directory.Packages.props
+++ b/Directory.Packages.props
@@ -14,6 +14,7 @@
+
diff --git a/tests/FSharp.Control.R3.Tests/AsyncObservableTests.fs b/tests/FSharp.Control.R3.Tests/AsyncObservableTests.fs
new file mode 100644
index 0000000..8b5dbc1
--- /dev/null
+++ b/tests/FSharp.Control.R3.Tests/AsyncObservableTests.fs
@@ -0,0 +1,837 @@
+namespace FSharp.Control.R3.Tests
+
+open System
+open System.Collections.Concurrent
+open System.Linq
+open System.Threading
+open System.Threading.Tasks
+open Microsoft.VisualStudio.TestTools.UnitTesting
+open R3
+open FSharp.Control.R3
+open FSharp.Control.R3.Async
+open FSharp.Control.R3.Tests.TestHelpers
+
+///
+/// Integration tests of the cold Async functions of and
+/// against R3.
+///
+/// The processing modes of
+///
+/// are covered in MapAsyncTests.fs.
+///
+///
+[]
+type AsyncObservableTests (testContext : TestContext) =
+
+ // Typed as exn so that Assert.AreSame accepts it next to the InvalidOperationException that Assert.ThrowsAsync returns
+ let boom : exn = InvalidOperationException "boom"
+
+ // A synchronous source that completes without elements
+ let empty () : Observable = Sources.values [||]
+
+ // Starts the computation on the calling thread with the test token, so the test timeout cancels a computation that hangs
+ let run (computation : Async<'T>) = AsyncTest.start testContext.CancellationToken computation
+
+ // Elements that leave the result of the named terminal function undecided: firstAsync and existsAsync decide at the
+ // first element, all (x > 0) decides early only at an element that is not positive, the others wait for the completion
+ let undecidedElements (functionName : string) =
+ match functionName with
+ | "existsAsync"
+ | "firstAsync" -> [||]
+ | _ -> [| 1; 2 |]
+
+ /// Names of the terminal functions, as the data rows of the tests that cover every one of them.
+ static member TerminalFunctions : obj array seq = dataRows Terminals.names
+
+ // length
+
+ []
+ member _.``length subscribes when the Async starts and counts only the elements pushed after that`` () : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+ let lengthOfSubject = subject |> probe.Watch |> Observable.length
+
+ // Building the computation does not subscribe, and a subject drops the elements it receives without subscribers
+ subject.OnNext 1
+ Assert.AreEqual (0, probe.Subscribed, "Building the computation must not subscribe to the source")
+
+ let lengthTask = run lengthOfSubject
+ Assert.AreEqual (1, probe.Subscribed, "Starting the computation must subscribe to the source")
+ subject.OnNext 2
+ subject.OnNext 3
+ Assert.IsFalse (lengthTask.IsCompleted, "length must wait for the source to complete")
+ subject.OnCompleted (Result.Success)
+
+ let! count = lengthTask
+ Assert.AreEqual (2, count, "length must count only the elements pushed after the computation started")
+ }
+
+ []
+ member _.``length of a subject that completed before the Async started is zero`` () : Task = task {
+ use subject = new Subject ()
+ subject.OnNext 1
+ subject.OnCompleted (Result.Success)
+
+ // Subscribing to a completed subject delivers only its completion, during the subscription
+ let! count = run (Observable.length subject)
+
+ Assert.AreEqual (0, count, "length must not count the elements that a subject emitted before the computation started")
+ }
+
+ []
+ member _.``length subscribes to the source once on every run of the same Async`` () : Task = task {
+ let probe = SubscriptionProbe ()
+ let lengthOfSource =
+ Sources.values [| 1; 2; 3 |]
+ |> probe.Watch
+ |> Observable.length
+
+ let! first = run lengthOfSource
+ let! second = run lengthOfSource
+
+ Assert.AreEqual (3, first, "The first run must count every element")
+ Assert.AreEqual (3, second, "The second run must count every element again")
+ Assert.AreEqual (2, probe.Subscribed, "Every run of the computation must subscribe to the source once")
+ }
+
+ // aggregate
+
+ []
+ member _.``aggregate folds the elements in order starting from the seed`` () : Task = task {
+ let source = Sources.values [| 1; 2; 3; 4 |]
+
+ let! sum = run (source |> Observable.aggregate 0 (+))
+ let! digits =
+ run (
+ source
+ |> Observable.aggregate "" (fun digits x -> $"%s{digits}%d{x}")
+ )
+
+ Assert.AreEqual (10, sum, "aggregate must add every element to the seed")
+ Assert.AreEqual ("1234", digits, "aggregate must pass the elements to the accumulator in their order")
+ }
+
+ []
+ member _.``aggregate returns the seed for an empty source`` () : Task = task {
+ let! result = run (empty () |> Observable.aggregate 42 (+))
+ Assert.AreEqual (42, result, "aggregate must return the seed when the source has no element")
+ }
+
+ // all
+
+ []
+ member _.``all returns true for an empty source`` () : Task = task {
+ // R3 completes AllAsync with true when the source completes without an element that fails the predicate
+ let! result = run (empty () |> Observable.all (fun _ -> false))
+ Assert.IsTrue (result, "all must be true for a source without elements, whatever the predicate")
+ }
+
+ []
+ member _.``all returns true when every element satisfies the predicate`` () : Task = task {
+ let! result =
+ run (
+ Sources.values [| 1; 2; 3 |]
+ |> Observable.all (fun x -> x > 0)
+ )
+ Assert.IsTrue (result, "all must be true when every element satisfies the predicate")
+ }
+
+ []
+ member _.``all returns false at the first element that fails the predicate and unsubscribes`` () : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+ let allTask =
+ subject
+ |> probe.Watch
+ |> Observable.all (fun x -> x > 0)
+ |> run
+
+ subject.OnNext 1
+ Assert.IsFalse (allTask.IsCompleted, "all must wait while every element satisfies the predicate")
+ // R3 decides at the first element that fails the predicate and disposes its subscription without waiting for completion
+ subject.OnNext (-1)
+ Assert.AreEqual (1, probe.Disposed, "all must unsubscribe from the source as soon as the result is decided")
+
+ let! result = allTask
+ Assert.IsFalse (result, "all must be false once an element fails the predicate")
+ }
+
+ // existsAsync
+
+ []
+ member _.``existsAsync returns false for an empty source`` () : Task = task {
+ let! result = run (Observable.existsAsync (empty ()))
+ Assert.IsFalse (result, "existsAsync must be false for a source without elements")
+ }
+
+ []
+ member _.``existsAsync returns true at the first element and unsubscribes`` () : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+ let existsTask = subject |> probe.Watch |> Observable.existsAsync |> run
+
+ Assert.IsFalse (existsTask.IsCompleted, "existsAsync must wait for an element or for the completion")
+ // R3 decides AnyAsync at the first element and disposes its subscription without waiting for completion
+ subject.OnNext 5
+ Assert.AreEqual (1, probe.Disposed, "existsAsync must unsubscribe from the source as soon as an element arrives")
+
+ let! result = existsTask
+ Assert.IsTrue (result, "existsAsync must be true once an element arrives")
+ }
+
+ // firstAsync
+
+ []
+ member _.``firstAsync returns the first element and unsubscribes`` () : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+ let firstTask = subject |> probe.Watch |> Observable.firstAsync |> run
+
+ // R3 decides FirstAsync at the first element and disposes its subscription, so later elements are not observed
+ subject.OnNext 7
+ Assert.AreEqual (1, probe.Disposed, "firstAsync must unsubscribe from the source at the first element")
+ subject.OnNext 8
+
+ let! first = firstTask
+ Assert.AreEqual (7, first, "firstAsync must return the first element")
+ }
+
+ []
+ member _.``firstAsync of an empty source raises InvalidOperationException rather than an AggregateException`` () : Task = task {
+ // R3 faults FirstAsync with InvalidOperationException for a source without elements; the Async flavour raises that
+ // exception itself, where Async.AwaitTask would raise the AggregateException of the faulted task
+ do!
+ Assert.ThrowsExactlyAsync(
+ (fun () -> run (Observable.firstAsync (empty ())) :> Task),
+ "firstAsync must raise exactly InvalidOperationException for a source without elements"
+ )
+ :> Task
+ }
+
+ []
+ member _.``firstAsync of an empty source raises an exception that a typed handler in an async workflow catches`` () : Task = task {
+ let firstOrFallback = async {
+ try
+ let! first = Observable.firstAsync (empty ())
+ return $"first element %d{first}"
+ with :? InvalidOperationException ->
+ return "handled"
+ }
+
+ let! outcome = run firstOrFallback
+
+ // A handler for InvalidOperationException only matches when the computation raises that exception unwrapped
+ Assert.AreEqual ("handled", outcome, "The typed handler must catch the exception that firstAsync raises")
+ }
+
+ // iter
+
+ []
+ member _.``iter invokes the action for every element in order`` () : Task = task {
+ let seen = ResizeArray()
+
+ do! run (Sources.values [| 1; 2; 3 |] |> Observable.iter seen.Add)
+
+ CollectionAssert.AreEqual ([| 1; 2; 3 |], seen.ToArray (), "iter must invoke the action for every element in order")
+ }
+
+ []
+ member _.``iter completes without invoking the action for an empty source`` () : Task = task {
+ let seen = ResizeArray()
+
+ do! run (empty () |> Observable.iter seen.Add)
+
+ Assert.IsEmpty (seen, "iter must not invoke the action when the source has no element")
+ }
+
+ []
+ member _.``iter raises the exception of the action as the same instance and stops processing`` () : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+ let invoked = ResizeArray()
+
+ let action x =
+ invoked.Add x
+ if x = 2 then
+ raise boom
+
+ let iterTask = subject |> probe.Watch |> Observable.iter action |> run
+ subject.OnNext 1
+ // R3 turns the exception of the action into a fault of ForEachAsync, which disposes its subscription at once
+ subject.OnNext 2
+ Assert.AreEqual (1, probe.Disposed, "iter must unsubscribe from the source when the action fails")
+ subject.OnNext 3
+
+ let! error =
+ Assert.ThrowsAsync((fun () -> iterTask :> Task), "iter must raise the exception of the action")
+ Assert.AreSame (boom, error, "iter must raise the exception of the action itself, not a wrapper")
+ CollectionAssert.AreEqual ([| 1; 2 |], invoked.ToArray (), "The action must not be invoked after it failed")
+ }
+
+ // toArray and toList
+
+ []
+ member _.``toArray collects every element in order`` () : Task = task {
+ let! elements = run (Observable.toArray (Sources.values [| 1; 2; 3 |]))
+ CollectionAssert.AreEqual ([| 1; 2; 3 |], elements, "toArray must collect every element in order")
+ }
+
+ []
+ member _.``toList collects every element in order`` () : Task = task {
+ let! elements = run (Observable.toList (Sources.values [| 1; 2; 3 |]))
+ Assert.AreEqual([ 1; 2; 3 ], elements, "toList must collect every element in order")
+ }
+
+ []
+ member _.``toArray and toList return empty collections for an empty source`` () : Task = task {
+ let! collectedArray = run (Observable.toArray (empty ()))
+ let! collectedList = run (Observable.toList (empty ()))
+
+ Assert.IsEmpty (collectedArray, "toArray must return an empty array for a source without elements")
+ Assert.IsEmpty (collectedList, "toList must return an empty list for a source without elements")
+ }
+
+ // ofAsync
+
+ []
+ member _.``ofAsync emits the result of an immediate computation once and completes synchronously`` () =
+ // FromAsync awaits the task of the computation, which an immediate computation has already completed,
+ // so the result and the completion arrive during the subscription
+ use recorder = Observable.ofAsync (async { return 7 }) |> Recorder.Attach
+
+ CollectionAssert.AreEqual ([| 7 |], recorder.Values, "ofAsync must emit the result of the computation once")
+ Assert.IsEmpty (recorder.Errors, "ofAsync must not report errors for a successful computation")
+ Assert.IsTrue (recorder.IsCompletedSuccessfully, "ofAsync must complete successfully during the subscription")
+
+ []
+ member _.``ofAsync starts the computation once per subscription`` () =
+ let runs = ref 0
+
+ let numberOfRun =
+ Observable.ofAsync (
+ async {
+ runs.Value <- runs.Value + 1
+ return runs.Value
+ }
+ )
+
+ Assert.AreEqual (0, runs.Value, "Creating the sequence must not start the computation")
+ use first = Recorder.Attach numberOfRun
+ use second = Recorder.Attach numberOfRun
+
+ Assert.AreEqual (2, runs.Value, "Every subscription must start the computation once")
+ CollectionAssert.AreEqual ([| 1 |], first.Values, "The first subscription must receive the result of the first run")
+ CollectionAssert.AreEqual ([| 2 |], second.Values, "The second subscription must receive the result of its own run")
+
+ []
+ member _.``ofAsync completes with the failure of the computation as the same exception`` () =
+ // FromAsync turns a fault of the factory task into OnCompleted(Failure), and awaiting the task of the computation
+ // rethrows the exception that the computation raised
+ use recorder =
+ Observable.ofAsync (async { return (raise boom : int) })
+ |> Recorder.Attach
+
+ Assert.IsEmpty (recorder.Values, "A failed computation must not emit a value")
+ Assert.IsEmpty (recorder.Errors, "The failure must not be reported through OnErrorResume")
+ assertFailedWith boom recorder "ofAsync must complete with the exception of the computation itself"
+
+ []
+ member _.``disposing an ofAsync subscription cancels the computation`` () =
+ let token = ref CancellationToken.None
+ let resume = ref (fun (_ : int) -> ())
+
+ let computation = async {
+ let! cancellationToken = Async.CancellationToken
+ token.Value <- cancellationToken
+ // Suspends until the test resumes it, like a computation that does not observe its token
+ return! Async.FromContinuations (fun (onSuccess, _, _) -> resume.Value <- onSuccess)
+ }
+
+ use recorder = Observable.ofAsync computation |> Recorder.Attach
+ Assert.IsTrue (token.Value.CanBeCanceled, "Subscribing must start the computation with the token of the subscription")
+ Assert.IsFalse (token.Value.IsCancellationRequested, "The token must not be cancelled while the subscription is alive")
+ Assert.IsEmpty (recorder.Values, "Nothing may be emitted while the computation is running")
+ Assert.IsTrue (recorder.Completion.IsNone, "The sequence must not complete while the computation is running")
+
+ // The subscription of FromAsync is a CancellationDisposable whose token the computation runs with
+ recorder.Dispose ()
+ Assert.IsTrue (token.Value.IsCancellationRequested, "Disposing the subscription must cancel the token of the computation")
+
+ // R3's Observer ignores every notification once it is disposed, so a late result can never reach a subscriber;
+ // resuming only runs the abandoned continuation and nothing is asserted about it
+ let onSuccess = resume.Value
+ onSuccess 42
+
+ // iterAsync
+
+ []
+ member _.``iterAsync applies the action to every element in order and completes with the source`` () : Task = task {
+ let seen = ConcurrentQueue()
+
+ do!
+ Sources.values [| 1; 2; 3 |]
+ |> Observable.iterAsync ProcessingOptions.Default (fun x -> async { seen.Enqueue x })
+ |> run
+
+ // The default options run one action at a time in the order of the elements
+ CollectionAssert.AreEqual ([| 1; 2; 3 |], seen.ToArray (), "iterAsync must apply the action to every element in order")
+ }
+
+ []
+ member _.``iterAsync completes without invoking the action for an empty source`` () : Task = task {
+ let seen = ConcurrentQueue()
+
+ do!
+ empty ()
+ |> Observable.iterAsync ProcessingOptions.Default (fun x -> async { seen.Enqueue x })
+ |> run
+
+ Assert.IsEmpty (seen, "iterAsync must not invoke the action when the source has no element")
+ }
+
+ []
+ member _.``iterAsync raises a terminal failure of the source as the same exception`` () : Task = task {
+ // A failure of the source makes SelectAwait cancel its running actions and publish the failure at once,
+ // which faults the ForEachAsync of iterAsync with that exception
+ let iterTask =
+ Sources.failingAfter [| 1; 2 |] boom
+ |> Observable.iterAsync ProcessingOptions.Default (fun _ -> async { return () })
+ |> run
+
+ let! error =
+ Assert.ThrowsAsync((fun () -> iterTask :> Task), "iterAsync must raise the failure of the source")
+ Assert.AreSame (boom, error, "iterAsync must raise the exception of the source itself, not a wrapper")
+ }
+
+ []
+ member _.``iterAsync raises the exception of the action as the same instance and stops processing`` () : Task = task {
+ use subject = new Subject ()
+ let invoked = ConcurrentQueue()
+
+ let action x = async {
+ invoked.Enqueue x
+ if x = 2 then
+ raise boom
+ }
+
+ let iterTask =
+ subject
+ |> Observable.iterAsync ProcessingOptions.Default action
+ |> run
+ subject.OnNext 1
+ subject.OnNext 2
+ subject.OnNext 3
+
+ // SelectAwait reports the exception of the action through OnErrorResume, which faults the ForEachAsync of iterAsync
+ let! error =
+ Assert.ThrowsAsync((fun () -> iterTask :> Task), "iterAsync must raise the exception of the action")
+ Assert.AreSame (boom, error, "iterAsync must raise the exception of the action itself, not a wrapper")
+ // The fault disposes the mapped sequence, whose sequential worker checks its cancelled token before the next element
+ CollectionAssert.AreEqual ([| 1; 2 |], invoked.ToArray (), "The action must not be invoked after it failed")
+ }
+
+ []
+ member _.``iterAsync stops invoking the action after it failed even when a synchronous source emits during the subscription`` () : Task = task {
+ let invoked = ConcurrentQueue()
+
+ let action x = async {
+ invoked.Enqueue x
+ if x = 2 then
+ raise boom
+ }
+
+ // The source emits every element while iterAsync subscribes, before R3 could dispose the mapped sequence,
+ // so it is the library that skips the elements after the failure
+ let iterTask =
+ Sources.values [| 1; 2; 3; 4; 5 |]
+ |> Observable.iterAsync ProcessingOptions.Default action
+ |> run
+
+ let! error =
+ Assert.ThrowsAsync((fun () -> iterTask :> Task), "iterAsync must raise the exception of the action")
+ Assert.AreSame (boom, error, "iterAsync must raise the exception of the action itself, not a wrapper")
+ CollectionAssert.AreEqual ([| 1; 2 |], invoked.ToArray (), "The action must not be invoked after it failed")
+ }
+
+ []
+ member _.``iterAsync raises a failure of the action that happens after the source completed`` () : Task = task {
+ let gate = TaskCompletionSource (TaskCreationOptions.RunContinuationsAsynchronously)
+
+ let action _ = async {
+ do! Async.AwaitTask gate.Task
+ raise boom
+ }
+
+ let iterTask =
+ Sources.values [| 1 |]
+ |> Observable.iterAsync ProcessingOptions.Default action
+ |> run
+
+ // The source has completed while the action still runs
+ Assert.IsFalse (iterTask.IsCompleted, "iterAsync must wait for the running action")
+ // R3 drops an error that the mapped sequence reports after its source completed and would complete the iteration
+ // successfully; the library raises the failure it recorded when the action failed
+ gate.SetResult ()
+
+ let! error =
+ Assert.ThrowsAsync((fun () -> iterTask :> Task), "iterAsync must raise the late failure of the action")
+ Assert.AreSame (boom, error, "iterAsync must raise the exception of the action itself")
+ }
+
+ []
+ member _.``iterAsync completes successfully when an action superseded by AwaitSwitch raises a cancellation`` () : Task = task {
+ use subject = new Subject ()
+ // Without RunContinuationsAsynchronously, releasing the first gate resumes its action inline, so the action has raised
+ // its cancellation by the time SetResult returns. Its token is already cancelled, so FSharp.Core cancels the superseded
+ // computation without running the try/with handler of iterAsync, and R3 drops the cancelled invocation
+ let firstGate = TaskCompletionSource ()
+ let lastGate = TaskCompletionSource (TaskCreationOptions.RunContinuationsAsynchronously)
+ let options = { ProcessingOptions.Default with AwaitOperationConfiguration = AwaitSwitch }
+
+ let action x = async {
+ let! cancellationToken = Async.CancellationToken
+ if x = 1 then
+ do! Async.AwaitTask firstGate.Task
+ // A well-behaved action that notices its cancellation after a wait that does not observe the token
+ cancellationToken.ThrowIfCancellationRequested ()
+ else
+ do! Async.AwaitTask lastGate.Task
+ }
+
+ let iterTask = subject |> Observable.iterAsync options action |> run
+ subject.OnNext 1
+ // AwaitSwitch cancels the token of the running first action when the next element arrives
+ subject.OnNext 2
+ firstGate.SetResult ()
+ subject.OnCompleted (Result.Success)
+ lastGate.SetResult ()
+
+ // R3 swallows the cancellation of a superseded action, and the library must not count it as a failure either
+ do! iterTask
+ Assert.IsTrue (iterTask.IsCompletedSuccessfully, "iterAsync must complete successfully when only a superseded action was cancelled")
+ }
+
+ []
+ member _.``iterAsync stops at an OperationCanceledException that the action raises by itself instead of hanging`` () : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+ let invoked = ConcurrentQueue()
+ // Like the TaskCanceledException of an HttpClient timeout: raised while the computation of the action is not cancelled
+ let timeout : exn = TaskCanceledException "timeout"
+
+ let action x = async {
+ invoked.Enqueue x
+ if x = 2 then
+ raise timeout
+ }
+
+ let iterTask =
+ subject
+ |> probe.Watch
+ |> Observable.iterAsync ProcessingOptions.Default action
+ |> run
+ subject.OnNext 1
+ subject.OnNext 2
+
+ // R3 swallows an OperationCanceledException of the selector, which stops its sequential worker for good without completing;
+ // the library records it as the failure of the iteration instead
+ let! error =
+ Assert.ThrowsAsync((fun () -> iterTask :> Task), "iterAsync must raise the exception of the action")
+ Assert.AreSame (timeout, error, "iterAsync must raise the exception of the action itself")
+ subject.OnNext 3
+ CollectionAssert.AreEqual ([| 1; 2 |], invoked.ToArray (), "The action must not be invoked after it failed")
+ // R3 completes the task of the iteration before it disposes the subscription, so the disposal is awaited, not assumed
+ do! probe.WaitForDisposedAsync (testContext.CancellationToken, 1)
+ Assert.AreEqual (0, probe.Active, "The failure of the action must unsubscribe from the source")
+ }
+
+ []
+ member _.``iterAsync started with an already cancelled token is cancelled without invoking the action`` () : Task = task {
+ let invoked = ConcurrentQueue()
+ let probe = SubscriptionProbe ()
+ use cancellation = new CancellationTokenSource ()
+ cancellation.Cancel ()
+
+ // FSharp.Core stops a cancelled computation at its first bind, the let! that binds the token of the iteration,
+ // before iterAsync subscribes. The guard of iterAsync would skip every action of a subscribed iteration anyway,
+ // so only the probe shows that the source is left alone
+ let iterTask =
+ Sources.values [| 1; 2; 3 |]
+ |> probe.Watch
+ |> Observable.iterAsync ProcessingOptions.Default (fun x -> async { invoked.Enqueue x })
+ |> AsyncTest.start cancellation.Token
+
+ let! _ =
+ Assert.ThrowsAsync((fun () -> iterTask :> Task), "A cancelled iterAsync must raise a cancellation")
+ Assert.IsTrue (iterTask.IsCanceled, "The task of the cancelled iterAsync must be cancelled, not faulted")
+ Assert.IsEmpty (invoked, "The action must not be invoked with an already cancelled token")
+ Assert.AreEqual (0, probe.Subscribed, "A computation started with an already cancelled token must not subscribe to the source")
+ }
+
+ []
+ member _.``cancelling iterAsync cancels the token of the running action and unsubscribes from the source`` () : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+ // The gate of the action is never released: the cancellation has to stop the computation on its own
+ let action = GatedSelector ignore
+ use cancellation = CancellationTokenSource.CreateLinkedTokenSource testContext.CancellationToken
+
+ // Parallel starts an action inside OnNext, so an element that still reached the iteration after the cancellation
+ // would be in Started at once; the sequential worker would stay parked on the unreleased action 1 instead
+ let iterTask =
+ subject
+ |> probe.Watch
+ |> Observable.iterAsync ProcessingOptions.Parallel action.InvokeAsync
+ |> AsyncTest.start cancellation.Token
+
+ subject.OnNext 1
+ do! action.WaitForStartedAsync (testContext.CancellationToken, 1)
+ Assert.IsFalse (action.Tokens[0].IsCancellationRequested, "The running action must not be cancelled before the computation is")
+
+ // R3 disposes the subscription of ForEachAsync inside Cancel, and disposing the mapped sequence cancels the token
+ // that SelectAwait passed to the running action as the token of its computation
+ cancellation.Cancel ()
+ Assert.AreEqual (1, probe.Disposed, "Cancelling must unsubscribe from the source before Cancel returns")
+ Assert.IsTrue (action.Tokens[0].IsCancellationRequested, "Cancelling must cancel the token of the running action")
+ subject.OnNext 2
+ CollectionAssert.AreEqual ([| 1 |], action.Started, "No action may start after the cancellation")
+
+ let! _ =
+ Assert.ThrowsAsync((fun () -> iterTask :> Task), "Awaiting a cancelled iterAsync must raise a cancellation")
+ Assert.IsTrue (iterTask.IsCanceled, "The task of a cancelled iterAsync must be cancelled, not faulted")
+ }
+
+ []
+ member _.``iterAsync throws ArgumentOutOfRangeException when called with a concurrency limit of zero`` () =
+ let options = {
+ ProcessingOptions.Default with
+ AwaitOperationConfiguration = AwaitOperationConfiguration.AwaitParallel 0
+ }
+
+ // R3 would only reject the limit when the mapped sequence is subscribed; the library validates it when called
+ assertArgumentRejected
+ (Observable.iterAsync options (fun _ -> async { return () }))
+ "options"
+ 0
+ "iterAsync must validate the options when it is called, before the computation starts"
+
+ // mapAsync
+
+ []
+ member _.``mapAsync with the default options projects every element in order`` () : Task = task {
+ use recorder =
+ Sources.values [| 1; 2; 3 |]
+ |> Observable.mapAsync ProcessingOptions.Default (fun x -> async { return x * 10 })
+ |> Recorder.Attach
+
+ // The sequential worker of SelectAwait completes the mapped sequence once it has drained its queue,
+ // which is not guaranteed to happen during the subscription
+ let! completion = recorder.WaitForCompletionAsync testContext.CancellationToken
+
+ Assert.IsTrue (completion.IsSuccess, "The mapped sequence must complete successfully with the source")
+ CollectionAssert.AreEqual ([| 10; 20; 30 |], recorder.Values, "mapAsync must emit the projections in the order of the elements")
+ Assert.IsEmpty (recorder.Errors, "No error may be reported")
+ }
+
+ // toLookup
+
+ []
+ member _.``toLookup groups the elements by key`` () : Task = task {
+ let! lookup = run (Observable.toLookup (Fruits.source (), Fruits.initial))
+
+ // The default comparer of string keys is case-sensitive, so every initial is a key of its own
+ Assert.HasCount (5, lookup, "Every distinct initial must be a key")
+ CollectionAssert.AreEqual ([| "apple" |], lookup["a"] |> Seq.toArray, "The group of a key must hold its elements")
+ CollectionAssert.AreEqual ([| "Avocado" |], lookup["A"] |> Seq.toArray, "Keys that differ in case must be separate groups")
+ }
+
+ []
+ member _.``toLookup with a positional keyComparer groups the keys with the comparer`` () : Task = task {
+ // Regression: a positional third argument used to bind to an unused cancellationToken parameter, which dropped the comparer
+ let! positional =
+ run (Observable.toLookup (Fruits.source (), Fruits.initial, StringComparer.OrdinalIgnoreCase))
+ let! named =
+ run (Observable.toLookup (Fruits.source (), Fruits.initial, keyComparer = StringComparer.OrdinalIgnoreCase))
+
+ Assert.HasCount (3, positional, "A positional keyComparer must merge the initials that differ only in case")
+ CollectionAssert.AreEqual ([| "apple"; "Avocado" |], positional["a"] |> Seq.toArray, "A merged group must hold its elements in arrival order")
+ CollectionAssert.AreEqual (
+ [| "banana"; "Blueberry" |],
+ positional["B"] |> Seq.toArray,
+ "The comparer must also find a key that differs in case"
+ )
+ Assert.HasCount (3, named, "A named keyComparer must group the keys the same way")
+ CollectionAssert.AreEqual ([| "apple"; "Avocado" |], named["A"] |> Seq.toArray, "A named keyComparer must merge the same groups")
+ }
+
+ []
+ member _.``toLookup with a positional elementSelector projects the grouped elements`` () : Task = task {
+ // Regression: a positional elementSelector used to bind to an unused cancellationToken parameter, so the result was a
+ // lookup of the unprojected strings; the annotated element types are part of the assertion
+ let! (positional : ILookup) =
+ run (Observable.toLookup (Fruits.source (), Fruits.initial, Fruits.nameLength))
+ let! (named : ILookup) =
+ run (Observable.toLookup (Fruits.source (), Fruits.initial, elementSelector = Fruits.nameLength))
+
+ Assert.HasCount (5, positional, "Without a comparer every distinct initial must be a key")
+ CollectionAssert.AreEqual ([| 6 |], positional["b"] |> Seq.toArray, "A positional elementSelector must project the grouped elements")
+ CollectionAssert.AreEqual ([| 9 |], positional["B"] |> Seq.toArray, "Every group must hold the projected elements")
+ Assert.HasCount (5, named, "A named elementSelector must group the keys the same way")
+ CollectionAssert.AreEqual ([| 6 |], named["b"] |> Seq.toArray, "A named elementSelector must project the grouped elements")
+ }
+
+ []
+ member _.``toLookup with an elementSelector and a keyComparer projects the elements and groups the keys with the comparer`` () : Task = task {
+ let! positional =
+ run (Observable.toLookup (Fruits.source (), Fruits.initial, Fruits.nameLength, StringComparer.OrdinalIgnoreCase))
+ let! named =
+ run (
+ Observable.toLookup (
+ Fruits.source (),
+ Fruits.initial,
+ elementSelector = Fruits.nameLength,
+ keyComparer = StringComparer.OrdinalIgnoreCase
+ )
+ )
+
+ Assert.HasCount (3, positional, "The keyComparer must merge the initials that differ only in case")
+ CollectionAssert.AreEqual ([| 6; 9 |], positional["B"] |> Seq.toArray, "The merged group must hold the projected elements in arrival order")
+ Assert.HasCount (3, named, "Named arguments must group the keys the same way")
+ CollectionAssert.AreEqual ([| 5; 7 |], named["a"] |> Seq.toArray, "Named arguments must project and merge the same groups")
+ }
+
+ []
+ member _.``toLookup returns an empty lookup for an empty source`` () : Task = task {
+ let! lookup = run (Observable.toLookup (empty (), fun x -> x % 2))
+
+ Assert.IsEmpty (lookup, "A source without elements must give a lookup without keys")
+ Assert.IsFalse (lookup.Contains 0, "The empty lookup must not contain any key")
+ // The lookup of R3 returns an empty sequence for a missing key
+ Assert.IsEmpty (lookup[0], "A missing key must map to an empty group")
+ }
+
+ // Every terminal function
+
+ []
+ member _.``a terminal function raises a terminal failure of the source as the same exception`` (functionName : string) : Task = task {
+ let source = Sources.failingAfter (undecidedElements functionName) boom
+
+ // R3 faults every terminal task with the exception of OnCompleted(Failure) itself, and the Async flavour raises the
+ // single inner exception of the faulted task rather than the AggregateException that Async.AwaitTask raises
+ let! error =
+ Assert.ThrowsAsync(
+ (fun () -> run (Terminals.runAsync functionName source) :> Task),
+ $"%s{functionName} must raise the failure of the source"
+ )
+
+ Assert.AreSame (boom, error, $"%s{functionName} must raise the exception of the source itself, not a wrapper")
+ }
+
+ []
+ member _.``a terminal function raises an OnErrorResume error as the same exception and unsubscribes`` (functionName : string) : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+ let terminalTask =
+ subject
+ |> probe.Watch
+ |> Terminals.runAsync functionName
+ |> run
+
+ for element in undecidedElements functionName do
+ subject.OnNext element
+
+ Assert.IsFalse (terminalTask.IsCompleted, $"%s{functionName} must wait while the source neither fails nor completes")
+ // R3 terminal operators treat OnErrorResume as fatal: they fault with that exception and dispose their subscription
+ subject.OnErrorResume boom
+ Assert.AreEqual (1, probe.Disposed, $"%s{functionName} must unsubscribe from the source when the source reports an error")
+
+ let! error =
+ Assert.ThrowsAsync(
+ (fun () -> terminalTask :> Task),
+ $"%s{functionName} must raise the error that the source reported"
+ )
+
+ Assert.AreSame (boom, error, $"%s{functionName} must raise the reported exception itself, not a wrapper")
+ }
+
+ []
+ member _.``cancelling the Async token cancels a terminal function and unsubscribes from the source`` (functionName : string) : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+ use cancellation = CancellationTokenSource.CreateLinkedTokenSource testContext.CancellationToken
+ let terminalTask =
+ subject
+ |> probe.Watch
+ |> Terminals.runAsync functionName
+ |> AsyncTest.start cancellation.Token
+
+ for element in undecidedElements functionName do
+ subject.OnNext element
+
+ Assert.AreEqual (1, probe.Active, $"%s{functionName} must stay subscribed while its result is undecided")
+ // R3 registers on the token of the computation and disposes its subscription inside Cancel
+ cancellation.Cancel ()
+ Assert.AreEqual (1, probe.Disposed, $"%s{functionName} must unsubscribe from the source as soon as the token is cancelled")
+
+ let! _ =
+ Assert.ThrowsAsync(
+ (fun () -> terminalTask :> Task),
+ $"Awaiting a cancelled %s{functionName} must raise a cancellation"
+ )
+
+ // The cancelled R3 task cancels the computation instead of failing it with a TaskCanceledException
+ Assert.IsTrue (terminalTask.IsCanceled, $"The task of a cancelled %s{functionName} must be cancelled, not faulted")
+ }
+
+ []
+ member _.``a terminal function started with a cancelled token is cancelled without keeping a subscription`` (functionName : string) : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+ use cancellation = CancellationTokenSource.CreateLinkedTokenSource testContext.CancellationToken
+ cancellation.Cancel ()
+
+ let terminalTask =
+ subject
+ |> probe.Watch
+ |> Terminals.runAsync functionName
+ |> AsyncTest.start cancellation.Token
+
+ // FSharp.Core stops a cancelled computation at its first bind, before the R3 operator subscribes, and R3 would
+ // dispose a subscription made with a cancelled token at once, so no subscription survives either way
+ Assert.AreEqual (0, probe.Active, $"%s{functionName} must not keep a subscription when started with a cancelled token")
+
+ let! _ =
+ Assert.ThrowsAsync(
+ (fun () -> terminalTask :> Task),
+ $"Awaiting %s{functionName} started with a cancelled token must raise a cancellation"
+ )
+
+ Assert.IsTrue (terminalTask.IsCanceled, $"The task of %s{functionName} started with a cancelled token must be cancelled")
+ }
+
+ []
+ member _.``a terminal function resumes on the synchronization context that was current when it started to wait`` (functionName : string) : Task =
+ task {
+ use subject = new Subject ()
+ let context = RecordingSynchronizationContext ()
+ let resumed = TaskCompletionSource (TaskCreationOptions.RunContinuationsAsynchronously)
+
+ // The computation subscribes and starts to wait while the context is current, like a UI handler
+ context.Run (fun () ->
+ Async.StartImmediate (
+ async {
+ do! Terminals.runAsync functionName subject
+ resumed.SetResult ()
+ },
+ testContext.CancellationToken
+ )
+ )
+
+ // The source completes on the test thread, which has no context: like Async.AwaitTask, the terminal function must post
+ // the rest of the computation to the captured context instead of running it on the thread that completed the source
+ subject.OnNext 1
+ subject.OnCompleted (Result.Success)
+ do! resumed.Task.WaitAsync testContext.CancellationToken
+
+ Assert.IsTrue (context.Posts >= 1, $"%s{functionName} must resume through the synchronization context it started to wait on")
+ }
diff --git a/tests/FSharp.Control.R3.Tests/BuilderTests.fs b/tests/FSharp.Control.R3.Tests/BuilderTests.fs
index 53ec263..d070165 100644
--- a/tests/FSharp.Control.R3.Tests/BuilderTests.fs
+++ b/tests/FSharp.Control.R3.Tests/BuilderTests.fs
@@ -1,53 +1,962 @@
namespace FSharp.Control.R3.Tests
open System
+open System.Threading
open System.Threading.Tasks
-open FSharp.Control.R3.Async
-open FSharp.Control.R3.Observable.Builders
open Microsoft.VisualStudio.TestTools.UnitTesting
-open Swensen.Unquote
+open R3
+open FSharp.Control.R3.Observable.Builders
+open FSharp.Control.R3.Task
+open FSharp.Control.R3.Tests.TestHelpers
+
+// Public, because F# solves the (+) constraint of the inline sumBy only with public members
+/// Types that the builder tests aggregate.
+[]
+module BuilderTestTypes =
+
+ ///
+ /// An amount of money: a reference type whose addition reads both operands,
+ /// so adding an amount to a seed throws a .
+ ///
+ type Money = {
+ Cents : int64
+ } with
+
+ /// Adds two amounts.
+ static member (+) (left : Money, right : Money) = { Cents = left.Cents + right.Cents }
+///
+/// Integration tests of the query expressions over R3.
+///
+/// and
+/// emit synchronously, so a query over
+/// synchronous sources keeps the order of the elements and completes during the subscription: the tests assert right after
+/// building or collecting such a query, without waiting.
+///
+///
[]
-type BuilderTests () =
+type BuilderTests (testContext : TestContext) =
+
+ // Collects a query over synchronous sources. Such a query completes during the subscription, so the task is already
+ // completed here; a faulted task rethrows its original exception when the test awaits it.
+ let collect (query : Observable<'T>) =
+ let valuesTask = query |> Observable.toArray testContext.CancellationToken
+ Assert.IsTrue (valuesTask.IsCompleted, "A query over synchronous sources must complete during the subscription")
+ valuesTask
+
+ []
+ member _.``rxquery for and select project every element in order`` () : Task = task {
+ let query = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ select (x * 10)
+ }
+
+ let! values = collect query
+ CollectionAssert.AreEqual ([| 10; 20; 30; 40; 50; 60 |], values, "select must project every element in the order of the source")
+ }
+
+ []
+ member _.``rxquery over a synchronous source completes synchronously`` () : Task = task {
+ let countTask = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ count
+ }
+
+ // Yield is Observable.Return, which emits during the subscription; when it was scheduled on TimeProvider.System,
+ // every element hopped to the thread pool and the task completed only later
+ Assert.IsTrue (countTask.IsCompletedSuccessfully, "count over a synchronous source must complete before the query returns")
+ let! elementCount = countTask
+ Assert.AreEqual (6, elementCount, "count must count every element of the source")
+ }
+
+ []
+ member _.``rxquery where keeps only the elements that satisfy the predicate`` () : Task = task {
+ let query = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ where (x % 2 = 0)
+ select x
+ }
+
+ let! values = collect query
+ CollectionAssert.AreEqual ([| 2; 4; 6 |], values, "where must keep the even elements in order")
+ }
+
+ []
+ member _.``rxquery if-then yield skips the other elements through Zero`` () : Task = task {
+ let query = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ if x % 2 = 0 then
+ yield x
+ }
+
+ let! values = collect query
+ CollectionAssert.AreEqual ([| 2; 4; 6 |], values, "An if-then without else must yield the even elements and skip the odd ones")
+ }
+
+ []
+ member _.``rxquery nested for produces the cartesian product in order`` () : Task = task {
+ let query = rxquery {
+ for x in Sources.values [| 1; 2 |] do
+ for y in Sources.values [| "a"; "b" |] do
+ yield struct (x, y)
+ }
+
+ // Every inner query emits and completes during its subscription, so SelectMany emits it whole before the next outer element
+ let! pairs = collect query
+ CollectionAssert.AreEqual (
+ [| struct (1, "a"); struct (1, "b"); struct (2, "a"); struct (2, "b") |],
+ pairs,
+ "Nested for must pair every outer element with every inner element, outer element first"
+ )
+ }
+
+ []
+ member _.``rxquery takeWhile completes at the first element that fails the predicate`` () : Task = task {
+ let query = rxquery {
+ for x in Sources.values [| 1; 2; 3; 4; 1 |] do
+ takeWhile (x < 4)
+ }
+
+ // The trailing 1 satisfies the predicate again, but TakeWhile has completed at 4
+ let! values = collect query
+ CollectionAssert.AreEqual ([| 1; 2; 3 |], values, "takeWhile must stop at the first element that fails the predicate")
+ }
+
+ []
+ member _.``rxquery take emits the first elements then completes and unsubscribes from a hot source`` () : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+
+ let valuesTask =
+ rxquery {
+ for x in probe.Watch subject do
+ take 2
+ }
+ |> Observable.toArray testContext.CancellationToken
+
+ subject.OnNext 1
+ Assert.IsFalse (valuesTask.IsCompleted, "take must wait for its second element")
+ subject.OnNext 2
+ // Take completes at its last element and the completed terminal disposes the query down to the subject;
+ // the subject never completes by itself, so the probe proves the unsubscription
+ Assert.IsTrue (valuesTask.IsCompletedSuccessfully, "take must complete at its second element")
+ Assert.AreEqual (1, probe.Disposed, "Completing take must dispose the subscription to the subject")
+ let! values = valuesTask.WaitAsync testContext.CancellationToken
+ CollectionAssert.AreEqual ([| 1; 2 |], values, "take must emit the first two elements in order")
+ }
+
+ []
+ member _.``rxquery skipWhile bypasses only the leading elements that satisfy the predicate`` () : Task = task {
+ let query = rxquery {
+ for x in Sources.values [| 1; 2; 3; 1 |] do
+ skipWhile (x < 3)
+ }
+
+ let! values = collect query
+ CollectionAssert.AreEqual ([| 3; 1 |], values, "skipWhile must emit every element from the first one that fails the predicate")
+ }
+
+ []
+ member _.``rxquery skip bypasses the first elements`` () : Task = task {
+ let query = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ skip 4
+ }
+
+ let! values = collect query
+ CollectionAssert.AreEqual ([| 5; 6 |], values, "skip must bypass the first four elements")
+ }
[]
- member _.``Test builder rxquery`` () =
+ member _.``rxquery distinct removes repeated elements and keeps the first occurrences in order`` () : Task = task {
+ let query = rxquery {
+ for x in Sources.values [| 1; 2; 1; 3; 2 |] do
+ distinct
+ }
+
+ let! values = collect query
+ CollectionAssert.AreEqual ([| 1; 2; 3 |], values, "distinct must emit every element once, at its first occurrence")
+ }
+
+ []
+ member _.``rxquery count counts the elements that pass the query`` () : Task = task {
+ let matchingTask = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ where (x > 2)
+ count
+ }
+
+ let emptyTask = rxquery {
+ for x in Sources.values Array.empty do
+ count
+ }
+
+ let! matching = matchingTask.WaitAsync testContext.CancellationToken
+ let! emptyCount = emptyTask.WaitAsync testContext.CancellationToken
+ Assert.AreEqual (4, matching, "count must count the elements that pass where")
+ Assert.AreEqual (0, emptyCount, "count over an empty source must be 0")
+ }
+
+ []
+ member _.``rxquery all returns false at the first failing element and unsubscribes from a hot source`` () : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+
+ let allTask = rxquery {
+ for x in probe.Watch subject do
+ all (x < 3)
+ }
+
+ subject.OnNext 1
+ subject.OnNext 2
+ Assert.IsFalse (allTask.IsCompleted, "all must wait while every element satisfies the predicate")
+ subject.OnNext 3
+ // R3 AllAsync decides at the first failing element and disposes its subscription, which unsubscribes from the subject;
+ // the subject never completes by itself, so the probe proves the short-circuit
+ Assert.IsTrue (allTask.IsCompletedSuccessfully, "all must complete at the first failing element, before the source completes")
+ Assert.AreEqual (1, probe.Disposed, "all must dispose the subscription to the source at the first failing element")
+ let! result = allTask.WaitAsync testContext.CancellationToken
+ Assert.IsFalse (result, "all must be false when an element fails the predicate")
+ }
+
+ []
+ member _.``rxquery all returns true when every element satisfies the predicate or the source is empty`` () : Task = task {
+ let everyTask = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ all (x > 0)
+ }
+
+ let emptyTask = rxquery {
+ for x in Sources.values Array.empty do
+ all (x > 0)
+ }
+
+ let! every = everyTask.WaitAsync testContext.CancellationToken
+ let! overEmpty = emptyTask.WaitAsync testContext.CancellationToken
+ Assert.IsTrue (every, "all must be true when every element satisfies the predicate")
+ // R3 AllAsync completes with true when the source completes without a failing element
+ Assert.IsTrue (overEmpty, "all over an empty source must be true")
+ }
+
+ []
+ member _.``rxquery contains finds an element and returns false otherwise`` () : Task = task {
+ let presentTask = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ contains 3
+ }
+
+ let missingTask = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ contains 42
+ }
- let mutable hasvisited = false
- use r3Bus = new R3.Subject ()
+ let emptyTask = rxquery {
+ for x in Sources.values Array.empty do
+ contains 3
+ }
+
+ let! present = presentTask.WaitAsync testContext.CancellationToken
+ let! missing = missingTask.WaitAsync testContext.CancellationToken
+ let! inEmpty = emptyTask.WaitAsync testContext.CancellationToken
+ Assert.IsTrue (present, "contains must find an element of the source")
+ Assert.IsFalse (missing, "contains must be false for an element that is not in the source")
+ Assert.IsFalse (inEmpty, "contains over an empty source must be false")
+ }
+
+ []
+ member _.``rxquery exactlyOne returns the only element and fails for an empty source`` () : Task = task {
+ let singleTask = rxquery {
+ for x in Sources.values [| 7 |] do
+ exactlyOne
+ }
+
+ let emptyTask = rxquery {
+ for x in Sources.values Array.empty do
+ exactlyOne
+ }
+
+ // R3 SingleAsync fails with an InvalidOperationException when the source completes without an element
+ let! _ =
+ Assert.ThrowsExactlyAsync(
+ (fun () -> emptyTask.WaitAsync testContext.CancellationToken :> Task),
+ "exactlyOne over an empty source must fail"
+ )
+ let! single = singleTask.WaitAsync testContext.CancellationToken
+ Assert.AreEqual (7, single, "exactlyOne must return the only element")
+ }
+
+ []
+ member _.``rxquery exactlyOne fails at the second element of a hot source without waiting for completion`` () : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+
+ let singleTask = rxquery {
+ for x in probe.Watch subject do
+ exactlyOne
+ }
+
+ subject.OnNext 7
+ Assert.IsFalse (singleTask.IsCompleted, "exactlyOne must wait for the source to complete after the first element")
+ subject.OnNext 8
+ // R3 SingleAsync fails as soon as a second element arrives and disposes its subscription;
+ // the subject never completes by itself, so the probe proves the unsubscription
+ Assert.IsTrue (singleTask.IsFaulted, "exactlyOne must fail at the second element, before the source completes")
+ Assert.AreEqual (1, probe.Disposed, "exactlyOne must dispose the subscription to the source at the second element")
+ let! _ =
+ Assert.ThrowsExactlyAsync(
+ (fun () -> singleTask.WaitAsync testContext.CancellationToken :> Task),
+ "exactlyOne over two elements must fail"
+ )
+ ()
+ }
+
+ []
+ member _.``rxquery exactlyOneOrDefault returns the default only for an empty source`` () : Task = task {
+ let singleTask = rxquery {
+ for x in Sources.values [| 7 |] do
+ exactlyOneOrDefault
+ }
+
+ let emptyTask = rxquery {
+ for x in Sources.values Array.empty do
+ exactlyOneOrDefault
+ }
+
+ let manyTask = rxquery {
+ for x in Sources.values [| 1; 2; 3 |] do
+ exactlyOneOrDefault
+ }
+
+ // R3 SingleOrDefaultAsync applies the default value to an empty source only; a second element still fails
+ let! _ =
+ Assert.ThrowsExactlyAsync(
+ (fun () -> manyTask.WaitAsync testContext.CancellationToken :> Task),
+ "exactlyOneOrDefault over several elements must fail"
+ )
+
+ let! single = singleTask.WaitAsync testContext.CancellationToken
+ let! orDefault = emptyTask.WaitAsync testContext.CancellationToken
+ Assert.AreEqual (7, single, "exactlyOneOrDefault must return the only element")
+ Assert.AreEqual (0, orDefault, "exactlyOneOrDefault over an empty source must return the default value")
+ }
+
+ []
+ member _.``rxquery find returns the first matching element and fails when none matches`` () : Task = task {
+ let foundTask = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ find (x > 2)
+ }
+
+ let missingTask = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ find (x > 9)
+ }
+
+ // R3 FirstAsync with a predicate fails with an InvalidOperationException when the source completes without a match
+ let! _ =
+ Assert.ThrowsExactlyAsync(
+ (fun () -> missingTask.WaitAsync testContext.CancellationToken :> Task),
+ "find without a matching element must fail"
+ )
+ let! found = foundTask.WaitAsync testContext.CancellationToken
+ Assert.AreEqual (3, found, "find must return the first element that satisfies the predicate")
+ }
+
+ []
+ member _.``rxquery head returns the first element and fails for an empty source`` () : Task = task {
+ let headTask = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ head
+ }
+
+ let emptyTask = rxquery {
+ for x in Sources.values Array.empty do
+ head
+ }
+
+ // R3 FirstAsync fails with an InvalidOperationException when the source completes without an element
+ let! _ =
+ Assert.ThrowsExactlyAsync(
+ (fun () -> emptyTask.WaitAsync testContext.CancellationToken :> Task),
+ "head over an empty source must fail"
+ )
+ let! first = headTask.WaitAsync testContext.CancellationToken
+ Assert.AreEqual (1, first, "head must return the first element")
+ }
+
+ []
+ member _.``rxquery head completes at the first element of a hot source and unsubscribes`` () : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+
+ let headTask = rxquery {
+ for x in probe.Watch subject do
+ head
+ }
+
+ Assert.IsFalse (headTask.IsCompleted, "head must wait for the first element")
+ subject.OnNext 5
+ // R3 FirstAsync completes at the first element and disposes its subscription;
+ // the subject never completes by itself, so the probe proves the unsubscription
+ Assert.IsTrue (headTask.IsCompletedSuccessfully, "head must complete at the first element, before the source completes")
+ Assert.AreEqual (1, probe.Disposed, "head must dispose the subscription to the source at the first element")
+ let! first = headTask.WaitAsync testContext.CancellationToken
+ Assert.AreEqual (5, first, "head must return the first element pushed after the subscription")
+ }
+
+ []
+ member _.``rxquery headOrDefault returns the first element or the default for an empty source`` () : Task = task {
+ let headTask = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ headOrDefault
+ }
+
+ let emptyTask = rxquery {
+ for x in Sources.values Array.empty do
+ headOrDefault
+ }
+
+ let! first = headTask.WaitAsync testContext.CancellationToken
+ let! orDefault = emptyTask.WaitAsync testContext.CancellationToken
+ Assert.AreEqual (1, first, "headOrDefault must return the first element")
+ Assert.AreEqual (0, orDefault, "headOrDefault over an empty source must return the default value")
+ }
+
+ []
+ member _.``rxquery last returns the last element and fails for an empty source`` () : Task = task {
+ let lastTask = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ last
+ }
+
+ let emptyTask = rxquery {
+ for x in Sources.values Array.empty do
+ last
+ }
+
+ // R3 LastAsync fails with an InvalidOperationException when the source completes without an element
+ let! _ =
+ Assert.ThrowsExactlyAsync(
+ (fun () -> emptyTask.WaitAsync testContext.CancellationToken :> Task),
+ "last over an empty source must fail"
+ )
+ let! final = lastTask.WaitAsync testContext.CancellationToken
+ Assert.AreEqual (6, final, "last must return the last element")
+ }
+
+ []
+ member _.``rxquery lastOrDefault returns the last element or the default for an empty source`` () : Task = task {
+ let lastTask = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ lastOrDefault
+ }
+
+ let emptyTask = rxquery {
+ for x in Sources.values Array.empty do
+ lastOrDefault
+ }
+
+ let! final = lastTask.WaitAsync testContext.CancellationToken
+ let! orDefault = emptyTask.WaitAsync testContext.CancellationToken
+ Assert.AreEqual (6, final, "lastOrDefault must return the last element")
+ Assert.AreEqual (0, orDefault, "lastOrDefault over an empty source must return the default value")
+ }
+
+ []
+ member _.``rxquery maxBy and minBy return the first element with the extreme projected value`` () : Task = task {
+ // "ccc" and "eee" share the largest length, "a" and "f" the smallest one
+ let words = Sources.values [| "bb"; "a"; "ccc"; "dd"; "eee"; "f" |]
- let interesting = rxquery {
- for i in r3Bus do
+ let longestTask = rxquery {
+ for word in words do
+ maxBy word.Length
+ }
+
+ let shortestTask = rxquery {
+ for word in words do
+ minBy word.Length
+ }
+
+ let! longest = longestTask.WaitAsync testContext.CancellationToken
+ let! shortest = shortestTask.WaitAsync testContext.CancellationToken
+ // R3 MaxByAsync and MinByAsync replace the current element only on a strictly larger or smaller key, so the first one wins ties
+ Assert.AreEqual ("ccc", longest, "maxBy must return the first of the longest words")
+ Assert.AreEqual ("a", shortest, "minBy must return the first of the shortest words")
+ }
+
+ []
+ member _.``rxquery maxBy and minBy fail for an empty source`` () : Task = task {
+ let words = Sources.values Array.empty
+
+ let longestTask = rxquery {
+ for word in words do
+ maxBy word.Length
+ }
+
+ let shortestTask = rxquery {
+ for word in words do
+ minBy word.Length
+ }
+
+ // R3 MaxByAsync and MinByAsync fail with an InvalidOperationException when the source has no element
+ let! _ =
+ Assert.ThrowsExactlyAsync(
+ (fun () -> longestTask.WaitAsync testContext.CancellationToken :> Task),
+ "maxBy over an empty source must fail"
+ )
+ let! _ =
+ Assert.ThrowsExactlyAsync(
+ (fun () -> shortestTask.WaitAsync testContext.CancellationToken :> Task),
+ "minBy over an empty source must fail"
+ )
+ ()
+ }
+
+ []
+ member _.``rxquery sumBy adds integer float and TimeSpan values`` () : Task = task {
+ let numbers = Sources.values [| 1; 2; 3; 4 |]
+
+ let integersTask = rxquery {
+ for x in numbers do
+ sumBy x
+ }
+
+ let halvesTask = rxquery {
+ for x in numbers do
+ sumBy (float x / 2.0)
+ }
+
+ let durationsTask = rxquery {
+ for x in numbers do
+ sumBy (TimeSpan.FromSeconds (float x))
+ }
+
+ let! integers = integersTask.WaitAsync testContext.CancellationToken
+ let! halves = halvesTask.WaitAsync testContext.CancellationToken
+ let! durations = durationsTask.WaitAsync testContext.CancellationToken
+ Assert.AreEqual (10, integers, "sumBy must add the integers")
+ Assert.AreEqual (5.0, halves, "sumBy must add the floats")
+ // TimeSpan has an addition operator but no Zero member, which the first-element seed does not need
+ Assert.AreEqual (TimeSpan.FromSeconds 10.0, durations, "sumBy must add the TimeSpan values")
+ }
+
+ []
+ member _.``rxquery sumBy returns the default value for an empty source`` () : Task = task {
+ let integersTask = rxquery {
+ for x in Sources.values Array.empty do
+ sumBy x
+ }
+
+ let floatsTask = rxquery {
+ for x in Sources.values Array.empty do
+ sumBy x
+ }
+
+ let durationsTask = rxquery {
+ for x in Sources.values Array.empty do
+ sumBy x
+ }
+
+ let! integers = integersTask.WaitAsync testContext.CancellationToken
+ let! floats = floatsTask.WaitAsync testContext.CancellationToken
+ let! durations = durationsTask.WaitAsync testContext.CancellationToken
+ Assert.AreEqual (0, integers, "sumBy of no integer must be 0")
+ Assert.AreEqual (0.0, floats, "sumBy of no float must be 0.0")
+ Assert.AreEqual (TimeSpan.Zero, durations, "sumBy of no TimeSpan must be TimeSpan.Zero")
+ }
+
+ []
+ member _.``rxquery sumBy adds a record type that defines the addition operator`` () : Task = task {
+ let amounts = Sources.values [| { Cents = 150L }; { Cents = 250L }; { Cents = 100L } |]
+
+ let totalTask = rxquery {
+ for amount in amounts do
+ sumBy amount
+ }
+
+ // Regression: the sum was seeded with Unchecked.defaultof, a null Money, so the first addition threw a
+ // NullReferenceException and faulted the task; the first element seeds the sum now
+ let! total = totalTask.WaitAsync testContext.CancellationToken
+ Assert.AreEqual ({ Cents = 500L }, total, "sumBy must add the amounts with their addition operator")
+ }
+
+ []
+ member _.``rxquery zip pairs elements by position and completes with the shorter source`` () : Task = task {
+ let shorterQuery = rxquery {
+ for x in Sources.values [| 1; 2; 3 |] do
+ zip letter in Sources.values [| "a"; "b"; "c"; "d" |]
+ select (struct (x, letter))
+ }
+
+ let shorterOther = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ zip letter in Sources.values [| "a"; "b" |]
+ select (struct (x, letter))
+ }
+
+ // R3 Zip pairs the queued elements of both sources in arrival order and completes once a completed source has no
+ // queued element left, so the extra elements of the longer source are dropped and collect sees the completion
+ let! pairs = collect shorterQuery
+ let! otherPairs = collect shorterOther
+ CollectionAssert.AreEqual (
+ [| struct (1, "a"); struct (2, "b"); struct (3, "c") |],
+ pairs,
+ "zip must pair by position and complete with the shorter query"
+ )
+
+ CollectionAssert.AreEqual (
+ [| struct (1, "a"); struct (2, "b") |],
+ otherPairs,
+ "zip must pair by position and complete with the shorter other source"
+ )
+ }
+
+ []
+ member _.``rxquery zip subscribes to the query before the zipped source`` () =
+ let subscriptions = ResizeArray()
+ // Every source is synchronous, so the subscriptions are recorded on the test thread in the order they happen
+ let recordSubscription name (source : Observable<'T>) = source.Do (onSubscribe = (fun () -> subscriptions.Add name))
+
+ use recorder =
+ rxquery {
+ for x in Sources.values [| 1; 2 |] |> recordSubscription "query" do
+ zip letter in Sources.values [| "a"; "b" |] |> recordSubscription "zipped"
+ select (struct (x, letter))
+ }
+ |> Recorder.Attach
+
+ // R3 Zip subscribes its first source before its second one, so this order shows that the builder passes the query
+ // as the first source; the pairs cannot show it, because Zip queues the elements of each source and pairs them by
+ // position whichever source emits first
+ CollectionAssert.AreEqual ([| "query"; "zipped" |], subscriptions.ToArray (), "zip must subscribe to the query before the zipped source")
+ CollectionAssert.AreEqual ([| struct (1, "a"); struct (2, "b") |], recorder.Values, "zip must pair the elements by position")
+
+ []
+ member _.``rxquery zip fails with the failure of the query when both sources fail on subscription`` () =
+ let queryFailure : exn = InvalidOperationException "query"
+ let zippedFailure : exn = InvalidOperationException "zipped"
+
+ use recorder =
+ rxquery {
+ for x in Sources.failingAfter Array.empty queryFailure do
+ zip letter in Sources.failingAfter Array.empty zippedFailure
+ select (struct (x, letter))
+ }
+ |> Recorder.Attach
+
+ // R3 Zip subscribes the query first and fails at the first failure of either source; the zipped source, subscribed
+ // after that, fails an observer that Zip has already disposed
+ assertFailedWith queryFailure recorder "zip must fail with the failure of the query, which it subscribes first"
+
+ []
+ member _.``rxquery iter runs the action for every element in order`` () : Task = task {
+ let seen = ResizeArray()
+ let untouched = ResizeArray()
+
+ let iterTask = rxquery {
+ for x in Sources.values [| 1..6 |] do
+ iter (seen.Add x)
+ }
+
+ let emptyTask = rxquery {
+ for x in Sources.values Array.empty do
+ iter (untouched.Add x)
+ }
+
+ do! iterTask.WaitAsync testContext.CancellationToken
+ do! emptyTask.WaitAsync testContext.CancellationToken
+ CollectionAssert.AreEqual ([| 1..6 |], seen.ToArray (), "iter must run the action for every element in order")
+ Assert.IsEmpty (untouched, "iter over an empty source must never run the action")
+ }
+
+ []
+ member _.``rxquery iter faults with the exception of the action and stops processing`` () : Task = task {
+ let boom : exn = InvalidOperationException "boom"
+ let seen = ResizeArray()
+
+ let iterTask = rxquery {
+ for x in Sources.values [| 1; 2; 3 |] do
+ iter (if x = 2 then raise boom else seen.Add x)
+ }
+
+ // R3 reports the exception of the action through OnErrorResume, which faults ForEachAsync and disposes it,
+ // so the elements after it never reach the action
+ let! error =
+ Assert.ThrowsAsync(
+ (fun () -> iterTask.WaitAsync testContext.CancellationToken),
+ "iter must fault when the action throws"
+ )
+
+ Assert.AreSame (boom, error, "iter must fault with the exception of the action")
+ CollectionAssert.AreEqual ([| 1 |], seen.ToArray (), "iter must stop processing at the element whose action throws")
+ }
+
+ []
+ member _.``rxquery over a hot subject only sees elements pushed after subscription`` () : Task = task {
+ use subject = new Subject ()
+
+ let query = rxquery {
+ for i in subject do
where (i % 2 = 0)
select i
+ }
- // Same as:
- // if i % 2 = 0 then
- // yield i
+ // Building the query does not subscribe, and a subject drops the elements nobody listens to, so 2 is lost
+ subject.OnNext 2
+ let valuesTask = query |> Observable.toArray testContext.CancellationToken
+ subject.OnNext 3
+ subject.OnNext 4
+ subject.OnNext 5
+ Assert.IsFalse (valuesTask.IsCompleted, "The query must stay open while the subject is open")
+ subject.OnCompleted (Result.Success)
+ Assert.IsTrue (valuesTask.IsCompletedSuccessfully, "Completing the subject must complete the query synchronously")
+ let! values = valuesTask.WaitAsync testContext.CancellationToken
+ CollectionAssert.AreEqual ([| 4 |], values, "Only the even element pushed after the subscription must arrive")
+ }
+
+ []
+ member _.``a failing source faults a task operator of rxquery with the same exception`` () : Task = task {
+ let boom : exn = InvalidOperationException "boom"
+ let countTask = rxquery {
+ for x in Sources.failingAfter [| 1; 2 |] boom do
+ count
}
- // No-one listens yet (vs R3.ReplaySubject
- r3Bus.OnNext 2
+ // SelectMany forwards the failure of the source at once, and an R3 task operator faults with the exception of a failure
+ Assert.IsTrue (countTask.IsFaulted, "count must fault as soon as the synchronous source fails")
+ let! error =
+ Assert.ThrowsAsync(
+ (fun () -> countTask.WaitAsync testContext.CancellationToken :> Task),
+ "Awaiting count over a failing source must throw"
+ )
+ Assert.AreSame (boom, error, "count must fault with the exception of the source failure")
+ }
+
+ []
+ member _.``a failing source fails an observable rxquery after its earlier elements`` () =
+ let boom : exn = InvalidOperationException "boom"
- use subscription =
- R3.ObservableExtensions.SubscribeAwait (
- interesting,
- fun i cancellationToken ->
- task {
- // Listen events
+ use recorder =
+ rxquery {
+ for x in Sources.failingAfter [| 1; 2 |] boom do
+ select (x * 10)
+ }
+ |> Recorder.Attach
- hasvisited <- true
+ CollectionAssert.AreEqual ([| 10; 20 |], recorder.Values, "The elements emitted before the failure must arrive")
+ Assert.IsEmpty (recorder.Errors, "A terminal failure must not be reported through OnErrorResume")
+ assertFailedWith boom recorder "The query must fail with the exception of the source failure"
- Assert.AreEqual(4, i)
+ []
+ member _.``an error resumed by the source faults a task operator of rxquery`` () : Task = task {
+ let boom : exn = InvalidOperationException "boom"
- return ()
- }
- |> System.Threading.Tasks.ValueTask
+ let countTask = rxquery {
+ for x in Sources.resumingError boom do
+ count
+ }
+
+ // An R3 task operator turns the first OnErrorResume into a fault and disposes its subscription
+ Assert.IsTrue (countTask.IsFaulted, "count must fault at the first resumed error")
+ let! error =
+ Assert.ThrowsAsync(
+ (fun () -> countTask.WaitAsync testContext.CancellationToken :> Task),
+ "Awaiting count after a resumed error must throw"
)
+ Assert.AreSame (boom, error, "count must fault with the resumed exception")
+ }
+
+ []
+ member _.``an error resumed by the source passes through an observable rxquery`` () =
+ let boom : exn = InvalidOperationException "boom"
+
+ use recorder =
+ rxquery {
+ for x in Sources.resumingError boom do
+ select (x * 10)
+ }
+ |> Recorder.Attach
+
+ // SelectMany and Select forward OnErrorResume without stopping, so the element after the error still arrives
+ CollectionAssert.AreEqual ([| 10; 20 |], recorder.Values, "The elements around the resumed error must arrive")
+ let error = Assert.ContainsSingle (recorder.Errors, "The resumed error must be forwarded once")
+ Assert.AreSame (boom, error, "The forwarded error must be the resumed exception")
+ Assert.IsTrue (recorder.IsCompletedSuccessfully, "A resumed error must not terminate the query")
+
+ []
+ member _.``a throwing select faults toArray of the rxquery with the same exception`` () : Task = task {
+ let boom : exn = InvalidOperationException "boom"
+
+ let query = rxquery {
+ for x in Sources.values [| 1; 2; 3 |] do
+ select (if x = 2 then raise boom else x * 10)
+ }
+
+ let valuesTask = query |> Observable.toArray testContext.CancellationToken
+ // R3 reports the exception of the selector through OnErrorResume, which ToArrayAsync turns into a fault
+ Assert.IsTrue (valuesTask.IsFaulted, "toArray must fault at the element whose projection throws")
+ let! error =
+ Assert.ThrowsAsync(
+ (fun () -> valuesTask.WaitAsync testContext.CancellationToken :> Task),
+ "Awaiting toArray of the query must throw"
+ )
+ Assert.AreSame (boom, error, "toArray must fault with the exception of the selector")
+ }
+
+ []
+ member _.``a throwing select is resumed by an observable rxquery and the later elements still arrive`` () =
+ let boom : exn = InvalidOperationException "boom"
+
+ use recorder =
+ rxquery {
+ for x in Sources.values [| 1; 2; 3 |] do
+ select (if x = 2 then raise boom else x * 10)
+ }
+ |> Recorder.Attach
+
+ // R3 Select turns the exception of the selector into OnErrorResume and keeps projecting
+ CollectionAssert.AreEqual ([| 10; 30 |], recorder.Values, "The elements whose projection succeeds must arrive")
+ let error =
+ Assert.ContainsSingle (recorder.Errors, "The exception of the selector must be resumed once")
+ Assert.AreSame (boom, error, "The resumed error must be the exception of the selector")
+ Assert.IsTrue (recorder.IsCompletedSuccessfully, "An exception of the selector must not terminate the query")
+
+ []
+ member _.``rxqueryWith cancels head and unsubscribes from the source when its token is cancelled`` () : Task = task {
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+ use cancellation = CancellationTokenSource.CreateLinkedTokenSource testContext.CancellationToken
+
+ // The documented form: the application of rxqueryWith must be parenthesized in front of the query
+ let headTask =
+ (rxqueryWith cancellation.Token)
+ { for x in probe.Watch subject do
+ head }
+
+ Assert.AreEqual (1, probe.Active, "head must subscribe to the source when the query is built")
+ Assert.IsFalse (headTask.IsCompleted, "head must wait for the first element")
+ // R3 registers on the token before subscribing; the callback runs inside Cancel,
+ // disposes the subscription and then cancels the task
+ cancellation.Cancel ()
+ Assert.AreEqual (TaskStatus.Canceled, headTask.Status, "Cancel must cancel the task before it returns")
+ Assert.AreEqual (1, probe.Disposed, "Cancel must dispose the subscription to the subject, which never completes by itself")
+
+ // Awaited directly rather than through WaitAsync: the task is already cancelled and must carry its own token
+ let! error =
+ Assert.ThrowsAsync((fun () -> headTask :> Task), "Awaiting the cancelled head must throw")
+
+ Assert.AreEqual (cancellation.Token, error.CancellationToken, "The exception must carry the token of rxqueryWith")
+ }
+
+ []
+ member _.``rxqueryWith passes its token to every query operator that returns a task`` () =
+ use cancellation = new CancellationTokenSource ()
+ cancellation.Cancel ()
+ let cancelledQuery = rxqueryWith cancellation.Token
+ let source = Sources.values [| 3; 1; 2 |]
+ let processed = ResizeArray()
+
+ // R3 registers on the token before it subscribes, and registering on a cancelled token runs the callback at once,
+ // so every operator that observes the token returns a cancelled task even over a synchronous source
+ let assertCancelled name (operatorTask : #Task) =
+ Assert.AreEqual (TaskStatus.Canceled, operatorTask.Status, $"%s{name} must be cancelled by the token of rxqueryWith")
+
+ cancelledQuery {
+ for x in source do
+ count
+ }
+ |> assertCancelled "count"
+
+ cancelledQuery {
+ for x in source do
+ all (x > 0)
+ }
+ |> assertCancelled "all"
+
+ cancelledQuery {
+ for x in source do
+ contains 1
+ }
+ |> assertCancelled "contains"
+
+ cancelledQuery {
+ for x in source do
+ exactlyOne
+ }
+ |> assertCancelled "exactlyOne"
+
+ cancelledQuery {
+ for x in source do
+ exactlyOneOrDefault
+ }
+ |> assertCancelled "exactlyOneOrDefault"
+
+ cancelledQuery {
+ for x in source do
+ find (x > 0)
+ }
+ |> assertCancelled "find"
+
+ cancelledQuery {
+ for x in source do
+ head
+ }
+ |> assertCancelled "head"
+
+ cancelledQuery {
+ for x in source do
+ headOrDefault
+ }
+ |> assertCancelled "headOrDefault"
+
+ cancelledQuery {
+ for x in source do
+ last
+ }
+ |> assertCancelled "last"
+
+ cancelledQuery {
+ for x in source do
+ lastOrDefault
+ }
+ |> assertCancelled "lastOrDefault"
+
+ cancelledQuery {
+ for x in source do
+ maxBy x
+ }
+ |> assertCancelled "maxBy"
+
+ cancelledQuery {
+ for x in source do
+ minBy x
+ }
+ |> assertCancelled "minBy"
+
+ cancelledQuery {
+ for x in source do
+ sumBy x
+ }
+ |> assertCancelled "sumBy"
+
+ cancelledQuery {
+ for x in source do
+ iter (processed.Add x)
+ }
+ |> assertCancelled "iter"
+
+ // The observer is disposed before the subscription, so the synchronous elements never reach the action
+ Assert.IsEmpty (processed, "No element may reach iter once its token is cancelled")
+
+ []
+ member _.``RxQueryBuilder exposes its token while rxquery uses an uncancellable token`` () =
+ use cancellation = new CancellationTokenSource ()
+ let builder = RxQueryBuilder cancellation.Token
+ let withToken = rxqueryWith cancellation.Token
+ let withoutToken = RxQueryBuilder ()
- // Publish some events, "4" should be heard
- [ 3..5 ] |> List.iter r3Bus.OnNext
- // Note: Query will not be awaited, that's why delay.
- System.Threading.Thread.Sleep 300
- Assert.AreEqual(true, hasvisited)
+ Assert.AreEqual (cancellation.Token, builder.CancellationToken, "RxQueryBuilder must expose the token it was created with")
+ Assert.AreEqual (cancellation.Token, withToken.CancellationToken, "rxqueryWith must create a builder with its token")
+ Assert.AreEqual (CancellationToken.None, rxquery.CancellationToken, "rxquery must not observe any token")
+ Assert.AreEqual (CancellationToken.None, withoutToken.CancellationToken, "The parameterless RxQueryBuilder must not observe any token")
diff --git a/tests/FSharp.Control.R3.Tests/ChunkTests.fs b/tests/FSharp.Control.R3.Tests/ChunkTests.fs
new file mode 100644
index 0000000..c7a0b78
--- /dev/null
+++ b/tests/FSharp.Control.R3.Tests/ChunkTests.fs
@@ -0,0 +1,767 @@
+namespace FSharp.Control.R3.Tests
+
+open System
+open System.Threading
+open System.Threading.Tasks
+open Microsoft.Extensions.Time.Testing
+open Microsoft.VisualStudio.TestTools.UnitTesting
+open R3
+open FSharp.Control.R3
+open FSharp.Control.R3.Tests.TestHelpers
+
+/// Assertions and building blocks shared by the chunk tests.
+[]
+module private ChunkTestHelpers =
+
+ ///
+ /// Compares the chunks one by one, because
+ /// compares nested arrays by reference.
+ ///
+ let assertChunks (expected : 'T array array) (actual : 'T array array) (message : string) =
+ Assert.HasCount (expected.Length, actual, message)
+
+ expected
+ |> Array.iteri (fun index chunk -> CollectionAssert.AreEqual (chunk, actual[index], $"%s{message} (chunk %d{index})"))
+
+ /// Asserts the whole outcome of a chunked sequence that completed successfully without reporting an error.
+ let assertCompletedWith (expected : 'T array array) (recorder : Recorder<'T array>) (message : string) =
+ assertChunks expected recorder.Values message
+ Assert.IsEmpty (recorder.Errors, "The chunked sequence must not report an error")
+ Assert.IsTrue (recorder.IsCompletedSuccessfully, "The chunked sequence must complete successfully")
+
+ /// Asserts that every element opened a window that failed with the error and still emitted that element as a chunk.
+ let assertEveryWindowFailed (error : exn) (elements : int array) (recorder : Recorder) =
+ assertChunks (elements |> Array.map Array.singleton) recorder.Values "Every failed window must still emit its chunk"
+ Assert.HasCount (elements.Length, recorder.Errors, "Every window failure must be reported once")
+
+ for reported in recorder.Errors do
+ Assert.AreSame (error, reported, "The original exception must be reported, without a wrapper")
+
+ Assert.IsTrue (recorder.IsCompletedSuccessfully, "A window failure must not terminate the chunked sequence")
+
+ ///
+ /// A window that runs until the test releases
+ /// the gate of the element that opened it.
+ ///
+ let gatedWindow (selector : GatedSelector) =
+ Func(fun value cancellationToken -> ValueTask (selector.InvokeTask cancellationToken value :> Task))
+
+ ///
+ /// Every chunking function, configured so that a chunk closes only when it holds two elements or when the source
+ /// terminates: the test never advances the fake time, the window never finishes and the boundaries never emit.
+ ///
+ let everyChunker (time : TimeProvider) (boundaries : Observable) : struct (string * (Observable -> Observable)) array =
+ // A window that never finishes leaves its elements to the flush on completion
+ let endlessWindow =
+ Func(fun _ _ -> ValueTask ((TaskCompletionSource ()).Task))
+
+ [|
+ struct ("chunkBySize", Observable.chunkBySize 2)
+ struct ("chunkBy ChunkCount", Observable.chunkBy (ChunkCount 2))
+ struct ("chunkBy ChunkTimeSpan", Observable.chunkBy (ChunkTimeSpan (TimeSpan.FromSeconds 1., time)))
+ struct ("chunkBy ChunkTimeSpanCount", Observable.chunkBy (ChunkTimeSpanCount (TimeSpan.FromSeconds 1., 2, time)))
+ struct ("chunkBy ChunkMilliseconds", Observable.chunkBy (ChunkMilliseconds (1000, time)))
+ struct ("chunkBy ChunkMillisecondsCount", Observable.chunkBy (ChunkMillisecondsCount (1000, 2, time)))
+ struct ("chunkBy ChunkAsyncWindow", Observable.chunkBy (ChunkAsyncWindow (endlessWindow, true)))
+ struct ("chunkBy ChunkWindowBoundaries", Observable.chunkBy (ChunkWindowBoundaries boundaries))
+ struct ("chunkByBoundaries", Observable.chunkByBoundaries boundaries)
+ |]
+
+[]
+type ChunkTests (testContext : TestContext) =
+
+ // Behaviour that every chunking function shares
+
+ []
+ member _.``every chunking function emits no chunk for an empty source and completes successfully`` () =
+ let time = FakeTimeProvider ()
+ use boundaries = new Subject ()
+
+ for struct (name, chunk) in everyChunker time boundaries do
+ // Empty completes during Subscribe, and no R3 chunk operator flushes an empty buffer
+ use recorder =
+ (Observable.empty () : Observable)
+ |> chunk
+ |> Recorder.Attach
+ Assert.IsEmpty (recorder.Values, $"%s{name} must not emit a chunk for an empty source")
+ Assert.IsEmpty (recorder.Errors, $"%s{name} must not report an error for an empty source")
+ Assert.IsTrue (recorder.IsCompletedSuccessfully, $"%s{name} must complete when the empty source completes")
+
+ []
+ member _.``every chunking function forwards OnErrorResume of the source and keeps buffering`` () =
+ let boom : exn = InvalidOperationException "boom"
+ let time = FakeTimeProvider ()
+ use boundaries = new Subject ()
+
+ for struct (name, chunk) in everyChunker time boundaries do
+ // resumingError emits 1, reports the error, emits 2 and completes, all during Subscribe
+ use recorder = Sources.resumingError boom |> chunk |> Recorder.Attach
+ assertChunks [| [| 1; 2 |] |] recorder.Values $"%s{name} must keep the elements around the error in one chunk"
+ let error = Assert.ContainsSingle (recorder.Errors, $"%s{name} must forward the error once")
+ Assert.AreSame (boom, error, $"%s{name} must forward the original error")
+ Assert.IsTrue (recorder.IsCompletedSuccessfully, $"%s{name} must not terminate on a resumed error")
+
+ []
+ member _.``every chunking function flushes the buffered elements before forwarding a source failure`` () =
+ let boom : exn = InvalidOperationException "boom"
+ let time = FakeTimeProvider ()
+ use boundaries = new Subject ()
+
+ for struct (name, chunk) in everyChunker time boundaries do
+ // Every R3 chunk operator flushes its buffer on completion, whatever the result
+ use recorder =
+ Sources.failingAfter [| 1 |] boom
+ |> chunk
+ |> Recorder.Attach
+ assertChunks [| [| 1 |] |] recorder.Values $"%s{name} must flush the buffered element before the failure"
+ Assert.IsEmpty (recorder.Errors, $"%s{name} must not report the terminal failure as a resumed error")
+ assertFailedWith boom recorder $"%s{name} must forward the failure of the source"
+
+ // chunkBySize
+
+ []
+ member _.``chunkBySize emits each chunk as soon as it is full and flushes the remainder on completion`` () =
+ use subject = new Subject ()
+ use recorder = subject |> Observable.chunkBySize 2 |> Recorder.Attach
+
+ // Subject and Chunk(count) deliver synchronously on the calling thread
+ subject.OnNext 1
+ Assert.IsEmpty (recorder.Values, "A chunk must not be emitted before it is full")
+ subject.OnNext 2
+ assertChunks [| [| 1; 2 |] |] recorder.Values "A chunk must be emitted as soon as it is full"
+ subject.OnNext 3
+ subject.OnCompleted Result.Success
+
+ assertCompletedWith [| [| 1; 2 |]; [| 3 |] |] recorder "Completion must flush the partial chunk"
+
+ []
+ member _.``chunkBySize emits no empty chunk when the length of the source is a multiple of the size`` () =
+ use recorder =
+ Sources.values [| 1..4 |]
+ |> Observable.chunkBySize 2
+ |> Recorder.Attach
+
+ assertCompletedWith [| [| 1; 2 |]; [| 3; 4 |] |] recorder "Completion must not add an empty chunk after full chunks"
+
+ []
+ member _.``chunkBySize flushes the partial chunk before forwarding a failure`` () =
+ let boom : exn = InvalidOperationException "boom"
+ use recorder =
+ Sources.failingAfter [| 1; 2; 3 |] boom
+ |> Observable.chunkBySize 2
+ |> Recorder.Attach
+
+ assertChunks [| [| 1; 2 |]; [| 3 |] |] recorder.Values "R3 must flush the partial chunk on a failed completion too"
+ Assert.IsEmpty (recorder.Errors, "The terminal failure must not be reported as a resumed error")
+ assertFailedWith boom recorder "The failure of the source must be forwarded"
+
+ []
+ member _.``chunkBySize rejects a non-positive size when called, before subscribing`` (size : int) =
+ // The library validates the size itself: R3 would also reject it, but with its message as the parameter name and no value
+ assertArgumentRejected (Observable.chunkBySize size) "chunkSize" size "A non-positive chunk size must be rejected when chunkBySize is called"
+
+ // chunkBy ChunkCount
+
+ []
+ member _.``chunkBy ChunkCount splits the source into chunks of the count and flushes the remainder`` () =
+ use recorder =
+ Sources.values [| 1..5 |]
+ |> Observable.chunkBy (ChunkCount 2)
+ |> Recorder.Attach
+
+ assertCompletedWith [| [| 1; 2 |]; [| 3; 4 |]; [| 5 |] |] recorder "ChunkCount must chunk like chunkBySize"
+
+ []
+ member _.``chunkBy ChunkCount rejects a non-positive count when called, before subscribing`` (count : int) =
+ assertArgumentRejected
+ (Observable.chunkBy (ChunkCount count))
+ "configuration"
+ count
+ "A non-positive count must be rejected when chunkBy is called"
+
+ // chunkBy ChunkTimeSpan
+
+ []
+ member _.``chunkBy ChunkTimeSpan emits the chunk when the window opened by its first element elapses`` () =
+ let time = FakeTimeProvider ()
+ use subject = new Subject ()
+ use recorder =
+ subject
+ |> Observable.chunkBy (ChunkTimeSpan (TimeSpan.FromSeconds 3., time))
+ |> Recorder.Attach
+
+ // R3 starts the window timer at the first element of a window, so idle time opens no window
+ time.Advance (TimeSpan.FromSeconds 10.)
+ subject.OnNext 1
+ time.Advance (TimeSpan.FromSeconds 2.)
+ subject.OnNext 2
+ Assert.IsEmpty (recorder.Values, "No chunk may be emitted before the window elapses")
+ // FakeTimeProvider fires the due timer synchronously inside Advance
+ time.Advance (TimeSpan.FromSeconds 1.)
+ assertChunks [| [| 1; 2 |] |] recorder.Values "The window must emit its elements 3 seconds after its first element"
+ subject.OnNext 3
+ subject.OnCompleted Result.Success
+
+ assertCompletedWith [| [| 1; 2 |]; [| 3 |] |] recorder "Completion must flush the window opened by the last element"
+
+ []
+ member _.``chunkBy ChunkTimeSpan never emits an empty chunk while nothing is buffered`` () =
+ let time = FakeTimeProvider ()
+ use subject = new Subject ()
+ use recorder =
+ subject
+ |> Observable.chunkBy (ChunkTimeSpan (TimeSpan.FromSeconds 3., time))
+ |> Recorder.Attach
+
+ time.Advance (TimeSpan.FromHours 1.)
+ Assert.IsEmpty (recorder.Values, "No window may elapse before the first element")
+ subject.OnNext 1
+ time.Advance (TimeSpan.FromSeconds 3.)
+ // The window timer fires once, and the next one starts only with the next element
+ time.Advance (TimeSpan.FromHours 1.)
+ subject.OnCompleted Result.Success
+
+ assertCompletedWith [| [| 1 |] |] recorder "Only the window of the single element may emit a chunk"
+
+ // chunkBy ChunkTimeSpanCount
+
+ []
+ member _.``chunkBy ChunkTimeSpanCount emits on reaching the count and the next element opens a new window`` () =
+ let time = FakeTimeProvider ()
+ use subject = new Subject ()
+
+ use recorder =
+ subject
+ |> Observable.chunkBy (ChunkTimeSpanCount (TimeSpan.FromSeconds 3., 2, time))
+ |> Recorder.Attach
+
+ subject.OnNext 1
+ time.Advance (TimeSpan.FromSeconds 2.)
+ subject.OnNext 2
+ assertChunks [| [| 1; 2 |] |] recorder.Values "A full chunk must be emitted as soon as the count is reached"
+ // Reaching the count stops the window timer, which would otherwise emit an empty chunk at its due time
+ time.Advance (TimeSpan.FromSeconds 2.)
+ assertChunks [| [| 1; 2 |] |] recorder.Values "The stopped window of the full chunk must not emit"
+ subject.OnNext 3
+ time.Advance (TimeSpan.FromSeconds 2.)
+ assertChunks [| [| 1; 2 |] |] recorder.Values "The next window must start with its own first element"
+ time.Advance (TimeSpan.FromSeconds 1.)
+ assertChunks [| [| 1; 2 |]; [| 3 |] |] recorder.Values "A window that elapses before the count is reached must emit the partial chunk"
+ subject.OnNext 4
+ subject.OnCompleted Result.Success
+
+ assertCompletedWith [| [| 1; 2 |]; [| 3 |]; [| 4 |] |] recorder "Completion must flush the partial chunk"
+
+ []
+ member _.``chunkBy ChunkTimeSpanCount rejects a non-positive count when called, before subscribing`` (count : int) =
+ let time = FakeTimeProvider ()
+
+ // R3 does not validate this count: 0 would fail every element and a negative count would fail only at subscription
+ assertArgumentRejected
+ (Observable.chunkBy (ChunkTimeSpanCount (TimeSpan.FromSeconds 1., count, time)))
+ "configuration"
+ count
+ "A non-positive count must be rejected when chunkBy is called"
+
+ // chunkBy ChunkMilliseconds
+
+ []
+ member _.``chunkBy ChunkMilliseconds emits the chunk exactly when the window of that many milliseconds elapses`` () =
+ let time = FakeTimeProvider ()
+ use subject = new Subject ()
+ use recorder =
+ subject
+ |> Observable.chunkBy (ChunkMilliseconds (1500, time))
+ |> Recorder.Attach
+
+ subject.OnNext 1
+ subject.OnNext 2
+ time.Advance (TimeSpan.FromMilliseconds 1499.)
+ Assert.IsEmpty (recorder.Values, "The window must still be open 1 ms before its due time")
+ // FakeTimeProvider fires a timer exactly at its due time, so this proves a window of 1500 milliseconds
+ time.Advance (TimeSpan.FromMilliseconds 1.)
+ assertChunks [| [| 1; 2 |] |] recorder.Values "The window must emit its elements 1500 ms after its first element"
+ subject.OnNext 3
+ subject.OnCompleted Result.Success
+
+ assertCompletedWith [| [| 1; 2 |]; [| 3 |] |] recorder "Completion must flush the window opened by the last element"
+
+ // chunkBy ChunkMillisecondsCount
+
+ []
+ member _.``chunkBy ChunkMillisecondsCount emits on reaching the count or exactly when the window elapses`` () =
+ let time = FakeTimeProvider ()
+ use subject = new Subject ()
+ use recorder =
+ subject
+ |> Observable.chunkBy (ChunkMillisecondsCount (1500, 2, time))
+ |> Recorder.Attach
+
+ subject.OnNext 1
+ subject.OnNext 2
+ assertChunks [| [| 1; 2 |] |] recorder.Values "A full chunk must be emitted as soon as the count is reached"
+ subject.OnNext 3
+ time.Advance (TimeSpan.FromMilliseconds 1499.)
+ assertChunks [| [| 1; 2 |] |] recorder.Values "The window of the partial chunk must still be open 1 ms before its due time"
+ time.Advance (TimeSpan.FromMilliseconds 1.)
+ assertChunks [| [| 1; 2 |]; [| 3 |] |] recorder.Values "The window must emit the partial chunk exactly when it elapses"
+ subject.OnCompleted Result.Success
+
+ assertCompletedWith [| [| 1; 2 |]; [| 3 |] |] recorder "Completion must not emit an empty chunk"
+
+ []
+ member _.``chunkBy ChunkMillisecondsCount rejects a non-positive count when called, before subscribing`` (count : int) =
+ let time = FakeTimeProvider ()
+
+ assertArgumentRejected
+ (Observable.chunkBy (ChunkMillisecondsCount (1000, count, time)))
+ "configuration"
+ count
+ "A non-positive count must be rejected when chunkBy is called"
+
+ // chunkBy ChunkAsyncWindow
+
+ []
+ member _.``chunkBy ChunkAsyncWindow opens a window at the first element and emits the chunk when the window finishes`` () : Task = task {
+ use subject = new Subject ()
+ let window = GatedSelector ignore
+ use recorder =
+ subject
+ |> Observable.chunkBy (ChunkAsyncWindow (gatedWindow window, true))
+ |> Recorder.Attach
+
+ subject.OnNext 1
+ subject.OnNext 2
+ subject.OnNext 3
+ // R3 opens a window only for an element that arrives while no window runs, so 2 and 3 join the window of 1
+ CollectionAssert.AreEqual ([| 1 |], window.Started, "Only the first element may open a window")
+ Assert.IsEmpty (recorder.Values, "No chunk may be emitted while the window runs")
+
+ window.Release 1
+ do! recorder.WaitForValuesAsync (testContext.CancellationToken, 1)
+ // R3 emits the chunk and closes the window under one lock, so an element pushed now always opens a new window
+ subject.OnNext 4
+ CollectionAssert.AreEqual ([| 1; 4 |], window.Started, "The first element after a window closed must open a new window")
+
+ window.Release 4
+ do! recorder.WaitForValuesAsync (testContext.CancellationToken, 2)
+ subject.OnCompleted Result.Success
+
+ assertCompletedWith [| [| 1; 2; 3 |]; [| 4 |] |] recorder "Every window must emit the elements that arrived while it ran"
+ }
+
+ []
+ member _.``chunkBy ChunkAsyncWindow flushes the open window on completion and cancels the window token`` () =
+ use subject = new Subject ()
+ let window = GatedSelector ignore
+ use recorder =
+ subject
+ |> Observable.chunkBy (ChunkAsyncWindow (gatedWindow window, true))
+ |> Recorder.Attach
+
+ subject.OnNext 1
+ subject.OnNext 2
+ let token = Assert.ContainsSingle (window.Tokens, "Only the first element may open a window")
+ Assert.IsFalse (token.IsCancellationRequested, "The window token must stay active while the source runs")
+ // The window is never released: R3 cancels the window token, flushes the buffer and completes, all inside OnCompleted
+ subject.OnCompleted Result.Success
+
+ assertCompletedWith [| [| 1; 2 |] |] recorder "Completion must flush the elements of the open window"
+ Assert.IsTrue (token.IsCancellationRequested, "Completion must cancel the token of the open window")
+
+ []
+ member _.``chunkBy ChunkAsyncWindow with a synchronous window emits one chunk per element`` () =
+ let opened = ResizeArray()
+
+ let synchronousWindow =
+ Func(fun value _ ->
+ opened.Add value
+ ValueTask.CompletedTask
+ )
+
+ use recorder =
+ Sources.values [| 1; 2; 3 |]
+ |> Observable.chunkBy (ChunkAsyncWindow (synchronousWindow, true))
+ |> Recorder.Attach
+
+ // A window that completes synchronously closes inside OnNext, so every element opens and closes its own window
+ CollectionAssert.AreEqual ([| 1; 2; 3 |], opened.ToArray (), "Every element must open a window")
+ assertCompletedWith [| [| 1 |]; [| 2 |]; [| 3 |] |] recorder "A synchronous window must emit every element alone"
+
+ []
+ member _.``chunkBy ChunkAsyncWindow resumes a window failure and still emits the chunk`` () =
+ let boom : exn = InvalidOperationException "boom"
+ let failingWindow =
+ Func(fun _ _ -> ValueTask (Task.FromException boom))
+
+ use recorder =
+ Sources.values [| 1; 2 |]
+ |> Observable.chunkBy (ChunkAsyncWindow (failingWindow, true))
+ |> Recorder.Attach
+
+ // R3 reports a window exception through OnErrorResume and still emits the chunk when the window ends
+ assertEveryWindowFailed boom [| 1; 2 |] recorder
+
+ []
+ member _.``disposing a chunkBy ChunkAsyncWindow subscription cancels the window token and unsubscribes from the source`` () =
+ use subject = new Subject ()
+ let probe = SubscriptionProbe ()
+ let window = GatedSelector ignore
+
+ use recorder =
+ subject
+ |> probe.Watch
+ |> Observable.chunkBy (ChunkAsyncWindow (gatedWindow window, true))
+ |> Recorder.Attach
+
+ subject.OnNext 1
+ let token = Assert.ContainsSingle (window.Tokens, "The first element must open a window")
+ Assert.IsFalse (token.IsCancellationRequested, "The window token must stay active until the subscription is disposed")
+ recorder.Dispose ()
+
+ Assert.IsTrue (token.IsCancellationRequested, "Disposing must cancel the token of the running window")
+ // The subject never completes by itself, so this disposal proves the unsubscription
+ Assert.AreEqual (1, probe.Disposed, "Disposing must unsubscribe from the source")
+ // Nothing is asserted on the recorder: disposing only cancels the window token, R3 keeps the buffer and would still
+ // emit it when the window ends, and only the disposed recorder, which ignores every notification, discards it
+
+ []
+ member _.``chunkBy ChunkAsyncWindow resumes the window on the captured synchronization context only when ConfigureAwait is true``
+ (configureAwait : bool)
+ : Task
+ = task {
+ use subject = new Subject ()
+ // The gate resumes its awaiter asynchronously, so only ConfigureAwait decides where the window continuation runs
+ let gate = TaskCompletionSource (TaskCreationOptions.RunContinuationsAsynchronously)
+ let window = Func(fun _ _ -> ValueTask gate.Task)
+ use recorder =
+ subject
+ |> Observable.chunkBy (ChunkAsyncWindow (window, configureAwait))
+ |> Recorder.Attach
+ let context = RecordingSynchronizationContext ()
+
+ // R3 awaits the window inside OnNext, so the window captures the context that is current when the element arrives
+ context.Run (fun () -> subject.OnNext 1)
+ gate.SetResult ()
+ do! recorder.WaitForValuesAsync (testContext.CancellationToken, 1)
+
+ assertChunks [| [| 1 |] |] recorder.Values "The window must emit its element when it finishes"
+ let expectedPosts = if configureAwait then 1 else 0
+ Assert.AreEqual (expectedPosts, context.Posts, "Only a window awaited with ConfigureAwait may resume on the captured context")
+ }
+
+ // chunkBy ChunkWindowBoundaries
+
+ []
+ member _.``chunkBy ChunkWindowBoundaries emits the buffer on every boundary, including empty chunks`` () =
+ use subject = new Subject ()
+ use boundaries = new Subject ()
+ use recorder =
+ subject
+ |> Observable.chunkBy (ChunkWindowBoundaries boundaries)
+ |> Recorder.Attach
+
+ subject.OnNext 1
+ subject.OnNext 2
+ Assert.IsEmpty (recorder.Values, "Elements must be buffered until a boundary arrives")
+ boundaries.OnNext 0
+ // R3 emits an empty chunk for a boundary that finds nothing buffered
+ boundaries.OnNext 0
+ subject.OnNext 3
+ subject.OnCompleted Result.Success
+
+ assertCompletedWith [| [| 1; 2 |]; [||]; [| 3 |] |] recorder "Every boundary and the completion must emit the buffer"
+
+ []
+ member _.``chunkBy ChunkWindowBoundaries forwards a boundary OnErrorResume and keeps chunking`` () =
+ let boom : exn = InvalidOperationException "boom"
+ use subject = new Subject ()
+ use boundaries = new Subject ()
+ use recorder =
+ subject
+ |> Observable.chunkBy (ChunkWindowBoundaries boundaries)
+ |> Recorder.Attach
+
+ subject.OnNext 1
+ // R3 reports an error of the boundaries through the chunked sequence instead of terminating it
+ boundaries.OnErrorResume boom
+ subject.OnNext 2
+ boundaries.OnNext 0
+
+ let error =
+ Assert.ContainsSingle (recorder.Errors, "The error of the boundaries must be forwarded once")
+ Assert.AreSame (boom, error, "The original error of the boundaries must be forwarded")
+ assertChunks [| [| 1; 2 |] |] recorder.Values "The error of the boundaries must not close the chunk"
+ Assert.IsTrue (recorder.Completion.IsNone, "The error of the boundaries must not complete the chunked sequence")
+
+ []
+ member _.``chunkBy ChunkWindowBoundaries flushes and completes successfully when the boundaries fail`` () =
+ let boom : exn = InvalidOperationException "boom"
+ use subject = new Subject ()
+ use boundaries = new Subject ()
+ let probe = SubscriptionProbe ()
+
+ use recorder =
+ subject
+ |> probe.Watch
+ |> Observable.chunkBy (ChunkWindowBoundaries boundaries)
+ |> Recorder.Attach
+
+ subject.OnNext 1
+ // R3 completes the chunked sequence with Success whatever the result of the boundaries, so their failure is lost
+ boundaries.OnCompleted (Result.Failure boom)
+
+ assertChunks [| [| 1 |] |] recorder.Values "The completion of the boundaries must flush the buffer"
+ Assert.IsTrue (recorder.IsCompletedSuccessfully, "A failure of the boundaries must complete the chunked sequence successfully")
+ Assert.IsEmpty (recorder.Errors, "A failure of the boundaries must not be reported as an error either")
+ // The subject never completes by itself, so this disposal proves the unsubscription
+ Assert.AreEqual (1, probe.Disposed, "The completed chunked sequence must unsubscribe from the source")
+
+ []
+ member _.``disposing a chunkBy ChunkWindowBoundaries subscription unsubscribes from the source and the boundaries`` () =
+ use subject = new Subject ()
+ use boundaries = new Subject ()
+ let sourceProbe = SubscriptionProbe ()
+ let boundariesProbe = SubscriptionProbe ()
+
+ use recorder =
+ subject
+ |> sourceProbe.Watch
+ |> Observable.chunkBy (ChunkWindowBoundaries (boundariesProbe.Watch boundaries))
+ |> Recorder.Attach
+
+ Assert.AreEqual (1, sourceProbe.Active, "Subscribing must subscribe to the source")
+ Assert.AreEqual (1, boundariesProbe.Active, "Subscribing must subscribe to the boundaries")
+ subject.OnNext 1
+ recorder.Dispose ()
+
+ // Neither subject completes by itself, so these disposals prove the unsubscriptions
+ Assert.AreEqual (0, sourceProbe.Active, "Disposing must unsubscribe from the source")
+ Assert.AreEqual (0, boundariesProbe.Active, "Disposing must unsubscribe from the boundaries")
+
+ // chunkByBoundaries
+
+ []
+ member _.``chunkByBoundaries chunks on every tick of boundaries of another element type`` () =
+ use subject = new Subject ()
+ use ticks = new Subject ()
+ use recorder =
+ subject
+ |> Observable.chunkByBoundaries ticks
+ |> Recorder.Attach
+
+ subject.OnNext 1
+ subject.OnNext 2
+ ticks.OnNext ()
+ ticks.OnNext ()
+ subject.OnNext 3
+ subject.OnCompleted Result.Success
+
+ assertCompletedWith [| [| 1; 2 |]; [||]; [| 3 |] |] recorder "Every tick and the completion must emit the buffer"
+
+ []
+ member _.``chunkByBoundaries chunks on the ticks of Observable.Interval driven by fake time`` () =
+ let time = FakeTimeProvider ()
+ use subject = new Subject ()
+
+ use recorder =
+ subject
+ |> Observable.chunkByBoundaries (Observable.Interval (TimeSpan.FromSeconds 1., time))
+ |> Recorder.Attach
+
+ subject.OnNext 1
+ subject.OnNext 2
+ // The periodic timer of Interval fires synchronously inside Advance
+ time.Advance (TimeSpan.FromSeconds 1.)
+ assertChunks [| [| 1; 2 |] |] recorder.Values "The first tick must emit the elements buffered before it"
+ time.Advance (TimeSpan.FromSeconds 1.)
+ subject.OnNext 3
+ time.Advance (TimeSpan.FromSeconds 1.)
+ subject.OnCompleted Result.Success
+
+ assertCompletedWith [| [| 1; 2 |]; [||]; [| 3 |] |] recorder "Every tick must emit a chunk, an empty one when nothing is buffered"
+
+ []
+ member _.``chunkByBoundaries forwards an error of string boundaries and completes successfully when they fail`` () =
+ let boom : exn = InvalidOperationException "boom"
+ use subject = new Subject ()
+ use boundaries = new Subject ()
+ let probe = SubscriptionProbe ()
+ use recorder =
+ subject
+ |> probe.Watch
+ |> Observable.chunkByBoundaries boundaries
+ |> Recorder.Attach
+
+ subject.OnNext 1
+ boundaries.OnErrorResume boom
+ boundaries.OnNext "close"
+ subject.OnNext 2
+ // R3 completes the chunked sequence with Success whatever the result of the boundaries
+ boundaries.OnCompleted (Result.Failure (TimeoutException "the boundaries failed"))
+
+ assertChunks [| [| 1 |]; [| 2 |] |] recorder.Values "The tick and the completion of the boundaries must emit the buffer"
+ let error =
+ Assert.ContainsSingle (recorder.Errors, "Only the resumed error of the boundaries may be forwarded")
+ Assert.AreSame (boom, error, "The original error of the boundaries must be forwarded")
+ Assert.IsTrue (recorder.IsCompletedSuccessfully, "A failure of the boundaries must complete the chunked sequence successfully")
+ // The subject never completes by itself, so this disposal proves the unsubscription
+ Assert.AreEqual (1, probe.Disposed, "The completed chunked sequence must unsubscribe from the source")
+
+ // ChunkConfiguration helpers
+
+ []
+ member _.``ChunkConfiguration TimeSpan and TimeSpanCount keep their arguments and capture the default time provider`` () =
+ let windowTime = TimeSpan.FromSeconds 3.
+ // Only read: the default time provider is process-wide and this test runs in parallel with the others. It is the
+ // system provider here, so only the test that replaces it, run apart from the parallel ones, tells the two apart
+ let defaultTimeProvider = ObservableSystem.DefaultTimeProvider
+
+ match (ChunkConfiguration.TimeSpan windowTime : ChunkConfiguration) with
+ | ChunkTimeSpan (actualWindowTime, timeProvider) ->
+ Assert.AreEqual (windowTime, actualWindowTime, "TimeSpan must keep the window time")
+ Assert.AreSame (defaultTimeProvider, timeProvider, "TimeSpan must capture the default time provider")
+ | other -> Assert.Fail $"TimeSpan must build ChunkTimeSpan, not %A{other}"
+
+ match (ChunkConfiguration.TimeSpanCount windowTime 4 : ChunkConfiguration) with
+ | ChunkTimeSpanCount (actualWindowTime, windowLength, timeProvider) ->
+ Assert.AreEqual (windowTime, actualWindowTime, "TimeSpanCount must keep the window time")
+ Assert.AreEqual (4, windowLength, "TimeSpanCount must keep the window length")
+ Assert.AreSame (defaultTimeProvider, timeProvider, "TimeSpanCount must capture the default time provider")
+ | other -> Assert.Fail $"TimeSpanCount must build ChunkTimeSpanCount, not %A{other}"
+
+ []
+ member _.``ChunkConfiguration Milliseconds and MillisecondsCount keep their arguments and capture the default time provider`` () =
+ // Only read: the default time provider is process-wide and this test runs in parallel with the others. It is the
+ // system provider here, so only the test that replaces it, run apart from the parallel ones, tells the two apart
+ let defaultTimeProvider = ObservableSystem.DefaultTimeProvider
+
+ match (ChunkConfiguration.Milliseconds 1500 : ChunkConfiguration) with
+ | ChunkMilliseconds (windowTime, timeProvider) ->
+ Assert.AreEqual (1500, windowTime, "Milliseconds must keep the window time")
+ Assert.AreSame (defaultTimeProvider, timeProvider, "Milliseconds must capture the default time provider")
+ | other -> Assert.Fail $"Milliseconds must build ChunkMilliseconds, not %A{other}"
+
+ match (ChunkConfiguration.MillisecondsCount 1500 4 : ChunkConfiguration