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

這個工作對我來說：

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)))); 
} 

Then this：

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!! 
1z 
5z 
6z 
6a 
2a 
2b 
2c 

Answer 2

首先選擇了一個特殊的價值，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 
}