如何在可观测值之间建立依赖关系？

Question

我想要一个测试Rx组件的工具，它可以这样工作:给定指定为'v seq的事件的顺序和一个键选择器函数(keySelector :: 'v -> 'k)，我希望创建一个Map<'k, IObservable<'k>>，其中保证分组的可观测值按照上面可枚举的全局顺序产生值。

例如：makeObservables isEven [1;2;3;4;5;6] ...should produce

{ true : -2-4-6|,
  false: 1-3-5| }

这是我的尝试，看上去如下：

open System
open System.Reactive.Linq
open FSharp.Control.Reactive

let subscribeAfter (o1: IObservable<'a>) (o2 : IObservable<'b>) : IObservable<'b> =
    fun (observer : IObserver<'b>) ->
        let tempObserver = { new IObserver<'a> with
                                member this.OnNext x = ()
                                member this.OnError e = observer.OnError e
                                member this.OnCompleted () = o2 |> Observable.subscribeObserver observer |> ignore
                            }
        o1.Subscribe tempObserver
    |> Observable.Create

let makeObservables (keySelector : 'a -> 'k) (xs : 'a seq) : Map<'k, IObservable<'a>> =
    let makeDependencies : ('k * IObservable<'a>) seq -> ('k * IObservable<'a>) seq = 
        let makeDep ((_, o1), (k2, o2)) = (k2, subscribeAfter o1 o2)

        Seq.pairwise
        >> Seq.map makeDep

    let makeObservable x = (keySelector x, Observable.single x)

    let firstItem = 
        Seq.head xs 
        |> makeObservable 
        |> Seq.singleton

    let dependentObservables =
        xs
        |> Seq.map makeObservable
        |> makeDependencies

    dependentObservables
    |> Seq.append firstItem
    |> Seq.groupBy fst
    |> Seq.map (fun (k, obs) -> (k, obs |> Seq.map snd |> Observable.concatSeq))
    |> Map.ofSeq

[<EntryPoint>]
let main argv = 
    let isEven x = (x % 2 = 0)

    let splits : Map<bool, IObservable<int>> =
        [1;2;3;4;5]
        |> makeObservables isEven

    use subscription =
        splits
        |> Map.toSeq
        |> Seq.map snd
        |> Observable.mergeSeq
        |> Observable.subscribe (printfn "%A")


    Console.ReadKey() |> ignore
    0 // return an integer exit code

...but结果与预期的不一样，观测值也不符合全局顺序。

显然，每个组中的项都是正确的，但是当组被合并时，它更像是一个连接，然后是一个合并。

预期输出为：1 2 3 4 5 ...but实际输出为1 3 5 2 4

我做错了什么？

谢谢!

Answer 1

你说你想要这样：

{ true : -2-4-6|,
  false: 1-3-5| }

但你真的创造了这个：

{ true : 246|,
  false: 135| }

由于观测到的项目之间没有时间间隔，所以merge基本上有一个恒定的竞赛条件。Rx保证给定序列的元素1会在元素2之前触发，但是Merge在这种情况下不提供任何保证。

如果您希望Merge能够按照原来的顺序重新排序，则需要在可观测值中引入时间间隔。