GroupJoin 与一个 window 在流完成时结束

Question

我有两个数据源，它们从不同的线程转储数据。我正在尝试通过密钥加入这两个来源。我可以使用 GroupJoin 来做到这一点。我必须使用 Observable.Never 以便 window 永远不会结束。在加入流方面一切正常。当两个源都完成转储数据时，他们调用各自的 onComplete on Observers。我希望流媒体在收到两个源的 OnComplete 后立即结束。正如我使用的 Observable.Never 流永远不会结束并且我的 aggSource 的 Oncomplete 事件永远不会被调用。

有没有办法告诉 Rx 在 OnComplete 收到两个源时关闭 window 而不是无限地保持打开状态？

我是 Rx 的新手，不确定这是否可以实现。下面是代码片段。提前致谢！！

var l = Source1;
var r = Source2;

var q = r.GroupJoin(l,
                _ => Observable.Never<Unit>(), // windows from each left event going on forever 
                _ => Observable.Never<Unit>(), // windows from each right event going on forever
                (left, obsOfRight) => Tuple.Create(left, obsOfRight)); // create tuple of left event with observable of right events

        var joinSource =   q.SelectMany(e => {
            return e.Item2.Where(
               x =>
               {
                 return x.ID== e.Item1.ID;
               })
               .Select(v=>  (Item1:v.Value, Item2: e.Item1.Value));
        });



var aggSource = joinSource.GroupBy(x => x.Item1).SelectMany(grp =>
            {
                return grp.Scan(0.0, (accumulator, current) => accumulator + current.Item2).Select(z => (Group: grp.Key, Value: z));
            });



aggSource.Subscribe(x => dictResults[x.Group] = x,
              y => { Console.WriteLine("Error Ocurred: " + y.Message); completed = true; },
              () => { completed = true; Console.WriteLine("Subcription comnpleted"); }
              );
// dict results is dictionary which is my projection which is shown to View. Right now my view is just console window.

Answer 1

持续时间选择器控制连接 window 重叠 - 我们需要在任一源完成时缩短重叠。首先，我们将使用 LastOrDefaultAsync 在发出 OnComplete 时获得通知。

var either = Observable.CombineLatest(l.LastOrDefaultAsync(), r.LastOrDefaultAsync());

现在我们可以修改示例中的持续时间选择器：

var q = r.GroupJoin(l,
                _ => Observable.Never<Unit>().TakeUntil(either), // windows from each left event until l or r completes
                _ => Observable.Never<Unit>().TakeUntil(either), // windows from each right event until l or r completes
                (left, obsOfRight) => Tuple.Create(left, obsOfRight)); // create tuple of left event with observable of right events

这应该会导致级联并完成其余的可观察管道。

GroupJoin 与一个 window 在流完成时结束

GroupJoin with one single window ending on completion of streams

c#

system.reactive

rx.net