System.Reactive - 将 2 个可观察量组合成一个具有所有排列的流

Question

我希望将 2 个可观察序列与一个选择器函数相结合，该函数将为两个序列的元素之间的每个可能组合推送结果：

  var o1 = new Subject<string>();
  var o2 = new Subject<string>();
  IObservable<string> result = o1.CreatePermutations(o2, (item1, item2) => $"{item1}{item2}");

预期成果应为：

  o1.OnNext("A"); // nothing happens
  o2.OnNext("1");  // result pushes: "A1"
  o1.OnNext("B");  // result pushes: "B1"
  o2.OnNext("2");   // result pushes: "A2", "B2"
  o1.OnNext("C");  // result pushes: "C1", "C2"

如何实现这样的 CreatePermutations 函数？

我能想到的所有组合运算符(CombineLatest、Zip 等(似乎只适用于每个流中推送的最新值，并且在需要时不会保留/组合每个流的完整历史记录(重播( - 所以目前真的想不出解决这个问题的好方法

谢谢！

Answer 1

这

似乎有效

public static IObservable<string> CreatePermutations(this IObservable<string> source, 
                                                     IObservable<string> other)
{
    return Observable.Create<string>(obs =>
    {
        var or = other.Replay();
        var sequence = source.SelectMany(i => or.Select(j => $"{i}{j}"));                             
        return new CompositeDisposable(new [] { sequence.Subscribe(obs),
                                                or.Connect()
                                              });
    });
}

Answer 2

这是一个

直截了当的Join问题：

public static IObservable<TResult> CreatePermutations<TLeft, TRight, TResult>(
    this IObservable<TLeft> o1, 
    IObservable<TRight> o2, 
    Func<TLeft, TRight, TResult> selector
)
{
    return o1.Join(o2, 
        _ => Observable.Never<Unit>(), 
        _ => Observable.Never<Unit>(), 
        selector
    );
}