F#Reactive等待Observable完成

我发现了各种各样的SO问题,但无法找出F#解决方案.我需要阻止等待事件向我发射以检查它返回的数据.我使用Rx接收事件3次:

let disposable =
    Observable.take 3 ackNack
    |> Observable.subscribe (
        fun (sender, data) ->
            Console.WriteLine("{0}", data.AckNack)
            Assert.True(data.TotalAckCount > 0u)
    )

我想将结果转换为列表,以便稍后可以通过测试框架(xUnit)进行检查,或等待所有3个事件完成并通过Assert.True.

在继续之前,我将如何等待3场比赛？我可以看到Observable.wait其他消息来源的建议Async.RunSynchronously.

我认为最简单的选择是使用Async.AwaitObservable函数 - 遗憾的是,这在F#核心库中尚不可用,但您可以从FSharpx.Async包中获取它,或者只是从GitHub复制函数 soruce .

使用该函数,您应该能够编写如下内容:

let _, data = 
  Observable.take 3 ackNack
  |> Async.AwaitObservable
  |> Async.RunSynchronously

Console.WriteLine("{0}", data.AckNack)
Assert.True(data.TotalAckCount > 0u)