Rx合并+ CombineLatest？

Question

当两个可观测数据都启动时，CombineLatest就会启动。

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

首先，您为A，B选择您的可观测值永远不会发出的特殊值，让它为空：

A specialA = null;
B specialB = null;

然后

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
{
  return a.StartWith(specialA).CombineLatest(b.StartWith(specialB), 
    (aval, bval) => {
      if (aval == specialA) return bval == specialB ? default(C) : bResultSelector(bval);
      if (bval == specialB) return aResultSelector(bval);
      return bothResultSelector(aval, bval);
    }
  ).skip(1);  // skip the first emission where both are special values
}