修改 RX 计算表达式生成器以保存以前的值答案

【问题标题】：Modifying RX Computational Expression Builder to hold previous values修改 RX 计算表达式生成器以保存以前的值
【发布时间】：2013-08-06 15:48:13
【问题描述】：

我正在使用此处介绍的 RX 构建器的略微修改版本：

http://mnajder.blogspot.com/2011/09/when-reactive-framework-meets-f-30.html

我的计算表达式不是直接采用IObservable<'T>，而是具有以下类型：

type MyType<'a,'b> = MyType of (IObservable<'a> -> IObservable<'b>)

let extract (MyType t) = t

然后组合子采取以下形式：

let where (f: 'b -> bool) (m:MyType<_,'b>) = MyType(fun input -> (extract m input).Where(f))

在表达式本身中，我经常需要引用回馈入流中的先前值。为此，我定义了一个 MyType，它维护了一个滚动的不可变列表，其中包含 n 最近的值。

let history n = 
    MyType(fun input ->
        Observable.Create(fun (o:IObserver<_>) ->
            let buffer = new History<_>(n)
            o.OnNext(HistoryReadOnly(buffer))
            input.Subscribe(buffer.Push, o.OnError, o.OnCompleted)
        )
    )

有了这个，我现在可以做类似的事情：

let f = obs {
    let! history = history 20
    // Run some other types, and possibly do something with history
}

我发现我经常使用此历史记录，理想情况下我希望将其直接嵌入到IObservable<'a>。显然我不能那样做。所以我的问题是，什么是一种合理的方式来介绍我在这里所拥有的这种历史概念。我应该扩展IObservable<'T>（不知道该怎么做），包装IObservable<'T>吗？

感谢任何建议。

编辑：添加了完整的示例代码。

open System
open System.Collections.Generic
open System.Reactive.Subjects
open System.Reactive.Linq

// Container function
type MyType<'a,'b> = MyType of (IObservable<'a> -> IObservable<'b>)
let extract (MyType t) = t

