Rx 合并 + 组合最新

本文关键字：最新 组合 合并 Rx | 更新日期: 2023-09-27 17:55:56

当两个可观察量都启动时，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

如何实现此运算符？

Rx 合并 + 组合最新

这对

我有用：

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!!1z5z6Z6一2一2b2c

首先，你为 A、B 选择你的可观察量永远不会发出的特殊值，让它为 null：

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
}
相关文章：