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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Changed

- **Breaking:** `AwaitOperationConfiguration` cases are prefixed with `Await` (`AwaitSequential`, `AwaitParallel 4`, ...) so they no longer collide with `System.Threading.Tasks.Parallel`
- **Breaking:** the `Async` functions raise the original exception instead of an `AggregateException`, and a cancelled operation cancels the computation instead of failing it
- **Breaking:** the `Async` `toLookup` overloads no longer take a `cancellationToken`; they observe the token of the computation
- **Breaking:** the `OptionExtensions`/`ValueOptionExtensions` types are replaced by `Observable.choose`, which takes a `voption` chooser, and `ObservableOption.choose` for `option`; the `Observable.Extensions` type is replaced by `Observable.ofSeq`
- **Breaking:** after `open FSharp.Control.R3`, `Observable.choose` is the R3 function, like `Observable.map` and `Observable.filter` already were; use `Microsoft.FSharp.Control.Observable.choose` for `IObservable` and F# events
Expand Down
62 changes: 48 additions & 14 deletions src/FSharp.Control.R3/AsyncObservable.fs
Original file line number Diff line number Diff line change
@@ -1,10 +1,44 @@
module FSharp.Control.R3.Async

open R3
open System
open System.Threading
open System.Threading.Tasks
open R3
open FSharp.Control.R3

/// Awaiting helpers shared by the Async wrappers.
module internal Interop =

// Async.AwaitTask resumes on the synchronization context of the caller, but it raises the AggregateException of a faulted task
// and turns a cancelled task into a failure, while the Task flavour of this library surfaces the original exception and the
// cancellation. The helpers keep Async.AwaitTask and translate its outcome on the thread it resumed on.

/// Raises the outcome of a task that did not complete successfully through the continuations of the computation,
/// which keeps the stack trace of the original throw.
let private failedOutcome (task : Task) (error : exn) : Async<'T> =
Async.FromContinuations (fun (_, onError, onCancel) ->
if task.IsCanceled then
onCancel (TaskCanceledException task)
else
match error with
| :? AggregateException as aggregate when aggregate.InnerExceptions.Count = 1 -> onError aggregate.InnerExceptions[0]
| error -> onError error
)

/// Awaits the task, raising the original exception of a fault and cancelling the computation when the task is cancelled.
let awaitTask (task : Task<'T>) : Async<'T> = async {
match! Async.AwaitTask task |> Async.Catch with
| Choice1Of2 result -> return result
| Choice2Of2 error -> return! failedOutcome task error
Comment on lines +29 to +32
}

/// Awaits the task, raising the original exception of a fault and cancelling the computation when the task is cancelled.
let awaitUnitTask (task : Task) : Async<unit> = async {
match! Async.AwaitTask task |> Async.Catch with
| Choice1Of2 () -> return ()
| Choice2Of2 error -> return! failedOutcome task error
}

/// <remarks>Caution! All functions returning <see cref="Async`1"/> are blocking and may never return if awaited</remarks>
module Observable =

Expand All @@ -14,15 +48,15 @@ module Observable =
let! ct = Async.CancellationToken
return!
ObservableExtensions.AggregateAsync (source, seed, f, ct)
|> Async.AwaitTask
|> Interop.awaitTask
}

/// Determines whether all elements of an observable satisfy a predicate
let all (f : 't -> bool) source = async {
let! ct = Async.CancellationToken
return!
ObservableExtensions.AllAsync (source, f, ct)
|> Async.AwaitTask
|> Interop.awaitTask
}

/// Determines whether an observable sequence contains a specified value
Expand All @@ -31,15 +65,15 @@ module Observable =
let! ct = Async.CancellationToken
return!
ObservableExtensions.AnyAsync (source, ct)
|> Async.AwaitTask
|> Interop.awaitTask
}

/// Returns the first element of an observable sequence
let firstAsync source = async {
let! ct = Async.CancellationToken
return!
ObservableExtensions.FirstAsync (source, ct)
|> Async.AwaitTask
|> Interop.awaitTask
}

/// <summary>
Expand All @@ -54,15 +88,15 @@ module Observable =
let! ct = Async.CancellationToken
return!
ObservableExtensions.ForEachAsync (source, action, ct)
|> Async.AwaitTask
|> Interop.awaitUnitTask
}

/// Returns the last element of an observable sequence till its completion or cancellation
let length source = async {
let! ct = Async.CancellationToken
return!
ObservableExtensions.CountAsync (source, ct)
|> Async.AwaitTask
|> Interop.awaitTask
}

/// <summary>Maps the given observable with the given asynchronous function</summary>
Expand All @@ -86,18 +120,18 @@ module Observable =
|> ValueTask<'T>
)

let inline toArray source = async {
let toArray source = async {
let! ct = Async.CancellationToken
return!
ObservableExtensions.ToArrayAsync (source, ct)
|> Async.AwaitTask
|> Interop.awaitTask
}

let toList source = async {
let! ct = Async.CancellationToken
let! array =
ObservableExtensions.ToArrayAsync (source, ct)
|> Async.AwaitTask
|> Interop.awaitTask
return List.ofArray array
}

Expand Down Expand Up @@ -126,21 +160,21 @@ module Extensions =
let! ct = Async.CancellationToken
return!
ObservableExtensions.ToLookupAsync (source, keySelector, ct)
|> Async.AwaitTask
|> Interop.awaitTask
}

static member toLookup (source : Observable<'T>, keySelector : 'T -> 'Key, keyComparer : IEqualityComparer<'Key>) = async {
let! ct = Async.CancellationToken
return!
ObservableExtensions.ToLookupAsync (source, keySelector, keyComparer = keyComparer, cancellationToken = ct)
|> Async.AwaitTask
|> Interop.awaitTask
}

static member toLookup (source : Observable<'T>, keySelector : 'T -> 'Key, elementSelector : 'T -> 'Element) = async {
let! ct = Async.CancellationToken
return!
ObservableExtensions.ToLookupAsync (source, keySelector, elementSelector = elementSelector, cancellationToken = ct)
|> Async.AwaitTask
|> Interop.awaitTask
}

static member toLookup
Expand All @@ -149,5 +183,5 @@ module Extensions =
let! ct = Async.CancellationToken
return!
ObservableExtensions.ToLookupAsync (source, keySelector, elementSelector, keyComparer = keyComparer, cancellationToken = ct)
|> Async.AwaitTask
|> Interop.awaitTask
}