似乎 GitHub 上的 Fsharpx 版本已经包含一个修复（不是我实现的）。但是，NuGet (1.8.41) 上的当前版本尚未更新以包含此修复程序。查看更改here。
编辑 1：
GitHub 上的代码在带有重播语义的 Observables 方面也存在一些问题。我现在已经解决了这个问题，但希望有一个更清洁的解决方案。在考虑有没有办法让它变得更简单之后，我会提交 PR。
/// Creates an asynchronous workflow that will be resumed when the
/// specified observables produces a value. The workflow will return
/// the value produced by the observable.
static member AwaitObservable(observable : IObservable<'T1>) =
let removeObj : IDisposable option ref = ref None
let removeLock = new obj()
let setRemover r =
lock removeLock (fun () -> removeObj := Some r)
let remove() =
lock removeLock (fun () ->
match !removeObj with
| Some d -> removeObj := None
d.Dispose()
| None -> ())
synchronize (fun f ->
let workflow =
Async.FromContinuations((fun (cont,econt,ccont) ->
let rec finish cont value =
remove()
f (fun () -> cont value)
setRemover <|
observable.Subscribe
({ new IObserver<_> with
member x.OnNext(v) = finish cont v
member x.OnError(e) = finish econt e
member x.OnCompleted() =
let msg = "Cancelling the workflow, because the Observable awaited using AwaitObservable has completed."
finish ccont (new System.OperationCanceledException(msg)) })
() ))
async {
let! cToken = Async.CancellationToken
let token : CancellationToken = cToken
#if NET40
use registration = token.Register(fun () -> remove())
#else
use registration = token.Register((fun _ -> remove()), null)
#endif
return! workflow
})
static member AwaitObservable(observable : IObservable<'T1>) =
let synchronize f =
let ctx = System.Threading.SynchronizationContext.Current
f (fun g ->
let nctx = System.Threading.SynchronizationContext.Current
if ctx <> null && ctx <> nctx then ctx.Post((fun _ -> g()), null)
else g() )
let continued = ref false
let continuedLock = new obj()
let removeObj : IDisposable option ref = ref None
let removeLock = new obj()
let setRemover r =
lock removeLock (fun () -> removeObj := Some r)
let remove() =
lock removeLock (fun () ->
match !removeObj with
| Some d ->
removeObj := None
d.Dispose()
| None -> ())
synchronize (fun f ->
let workflow =
Async.FromContinuations((fun (cont,econt,ccont) ->
let rec finish cont value =
remove()
f (fun () -> lock continuedLock (fun () ->
if not !continued then
cont value
continued := true))
let observer =
observable.Subscribe
({ new IObserver<_> with
member __.OnNext(v) = finish cont v
member __.OnError(e) = finish econt e
member __.OnCompleted() =
let msg = "Cancelling the workflow, because the Observable awaited using AwaitObservable has completed."
finish ccont (new System.OperationCanceledException(msg)) })
lock continuedLock (fun () -> if not !continued then setRemover observer else observer.Dispose())
() ))
async {
let! cToken = Async.CancellationToken
let token : CancellationToken = cToken
use __ = token.Register((fun _ -> remove()), null)
return! workflow
})
编辑 2：
对可观察到的热点问题进行了更整洁的修复...
let AwaitObservable(observable : IObservable<'T>) = async {
let! token = Async.CancellationToken // capture the current cancellation token
return! Async.FromContinuations(fun (cont, econt, ccont) ->
// start a new mailbox processor which will await the result
Agent.Start((fun (mailbox : Agent<Choice<'T, exn, OperationCanceledException>>) ->
async {
// register a callback with the cancellation token which posts a cancellation message
#if NET40
use __ = token.Register((fun _ ->
mailbox.Post (Choice3Of3 (new OperationCanceledException("The opeartion was cancelled.")))))
#else
use __ = token.Register((fun _ ->
mailbox.Post (Choice3Of3 (new OperationCanceledException("The opeartion was cancelled.")))), null)
#endif
// subscribe to the observable: if an error occurs post an error message and post the result otherwise
use __ =
observable.FirstAsync()
.Catch(fun exn -> mailbox.Post(Choice2Of3 exn) ; Observable.Empty())
.Subscribe(fun result -> mailbox.Post(Choice1Of3 result))
// wait for the first of these messages and call the appropriate continuation function
let! message = mailbox.Receive()
match message with
| Choice1Of3 reply -> cont reply
| Choice2Of3 exn -> econt exn
| Choice3Of3 exn -> ccont exn })) |> ignore) }