rx-开始对Onnext进行缓冲

Question

使用 Observable.Buffer(TimeSpan timeSpan) Method，将流分成10分钟的块并返回为 IList，效果很好

var stream = Observable.FromEventPattern<*>(*,*);
stream.Buffer(TimeSpan.FromSeconds(10));

试图实施更复杂的行为

• 开始块（缓冲事件列表）新事件被推到流中（而不是每10秒每10秒）
• 继续缓冲事件，直到没有将事件推入x秒
的流中

Answer 1

尝试以下：

var query =
    stream.Publish(
        ps => ps.Window(
            () => ps.Delay(TimeSpan.FromSeconds(1.0)).Take(1)));

Answer 2

我敢肯定有很多方法可以做到这一点。

我在这里有一个工作的测试示例

void Main()
{
    var scheduler = new TestScheduler();
    var stream = scheduler.CreateColdObservable(
        ReactiveTest.OnNext(1.Seconds(), 'A'),
        ReactiveTest.OnNext(2.Seconds(), 'B'),
        ReactiveTest.OnNext(13.Seconds(), 'C')
        );
    var observer = scheduler.CreateObserver<string>();
    var query = stream.Publish(s => {
        return s.Timeout(TimeSpan.FromSeconds(10), Observable.Empty<char>(), scheduler)
            .ToList()
            .Where(buffer=>buffer.Any())    
            //Project to string to make equality test easier for the example.       
            .Select(buffer=>string.Join(",", buffer))
            .Repeat();
    });
    query.Subscribe(observer);
    scheduler.AdvanceBy(100.Seconds());
    ReactiveAssert.AreElementsEqual(
        new []{
            ReactiveTest.OnNext(12.Seconds(), "A,B"),
            ReactiveTest.OnNext(23.Seconds(), "C")
        },
        observer.Messages);
}
// Define other methods and classes here
public static class TimeEx
{
    public static long Seconds(this int seconds)
    {
        return TimeSpan.FromSeconds(seconds).Ticks;
    }
}

在这里请注意，我只是将缓冲列表作为字符串，以使验证平等更容易。即"A,B"代替{'A', 'B'}

其他要考虑的选项是Window或GroupJoin操作员要这样做 - 请参见http：//www.introtorx.com/content/v1.0.0.10621.0/17_sequencesofcoincence.html。我敢肯定可以将Switch，Select，Timer，Timeout等缝合在一起。