Reaktive Erweiterungen: Puffer bis zum Abschluss eines anderen Datenstroms

Question

Ich mache eine Anfrage und warte auf eine Antwort auf diese Anfrage in einem Antwort-Stream. Während des Wartens kann es Updates im Updatestream geben.

Ich möchte diese Updates haben, wenn der Antwort-Stream abgeschlossen ist. Wie folgt:

Response (cold) |  x y z | 
Updates (hot)  | 1 2 3  4 | 

Result   |  x y z 123 4 | 

Ich kann nicht scheinen, es richtig zu machen. Gibt es eine clevere Mischung von Betreibern, die mir geben können, was ich brauche?

Answer 1

aus dem Diagramm Frage :

Puffern Sie alle Aktualisierungen, bis die Antwort abgeschlossen ist, und zeigen Sie dann den Rest der Aktualisierungen so an, wie er angezeigt wird. Antworten erscheinen sofort.

updates.TakeUntil(response.LastAsync()) 
     .ToArray() 
     .SelectMany(x => x) 
     .Concat(updates) 
     .Merge(response); 

Answer 2

Dies funktioniert, obwohl die Tatsache, dass Sie es nicht als mich stört Einzeiler tun können:

ist

var updatesReplayed = updates.Replay(); 
updatesReplayed.Connect(); 

var results = response.Concat(updatesReplayed); 

Tests hier die Funktionalität demonstriert:

// time    | 1 2 3 4 5 6 7 8 9 
// Response(cold)  |  x y z | 
// Updates(hot)  | 1 2 3  4 | 
// Result    |  x y z 123 4 | 

var scheduler = new TestScheduler(); 
    var updates = scheduler.CreateHotObservable(
     ReactiveTest.OnNext(100.ToMsTicks(), "1"), 
     ReactiveTest.OnNext(200.ToMsTicks(), "2"), 
     ReactiveTest.OnNext(400.ToMsTicks(), "3"), 
     ReactiveTest.OnNext(800.ToMsTicks(), "4"), 
     ReactiveTest.OnCompleted<string>(900.ToMsTicks()) 
    ); 

    var response = scheduler.CreateColdObservable(
     ReactiveTest.OnNext(300.ToMsTicks(), "x"), 
     ReactiveTest.OnNext(400.ToMsTicks(), "y"), 
     ReactiveTest.OnNext(500.ToMsTicks(), "z"), 
     ReactiveTest.OnCompleted<string>(600.ToMsTicks()) 
    ); 

    var expectedResults = scheduler.CreateHotObservable(
     ReactiveTest.OnNext(300.ToMsTicks(), "x"), 
     ReactiveTest.OnNext(400.ToMsTicks(), "y"), 
     ReactiveTest.OnNext(500.ToMsTicks(), "z"), 
     ReactiveTest.OnNext(600.ToMsTicks(), "1"), 
     ReactiveTest.OnNext(600.ToMsTicks(), "2"), 
     ReactiveTest.OnNext(600.ToMsTicks(), "3"), 
     ReactiveTest.OnNext(800.ToMsTicks(), "4"), 
     ReactiveTest.OnCompleted<string>(900.ToMsTicks()) 
    ); 

    var replayed = updates.Replay(); 
    replayed.Connect(); 

    var results = response.Concat(replayed); 
    var observer = scheduler.CreateObserver<string>(); 
    results.Subscribe(observer); 

    scheduler.Start(); 
    ReactiveAssert.AreElementsEqual(expectedResults.Messages, observer.Messages);