f# - 如何使用 FSharp.Control.Reactive 观察计算来实现 Observable.take？

Question

我试图了解 FSharp.Control.Reactive 的观察计算是如何工作的，所以我重新实现了 Observable.take 组合器

这是我的初步尝试:

let myTake (n : int) (source : IObservable<'a>) : IObservable<'a> =
    let rec go counter : IObservable<'a> =
        observe {
            let! v = source
            if counter < n
            then yield v
                 yield! go (counter + 1)
        }
    go 0

但是，当我运行下面的测试时:

use subcription =
    subj
    |> myTake 2
    |> Observable.subscribe (printfn "next:%d")

subj.OnNext 1
subj.OnNext 2
subj.OnNext 3

waitForKey "my take"

我得到输出:

next:1 
next:2 
next:2 
next:3 
next:3 
next:3 

我该如何解决这个问题？

我还尝试在使用 observable.ofSeq 创建的可观察对象上运行 myTake，但它失败得更糟，即它只是生成了重复多次的输入序列。我认为这与 ofSeq 返回冷可观察但不完全理解该行为的事实有关。

我怎样才能让它也适用于冷可观察量？

谢谢!

Answer 1

我不太熟悉observable 计算生成器，但快速查看源代码表明Bind 操作(在let! ) 是 implemented using the Rx SelectMany operation . SelectMany 操作将为事件的每次 启动启动工作流的其余部分，因此您看到的行为是预期的行为。

图片来自this article illustrates the behaviour well :

我不确定使用 observable 计算构建器实现 Observable.take 的好方法是什么 - 坦率地说，我一直认为 observables 不是特别好适合 F# 计算表达式，因为计算表达式附带的通常直觉根本不适用于基于推送的可观察对象。

我认为，如果您可以只组合内置操作，那么 observables 工作得很好，但是当您需要实现自己的自定义原语时，它们并不是特别好——而且大多数时候，我只使用 F#代理来实现逻辑并将其包装在后面并可观察到。