c# - 如何在 Rx 中交替缓冲和流动实时数据流

Question

我有两个流。一个是数据流(可以是任何类型)，另一个是充当门的 bool 流。我需要将它们组合成具有以下行为的流:

• 当门打开时(最近的值是真的)然后数据应该直接流过
• 当门关闭时(最近的值是假的)然后数据应该被缓冲以作为单个元素释放时下一次开门
• 解决方案应保留数据的所有元素并保留订单

我不太确定如何将它们放在一起。我一直在测试的输入是这样的:

// a demo data stream that emits every second
var dataStream = Observable.Interval(TimeSpan.FromSeconds(1));

// a demo flag stream that toggles every 5 seconds
var toggle = false;
var gateStream = Observable.Interval(TimeSpan.FromSeconds(5))
    .Select(_ => toggle = !toggle);

Answer 1

我会这样做:

• 使用门流作为结束选择器对数据流进行窗口化
• 我们可以在门流上使用DistinctUntilChanged来确保没有重复值
• 我们还将强制门流开始关闭(false)——它不会影响输出并允许一个巧妙的技巧
• 然后使用 Select 的重载，为每个元素提供一个索引号。有了这个我们就可以判断我们是否需要缓冲或只按原样发射窗口，因为我们知道偶数编号的窗口用于缓冲(因为我们确保门流以 false 开头)
• 我们可以使用 ToList() 来缓冲每个偶数窗口直到它关闭 - 这实际上等同于 Buffer() 一直等到 OnCompleted
• 我们使用身份 SelectMany 来展平缓冲窗口
• 最后我们连接窗口以保证保留顺序

看起来像这样:

dataStream.Window(gateStream.StartWith(false).DistinctUntilChanged())
          .Select((w, i) =>  i % 2 == 0 ? w.ToList().SelectMany(x => x) : w)
          .Concat()
          .Subscribe(Console.WriteLine);