c# - Rx 合并 + CombineLatest？

Question

CombineLatest 在两个 observable 都启动时启动。

A     1----------2---------------
B     -----a----------b---c------
C     -----1a----2a---2b--2c-----   C = A.CombineLatest(B)

Merge 运算符在 A 或 B 启动时启动。但是，它不能合并 A 和 B 的最新值。

A     1----------2---------------
B     -----a----------b---c------
C     1----a-----2----b---c------   C = A.Merge(B)

我需要一个像 Merge 一样的运算符，除了它允许我在两个可观察对象都启动时合并 A 和 B 的最新值:

A    1----------2---------------
B    -----a----------b---c------
C    1----1a----2a---2b--2c-----   C = A.MergeOrCombineLatest(B)

它的签名可能是这样的:

Observable<C> MergeOrCombineLatest<A, B, C>(
     this IObservable<A> a,
     IObservable<B> b,
     Func<A, C> aResultSelector, // When A starts before B
     Func<B, C> bResultSelector, // When B starts before A
     Func<A, B, C> bothResultSelector) // When both A and B have started

如何实现此运算符？

Answer 1

这对我有用:

public static IObservable<C> MergeOrCombineLatest<A, B, C>(
    this IObservable<A> a,
    IObservable<B> b,
    Func<A, C> aResultSelector, // When A starts before B
    Func<B, C> bResultSelector, // When B starts before A
    Func<A, B, C> bothResultSelector) // When both A and B have started
{
    return
        a.Publish(aa =>
            b.Publish(bb =>
                aa.CombineLatest(bb, bothResultSelector).Publish(xs =>
                    aa
                        .Select(aResultSelector)
                        .Merge(bb.Select(bResultSelector))
                        .TakeUntil(xs)
                        .SkipLast(1)
                        .Merge(xs))));
}

然后这个:

var a = new Subject<int>();
var b = new Subject<string>();

var C = a.MergeOrCombineLatest(b, x => $"{x}!!", y => $"{y}!!", (x, y) => $"{x}{y}");

C.Subscribe(x => Console.WriteLine(x));

b.OnNext("x");
b.OnNext("y");
b.OnNext("z");
a.OnNext(1);
a.OnNext(5);
a.OnNext(6);
b.OnNext("a");
a.OnNext(2);
b.OnNext("b");
b.OnNext("c");

...给出这个:

x!!y!!z!!1z5z6z6a2a2b2c