functional-programming - 如何积累 Observable

Question

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

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

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

Answer 1

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

没有积累，因为f t x始终使用相同的值调用 t最初传递给 accumulate .那个原版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 , 是糟糕的功能代码，因为它使用了一个外部可变变量，这是一个很大的 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可以看到。不过还是丑。