如何在 Rx.NET 中组合两个不同的 GroupedStreams？

如何解决如何在 Rx.NET 中组合两个不同的 GroupedStreams？

This question 类似，但不适用于我的情况，因为用户需要合并来自同一个 IGroupedobservable 的可观察流，而我想合并来自不同组的流。

我有以下结构和流：

type A = {
  Id: int
  Value: int
}

type B = {
  Id: int
  Value: int
}

//subjects to test input,just any source of As and Bs
let subjectA: Subject<A> = Subject.broadcast
let subjectB: Subject<B> = Subject.broadcast

//grouped streams
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 和 B 的内部可观察值，并获得 (A,B) 对的可观察值，其中 A.Id = B.Id

我想要的签名是这样的 IObservable<IGroupedobservable<int,A>> -> IObservable<IGroupedobservable<int,B>> -> IObservable<IGroupedobservable<int,(A,B)>> where for all (A,B),A.Id = B.Id

我尝试了很多 combineLatest、groupJoin、过滤器和地图变体，但都没有成功。

我将 F# 与 Rx.Net 和 FSharp.Control.Reactive 一起使用，但如果您知道 C#（或任何语言，真的）的答案，请发布

解决方法

这是您可以使用的自定义运算符 GroupJoin。它基于 Select、Merge、GroupBy 和 Where 运算符：

/// <summary>
/// Groups and joins the elements of two observable sequences,based on common keys.
/// </summary>
public static IObservable<(TKey Key,IObservable<TLeft> Left,IObservable<TRight> Right)>
    GroupJoin<TLeft,TRight,TKey>(
    this IObservable<TLeft> left,IObservable<TRight> right,Func<TLeft,TKey> leftKeySelector,Func<TRight,TKey> rightKeySelector,IEqualityComparer<TKey> keyComparer = null)
{
    // Arguments validation omitted
    keyComparer ??= EqualityComparer<TKey>.Default;
    return left
        .Select(x => (x,(TRight)default,Type: 1,Key: leftKeySelector(x)))
        .Merge(right.Select(x => ((TLeft)default,x,Type: 2,Key: rightKeySelector(x))))
        .GroupBy(e => e.Key,keyComparer)
        .Select(g => (
            g.Key,g.Where(e => e.Type == 1).Select(e => e.Item1),g.Where(e => e.Type == 2).Select(e => e.Item2)
        ));
}

用法示例：

var subjectA = new Subject<A>();
var subjectB = new Subject<B>();

IObservable<IGroupedObservable<int,(A,B)>> query = subjectA
    .GroupJoin(subjectB,a => a.Id,b => b.Id)
    .SelectMany(g => g.Left.Zip(g.Right,(a,b) => (g.Key,a,b)))
    .GroupBy(e => e.Key,e => (e.a,e.b));

,

我不清楚这是否是您想要的。因此，首先使用运行器代码进行澄清可能会有所帮助。假设以下运行程序代码：

var aSubject = new Subject<A>();
var bSubject = new Subject<B>();

var groupedA = aSubject.GroupBy(a => a.Id);
var groupedB = bSubject.GroupBy(b => b.Id);

//Initiate solution

solution.Merge()
    .Subscribe(t => Console.WriteLine($"(Id = {t.a.Id},AValue = {t.a.Value},BValue = {t.b.Value}  )"));

aSubject.OnNext(new A() { Id = 1,Value = 1 });
aSubject.OnNext(new A() { Id = 1,Value = 2 });

bSubject.OnNext(new B() { Id = 1,Value = 10 });
bSubject.OnNext(new B() { Id = 1,Value = 20 });
bSubject.OnNext(new B() { Id = 1,Value = 30 });

您是否想看到以下输出：

(Id = 1,AValue = 1,BValue = 10)
(Id = 1,AValue = 2,BValue = 20)
(Id = 1,BValue = 30)
(Id = 1,BValue = 30)

如果是这种情况，您可以通过以下方式获得解决方案：

var solution = groupedA.Merge()
    .Join(groupedB.Merge(),_ => Observable.Never<Unit>(),b) => (a,b)
    )
    .Where(t => t.a.Id == t.b.Id)
    .GroupBy(g => g.a.Id);

如果这是一个长期运行的进程的一部分，我会警告这里会有内存/性能影响。这会将所有 A 和 B 对象无限期地保留在内存中，等待查看它们是否可以配对。要缩短它们在内存中的保留时间，请将 Observable.Never() 调用更改为适当的窗口，以了解每个对象在内存中的保留时间。

,

首先，这有你想要的签名：

let cartesian left right =
    rxquery {
        for a in left do
        for b in right do
        yield a,b
    }

let mergeGroups left right =
    rxquery {
        for (leftGroup : IGroupedObservable<'key,'a>) in left do
        for (rightGroup : IGroupedObservable<'key,'b>) in right do
        if leftGroup.Key = rightGroup.Key then
            let merged = cartesian leftGroup rightGroup
            yield {
                new IGroupedObservable<_,_> with
                    member __.Key = leftGroup.Key
                    member __.Subscribe(observer) = merged.Subscribe(observer)
            }
    }

然而，在我的测试中，这些组都是空的。我没有足够的 Rx 经验不知道为什么，但也许其他人知道。