学习 Rx:如何在 .Window 的输出上使用 .Scan 以获得可观察的布尔值序列 [英] Learning Rx: How to use .Scan on the output of .Window for an observable sequence of bool values

问题描述

我有一个像这样的真假值序列

I have a sequence of true false values like so

        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 毫秒(最大)窗口/缓冲区来查看序列.如果在一个这样的窗口中只有一个真值(并且没有其他值)，我想翻转一个开关(现在只需调用一个命令，打印到控制台).然后，当下一个假值到达时，我想将开关翻转回来并关闭原始序列的当前窗口/缓冲区并开始一个新的.

I want to look at the sequence in terms of 500 ms (max) windows / buffers. If there is only one true value (and nothing else) inside one such window, I want to flip a switch (just call a command, print to console for now). Then, when the next false value arrives, I want to flip the switch back and close the current window / buffer of the original sequence and start a new one.

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

So far I've come up with a way to do this on a Buffer, with this. However, the buffers are open too long, they are always 500 ms.

        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 值.

I'm trying to figure out how to use Window instead of Buffer to be able to flip the switch back on the first false after an isolated true. But I can't seem to get it right, I'm not quite fluent in Rx yet and not sure how to use the windowOpening / Closing values for it.

原创

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

推荐答案

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

Here's a solution not using the Scan approach.

问题似乎是基于两个条件关闭缓冲区 - 最大时间或特定值.这个基于一个旧答案

The issue seems to be closing the buffer based on two conditions - maximum time or specific value. This one is based on 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 );

这篇关于学习 Rx:如何在 .Window 的输出上使用 .Scan 以获得可观察的布尔值序列的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！