【问题标题】：How to Accumulate Observables如何积累 Observables
【发布时间】：2018-12-22 04:36:03
【问题描述】：

我应该定义一个函数，它会返回 IObservable

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

所以我的函数 f t obs' 将 obs 的可观察事件累积到类型为 't 的累加器中，并在 'snd (f acc a)' 评估为观察事件 '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。我该如何解决？我在正确的轨道上吗？

【问题讨论】：

标签： functional-programming f#

【解决方案1】：

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

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

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

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

let mutable total = 0

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

注意这两个版本有相同的元素：

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

当然，第二个版本，即使它使用map，也是糟糕的函数式代码，因为它使用了一个很大的不可以的外部可变变量。

你原来的问题可以用同样的方法解决：

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 之外的代码无法看到它。不过还是很丑。

【讨论】：

【解决方案2】：

您可以在 Observable.scan 之后链接 Observable.choose 以获得正确的类型签名

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

【讨论】：

了解scan

了解`scan`