如何积累观测值

Question

我应该定义一个函数，它将返回 IObservable<'u>

accumulate : ('t -> 'a -> 't * 'u option) -> 't -> IObservable<'a> -> IObservable<'u>

这样我的函数 ft obs' 将 obs 的可观察事件累积到类型为 't 的累加器中，并在观察事件 'a' 的 'snd (f acc a)' 计算结果为 'Some u' 时发出可观察事件 u。

到目前为止我已经实现了以下功能：

let accumulate (f:'t -> 'a -> 't * 'u option) t obs = 
 Observable.scan (fun _ x -> snd (f t x)) None obs

我真的不明白这个可观察的扫描是如何工作的，在这种情况下我的函数返回 IObservable<'u option> 。我该如何解决这个问题？我走在正确的轨道上吗？

Answer 1

功能fun _ x -> snd (f t x)不完整。线索是第一个参数_被忽略，并且结果元组的第一部分被调用丢弃snd。

没有累积，因为调用时始终使用与最初传递给 的f t x相同值。该原始值应该是初始值，并且应该作为第二个参数的一部分传递。taccumulatetscan

生成的元组的第一部分f:'t -> 'a -> 't * 'u option是累积值。所以这就是需要返回的部分，scan以便它可以f再次传递并一遍又一遍地积累。

在您的问题中，要求是累积并在元组的第二部分为 时传递事件Some 'u。所以问题是如何同时做到：累积't和过滤'u？

答案是将累积的价值与它的Some 'u作用相结合f。因此，您需要将元组保留为状态scan，然后使用choose和只保留第二部分snd。

这就是您正在寻找的：

let accumulate (f:'t -> 'a -> 't * 'u option) t obs =
    obs
    |> Observable.scan (fun (acc, _) x -> f acc x) (t, None)
    |> Observable.choose snd

理解scan

scan是一个函数，通过将状态与一系列值一起传递给函数来承载变化的状态。特别是它可以用于累积值，例如int运行总计：

let keepTotal obs =
    obs
    |> Observable.scan (fun total v -> total + v) 0

这相当于在带有 mutable 的命令式代码中执行此操作total：

let mutable total = 0

let keepTotal2 obs =
    obs
    |> Observable.map (fun v -> 
        total <- total + v
        total
    )

请注意这两个版本如何具有相同的元素：

• 初始值：0
• 累加器功能：total + v

当然，第二个版本，即使它使用了map，也是糟糕的功能代码，因为它使用了一个外部可变变量，这是一个很大的NO NO。

您原来的问题可以用同样的方式解决：

let accumulate2 (f:'t -> 'a -> 't * 'u option) t obs =
    let mutable acc = t
    obs
    |> Observable.choose (fun x ->
        let acc2, uOp = f acc x
        acc <- acc2
        uOp
    )

尽管这个使用了一个可变变量，这在函数式编程中是丑陋的（并且不必要的），但它在功能上是可以的，因为该变量acc是内部的，外部没有代码accumulate2可以看到它。虽然还是丑。