Fsharpx Async.AwaitObservable不会调用取消延续

Question

我正在尝试在Async.AwaitObservable开始使用的异步工作流程中使用Fsharpx Async.StartWithContinuations.出于某种原因,如果用于启动此工作流的取消令牌在等待可观察时被取消(但在工作流的其他部分期间不被取消),则永远不会调用取消继续.但是,如果我把它放在一个中use! __ = Async.OnCancel (interruption),那么就会调用中断函数.有人可以澄清为什么会发生这种情况,最好的方法是做什么,并确保其中一个延续函数始终被调用？

open System
open System.Reactive.Linq
open FSharp.Control.Observable
open System.Threading

[<EntryPoint>]
let main _ =
    let cancellationCapability = new CancellationTokenSource()

    let tick = Observable.Interval(TimeSpan.FromSeconds 1.0)
    let test = async {
        let! __ = Async.AwaitObservable tick
        printfn "Got a thing." }

    Async.StartWithContinuations(test,
        (fun () -> printfn "Finished"),
        (fun exn -> printfn "Error!"),
        (fun exn -> printfn "Canceled!"),
        cancellationCapability.Token)

    Thread.Sleep 100
    printfn "Cancelling..."
    cancellationCapability.Cancel()

    Console.ReadLine() |> ignore
    0 // return an integer exit code

Answer 1

GitHub 上的 Fsharpx 版本似乎已经包含修复程序（不是我实现的）。但是，NuGet 上的当前版本 (1.8.41) 尚未更新以包含此修复。请参阅此处的更改。

编辑 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) }