c# - 学习 Rx : How to use . 扫描 .Window 的输出以获取可观察的 bool 值序列

Question

我有一系列像这样的真假值

        var alternatingTrueFalse = Observable.Generate(
            true,
            _ => true,
            x => !x,
            x => x,
            _ => TimeSpan.FromMilliseconds(new Random().Next(2000)))
            .Take(20).Publish();
        alternatingTrueFalse.Connect();
        var buffered = alternatingTrueFalse
            .Buffer(TimeSpan.FromMilliseconds(500))
            .Subscribe(x => Console.WriteLine($"{TimeStamp} {string.Join(",", x)}"));

我想根据 500 毫秒(最大)窗口/缓冲区来查看序列。如果在一个这样的窗口中只有一个真值(没有别的)，我想拨动一个开关(只需调用一个命令，暂时打印到控制台)。然后，当下一个假值到达时，我想将开关拨回去并关闭原始序列的当前窗口/缓冲区并开始一个新的。

使用Buffer + Scan来翻转开关

到目前为止，我已经想出了一种在 Buffer 上执行此操作的方法。但是，缓冲区打开的时间太长，它们总是 500 毫秒。

        var buffered = alternatingTrueFalse
            .Buffer(TimeSpan.FromMilliseconds(500));
        var output = buffered
            .Subscribe(x => Console.WriteLine($"{TimeStamp} {string.Join(",", x)}"));

        var isFlipped = buffered.Scan(false, 
                (x, y) => 
                { 
                    if (y.Count == 0)
                    {
                        return x;
                    }
                    return y.Count == 1 && y.First();
                });

        isFlipped.DumpTimes("Flipped");

我正在尝试弄清楚如何使用 Window 而不是 Buffer 以便能够在孤立的 true 之后将开关翻转回第一个 false。但我似乎做不对，我对 Rx 还不是很流利，也不确定如何为它使用 windowOpening/Closing 值。

示例输出

原创

2017-10-07 20:21:39.302 True,False   // Rapid values should not flip the switch (actually they should flip a different switch)
2017-10-07 20:21:39.797 True         // Flip the switch here
2017-10-07 20:21:40.302 False        // Flip the switch back and close the current window
2017-10-07 20:21:40.797 True         // Flip the switch here
2017-10-07 20:21:41.297 
2017-10-07 20:21:41.798 False        // Etc...
...
2017-10-07 20:21:43.297 True
2017-10-07 20:21:43.800 False,True   // Example of a window that is open too long, because it is not closed immediately upon the false value
...

缓冲+扫描

2017-10-07 20:47:15.154 True
2017-10-07 20:47:15.163 - Flipped-->True
2017-10-07 20:47:15.659 False,True   // Example of a window open too long
2017-10-07 20:47:15.661 - Flipped-->False

Answer 1

这是一个不使用 Scan 方法的解决方案。

问题似乎是根据两个条件关闭缓冲区 - 最长时间或特定值。这个是基于an old answer

public static IObservable<IList<TSource>> BufferWithClosingValue<TSource>(
    this IObservable<TSource> source, 
    TimeSpan maxTime, 
    TSource closingValue)
{
    return source.GroupByUntil(_ => true,
                               g => g.Where(i => i.Equals(closingValue)).Select(_ => Unit.Default)
                                     .Merge(Observable.Timer(maxTime).Select(_ => Unit.Default)))
                 .SelectMany(i => i.ToList());
}

示例用法是

alternatingTrueFalse.BufferWithClosingValue( TimeSpan.FromMilliseconds(500), false );