无效扩展：直到其他流完成缓冲

Question

我发出请求并等待响应流上的该请求的响应。在等待时，更新流可能会有更新。

我想在响应流完成时进行这些更新。就像这样：

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

Result   |  x y z 123 4 | 

我似乎无法得到它的权利。是否有一些聪明的运营商组合可以给我我需要的东西？

Answer 1

从问题的图：

缓存所有更新，直到响应完成，然后显示其他更新。 响应立即显示。

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

Answer 2

这将工作，但事实上，你不能做到这一点作为一个班轮困扰我：

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

var results = response.Concat(updatesReplayed); 

这里的测试演示功能：

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