Rx.NET中如何合并两个独立GroupedStreams得到同Key的(A,B)配对分组流
一个同类问题场景与我的需求相近,但实际并不适用:该问题的需求是合并同一个IGroupedObservable内部的可观察流,而我需要合并来自不同分组的可观察流。
我当前定义了如下数据结构与流:
type A = { Id: int Value: int } type B = { Id: int Value: int } // 用于测试输入的Subject,本质是A、B类型数据的输入源 let subjectA: Subject<A> = Subject.broadcast let subjectB: Subject<B> = Subject.broadcast // 按Id字段分组后的流 let groupedA: IObservable<IGroupedObservable<int, A>> = Observable.groupBy (fun a -> a.Id) subjectA let groupedB: IObservable<IGroupedObservable<int, B>> = Observable.groupBy (fun b -> b.Id) subjectB
核心需求:当groupedA的分组Key与groupedB的分组Key相等时,合并二者的内部可观察流,最终得到所有满足A.Id == B.Id的(A,B)配对组成的可观察流。
我期望实现的方法签名形态类似:IObservable<IGroupedObservable<int, A>> -> IObservable<IGroupedObservable<int, B>> -> IObservable<IGroupedObservable<int, (A, B)>>
即最终输出的所有(A,B)配对,都满足配对内的A.Id与B.Id相等。
我已经尝试过combineLatest、groupJoin、流过滤、映射的多种组合方案,但都没有达到预期效果。当前我使用F#结合Rx.NET与FSharp.Control.Reactive实现逻辑,若有C#或其他语言的实现方案也欢迎参考。
内容的提问来源于stack exchange,提问作者fnzr
相关产品推荐
相关产品推荐