// Mini Builder
let internal mbind (myTypeB:MyType<'a,'b>) (f:'b -> MyType<'a,'c>) = 
    MyType(fun input ->
        let obsB = extract myTypeB input
        let myTypeC= fun resB -> extract (f resB) input
        obsB.SelectMany(myTypeC)
    )

type MyTypeBuilder() = 
    member x.Bind (m,f) = mbind m f
    member x.Combine (a,b) = MyType(fun input -> (extract a input).Concat(extract b input))
    member x.Yield (r) = MyType(fun input -> Observable.Return(r))
    member x.YieldFrom (m:MyType<_,_>) = m
    member x.Zero() = MyType(fun input -> Observable.Empty())
    member x.Delay(f:unit -> MyType<'a,'b>) = f() 

let mtypeBuilder = new MyTypeBuilder()

// Combinators
let simplehistory = 
    MyType(fun input ->
        Observable.Create(fun (o:IObserver<_>) ->
            let buffer = new List<_>()
            o.OnNext(buffer)
            input.Subscribe(buffer.Add, o.OnError, o.OnCompleted)
        )
    )

let where (f: 'b -> bool) m = MyType(fun input -> (extract m input).Where(f))
let take (n:int) m = MyType(fun input -> (extract m input).Take(n))
let buffer m = MyType(fun input -> (extract m input).Buffer(1))
let stream = MyType(id)

// Example
let myTypeResult (t:MyType<'a,'b>) (input:'a[]) = (extract t (input.ToObservable().Publish().RefCount())).ToArray().Single()

let dat = [|1 .. 20|]

let example = mtypeBuilder {
    let! history = simplehistory
    let! someEven = stream |> where(fun v -> v % 2 = 0) // Foreach Even
    let! firstValAfterPrevMatch = stream |> take 1 // Potentially where a buffer operation would run, all values here are after i.e. we cant get values before last match
    let! odd = stream |> where(fun v -> v % 2 = 1) |> take 2 // Take 2 odds that follow it
    yield (history.[history.Count - 1], history.[0], someEven,firstValAfterPrevMatch, odd) // Return the last visited item in our stream, the very first item, an even, the first value after the even and an odd
}

let result = myTypeResult example dat

val result : (int * int * int * int * int) [] =
  [|(5, 1, 2, 3, 5); (7, 1, 2, 3, 7); (7, 1, 4, 5, 7); (9, 1, 4, 5, 9);
    (9, 1, 6, 7, 9); (11, 1, 6, 7, 11); (11, 1, 8, 9, 11); (13, 1, 8, 9, 13);
    (13, 1, 10, 11, 13); (15, 1, 10, 11, 15); (15, 1, 12, 13, 15);
    (17, 1, 12, 13, 17); (17, 1, 14, 15, 17); (19, 1, 14, 15, 19);
    (19, 1, 16, 17, 19)|]

【问题讨论】：

鉴于您在以下一些好的答案之后添加了赏金，您能否提供一些具体示例，这些选项不能满足您的需求？
@DaxFohl 添加了一个完整的示例。

标签： f# system.reactive

【解决方案1】：

使用standard Rx workflow builder，您可以创建一个处理您的示例用例的函数history：

let history (io:IObservable<_>) = 
  io.Scan(new List<_>(), (fun l t -> l.Add t; l)).Distinct()

let io = new Subject<int>()
let newio = rx { let! history = history io
                 let! even = io.Where(fun v -> v % 2 = 0)
                 let! next = io.Take 1
                 let! odd = io.Where(fun v -> v % 2 = 1).Take 2
                 yield (history.Last(), history.[0], even, next, odd) }

newio.Subscribe(printfn "%O") |> ignore

for i = 1 to 20 do
  io.OnNext i

扩展它以提供历史长度限制应该是微不足道的。您是否有特定的原因需要定义自己的类型/构建器，或者这样做只是为了达到这样的目的？

这是一个使用组合器的示例。您只需在 rx 块之外定义可观察对象。您可以让 history 以不同的方式处理不可变的历史记录而不是持久列表，所以无论您的需要是什么。

let history (io:IObservable<_>) = 
  io.Scan(new List<_>(), (fun l t -> l.Add t; l))

let newest (hist:'a List) = hist.Last()
let extract (ioHist:'a List IObservable) = ioHist.Select newest
let take (i:int) (ioHist: 'a List IObservable) = ioHist.Take i
let where (f: 'a -> bool) (ioHist: 'a List IObservable) = ioHist.Where(fun hist -> f(newest hist))

let io = new Subject<int>()
let history1 = history io
let newio =
 rx { let! hist = history1.Distinct()
      let! even = extract (history1 |> where (fun v -> v % 2 = 0))
      let! next = extract (history1 |> take 1)
      let! odd = extract (history1 |> where (fun v -> v % 2 = 1) |> take 2)
      yield (hist.Last(), hist.[0], even, next, odd) }

【讨论】：

封装类型的存在是因为它允许添加新的支持实现而不会破坏事情的发展。目前是IObservable<'a> -> IObservable<'b>，也许这会变成IHistoryObservable<'a> -> IObservable<'b>。这可能会改变。我在这里寻找的是一种使历史成为可观察对象本身的一部分的方法，即我希望所有组合器都可以从组合器中访问历史。这意味着不必在表达式中隐式定义历史记录，但可以在需要时访问它。寻找一种干净的方法来做到这一点。
@Dave 但history 函数根本不需要在表达式中使用。您可以直接调用 let obsWithHistory = history obs，这将创建一个 observable，每次 obs 发生事件时报告 obs 的完整历史记录。
@Dave 我不特别关心上述内容，因为List 是一个刚刚更新的可变变量。如果您查看我最初的 C# 解决方案（预编辑），您会看到有一些不同的选项，它们都是不可变的：您可以使 obsWithHistory 报告自其实例化以来的完整历史记录，或者仅将历史记录报告给每个观察者，因为那个特定的观察者一直在听。这些使得表达式稍微变得更加麻烦，因为每个变量现在都是一个数组，但我认为它给了你更多的灵活性。同样，也不是非常确定您需要什么。
如果你定义了一个组合器（基于我的例子）。你可以有类似MyType(input -> (extract m input)....do something) 的东西。希望能够从input 访问历史记录。基本上修改/扩展/包装用 observable 做一些事情。
@Dave 添加了组合子

【解决方案2】：

您已经可以使用Observable.Buffer 来执行此操作。抱歉，我的 F# 帽子今天没有考虑 C#。

IObservable<int> source = ...
IOBservable<IList<int>> buffered = source.Buffer(5,1)

将为您创建一个列表流。

或者尝试在 LINQ 中使用更像 F# 查询表达式的缓冲区

Console.WriteLine ("START");
var source = new List<int> () { 1, 2, 3, 4, 5 }.ToObservable ();

// LINQ C#'s Monad sugar
var r = 
        from buffer in source.Buffer (3, 1)
        from x in buffer
        from y in buffer
        select new { x,y};


r.Subscribe (o=>Console.WriteLine (o.x + " " + o.y));
Console.WriteLine ("END");

注意 LINQ 中的 from 与 f# 查询表达式中的 let! 完全相同/几乎相同。结果如下。还要注意我稍后在expression 中如何使用buffer，就像在f# 查询表达式中一样。

【讨论】：

你不能。 Observable 将缓冲 N 个元素、N 次或当被另一个流触发停止时，它将匹配这些条件并返回通过流的内容。在这种情况下，历史将匹配并允许表达式继续前进。在此之后，它只会将传入的值记录在一个可变变量中，它将始终具有先前的 N 个值。两者非常不同。我正在寻找一种在我的结构中烘焙这个历史概念的方法。即在 MyType 中，我想从 IObservable 访问历史记录，即使这意味着将其设为新类型。
使用缓冲区，您始终拥有前 N 个样本。您还拥有当前样本。我不明白你为什么不能以此为基础。
有了缓冲区，我还需要知道何时停止缓冲。有了我不知道的历史，历史也可以保存最新的元素。
@Dave 我同意这个答案。使用Buffer 比构建您自己的表达式构建器灵活得多，因为它允许您选择which 可观察来维护历史，而与您使用的表达式构建器无关。您可以维护输入 observable、结果 observable、一些中间 observable 的历史记录，等等。它只是将您的Observable<T> 转换为Observable<List<T>>，以显示每次新阅读的完整历史记录。你还想要什么？使用自定义表达式构建器，您可以锁定整个区块。
@Dave 他给出的 LINQ 示例将等同于您使用表达式生成器的方式。我认为您需要给出一个非常明确的示例，说明您设想的用例在哪里不起作用。如果Buffer 不能解决您的问题，我很想写一些可以解决的问题，但我看不到Buffer 无法解决您需要的用例。我确定我不是唯一一个。给我们一个您所设想的具体案例，Buffer 无济于事。

【解决方案3】：

抱歉，我的 F# 非常生锈，但也许您正在寻找 Scan 运算符。它会在源生成值时将值推送到累加器，然后您可以使用此累加器为您的投影生成下一个值。

在这里（在 C# 中，抱歉）我们采用 [0..10] 的序列，这些值相隔 100 毫秒产生这些值，然后我们返回一个正在运行的 Sum。

var source = Observable.Interval(TimeSpan.FromMilliseconds(100))
                       .Take(10);

source.Scan(
    new List<long>(),               //Seed value
    (acc, value)=>                  //Accumulator function
        {
            acc.Add(value);
            return acc;
        }
    )
    .Select(accumate=>accumate.Sum())

产生的值 [0,1,3,6,10,15,21,28,36,45] 相隔 100 毫秒。

我认为使用此工具，您可以管理值的历史记录（通过将它们添加到历史记录/累加器），然后在选择中使用此历史记录来投影合适的值。

【讨论】：