如何基于 Func 将 IObservable 窗口/缓冲为块

给定一个类：

class Foo { DateTime Timestamp {get; set;} }

...和IObservable<Foo>，有保证的单调递增 Timestamps，我怎样才能生成一个IObservable<IList<Foo>>根据这些分块到列表中Timestamps?

IE。每个IList<Foo>应该有五秒钟的事件，或者其他什么。我知道我可以使用Buffer with a TimeSpan超载，但我需要从事件本身而不是挂钟中获取时间。 （除非有一种巧妙的方法来提供IScheduler这里使用的是IObservable本身作为源.Now?)

如果我尝试使用Observable.Buffer(this IObservable<Foo> source, IObservable<Foo> bufferBoundaries)像这样重载：

IObservable<Foo> foos = //...;
var pub = foos.Publish();
var windows = pub.Select(x => new DateTime(
    x.Ticks - x.Ticks % TimeSpan.FromSeconds(5).Ticks)).DistinctUntilChanged();

pub.Buffer(windows).Subscribe(x => t.Dump()));  // linqpad
pub.Connect();

...然后IList实例包含导致窗口关闭的项目，但我真的希望这个项目进入下一个窗口/缓冲区。

例如。带时间戳[0, 1, 10, 11, 15]你会得到块[[0], [1, 10], [11, 15]]代替[[0, 1], [10, 11], [15]]

这是一个想法。组关键条件是“窗口号”，我使用GroupByUntil。这为您提供了示例中所需的输出（我使用了一个 int 流，就像该示例一样 - 但您可以替换为窗口编号所需的任何内容）。

public class Tests : ReactiveTest
{
    public void Test()
    {
        var scheduler = new TestScheduler();
        var xs = scheduler.CreateHotObservable<int>(
            OnNext(0, 0),
            OnNext(1, 1),
            OnNext(10, 10),
            OnNext(11, 11),
            OnNext(15, 15),
            OnCompleted(16, 0));                  
            
        xs.Publish(ps =>                                // (1)
            ps.GroupByUntil(
                p => p / 5,                             // (2)
                grp => ps.Where(p => p / 5 != grp.Key)) // (3)
            .SelectMany(x => x.ToList()))               // (4)
        .Subscribe(Console.WriteLine);
        
        scheduler.Start();
    }
}

Notes

1. 我们发布源流是因为我们将订阅多次。
2. 这是创建组密钥的函数 - 使用它根据您的项目类型生成窗口号。
3. 这是组终止条件 - 使用它来检查源流中另一个窗口中的项目。请注意，这意味着窗口不会关闭，直到窗口外部的元素到达或源流终止。如果您考虑一下，这是显而易见的 - 您想要的输出需要考虑窗口结束后的下一个元素。请注意，如果您的源与实时有任何关系，您可以将其与Observable.Timer+Select输出术语的空/默认实例以提前终止流。
4. SelectMany 将组放入列表并展平流。

如果包含 nuget 包 rx-testing，此示例将在 LINQPad 中运行得很好。新建一个测试实例并运行Test() method.