Rx Merge + CombateLatest?

Question

Die CombineLatest startet, wenn beide Observablen gestartet wurden.

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

Die Merge Operator beginnt, wenn entweder A oder B begonnen haben. Es kann jedoch die letzten Werte von A und B nicht kombinieren.

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

Ich brauche einen Operator wie Merge außer dass verhält es mir erlauben würde, A und B neuesten Werte zu kombinieren, wenn beide beobachtbaren begonnen haben:

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

Seine Unterschrift könnte wie folgt aussehen:

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 

Wie könnte dieser Operator implementiert werden?

Answer 1

Dies funktioniert für mich:

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

Dann dies:

... gibt dies:

 
x!! 
y!! 
z!! 
1z 
5z 
6z 
6a 
2a 
2b 
2c 

Answer 2

Zuerst wählen spezielle Werte für A, B, die Ihre Observable nie emittieren, lassen Sie es null sein:

A specialA = null; 
B specialB = null; 

dann

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 
}