Rx Buffer 运行一次并完成答案

【问题标题】：Rx Buffer running once and completeRx Buffer 运行一次并完成
【发布时间】：2020-01-27 13:29:24
【问题描述】：

我喜欢在 3 秒内收集流的第一个值并将它们连接到相同数据类型的另一个流。第一个 Observable 是一个单独的、已完成的数据块，无论如何都应该完成，这样我就可以使用 Concat() 连接第一个和第二个。需要使用 Concat() 来保持数据流的完整性。

            private IObservable<3DPoints> _1stBuffer = Observable.Empty<3DPoints>();
            ..

           _1stBuffer = someRawStreamObservableReceivingOnNext
                .Buffer(TimeSpan.FromSeconds(3), 100)
                .Where(item => item.Any())
                .SelectMany(item => item);


// later 
            var streamObservable = _1stBuffer.Concat(_some2ndStream); // doesn't emit, since 1stBuffer doesn't complete

我试过了：

            _1stBuffer = someRawStreamObservableReceivingOnNext

                // completes the observable, but i want the buffer so far, no emptiness !
                .TimeOut(TimeSpan.FromSeconds(3), Observable.Empty<3DPoints>)

                .Buffer(TimeSpan.FromSeconds(3), 100)
                .Where(item => item.Any())
                .SelectMany(item => item);

超时使 bufferObservable 执行 Concat()，但继续 Observable.Empty 是没有选择的。如何获取填充了 3 秒的第一个缓冲区并连接它们？

【问题讨论】：

标签： c# reactive-programming system.reactive reactiveui

【解决方案1】：

您可能正在寻找Take 变体之一：

source.Take(1) 将从源中获取第一个元素，然后完成。
source.TakeUntil(otherObservable) 将从源获取直到 otherObservable 的第一个元素发出。

【讨论】：