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

Question

我在这里使用稍微修改过的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)|]

Answer 1

您已经可以使用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 中的注释与 f# 查询表达式中的from完全相同/几乎相同。let!结果如下。buffer另请注意我稍后如何使用，expression就像在 f# 查询表达式中一样。

START
1 1
1 2
1 3
2 1
2 2
2 3
3 1
3 2
3 3
2 2
2 3
2 4
3 2
3 3
3 4
4 2
4 3
4 4
3 3
3 4
3 5
4 3
4 4
4 5
5 3
5 4
5 5
4 4
4 5
5 4
5 5
5 5
END