c# - 响应式订阅在一段时间内没有跟随不同事件的事件

Question

我正在尝试弄清楚如何为特定时间窗口内没有跟随不同事件的事件创建响应式订阅。

为了说明，这里有一个用例:

由两个事件触发的忙碌指示器。忙和不忙。它们可能会靠在一起发射，但指示灯不应经常闪烁。

当 NotBusy 触发时，应该没有指示符。当 Busy 触发并且 NotBusy 在 5 秒内没有触发时，它应该显示。

有没有办法在不添加外部状态的情况下完全在 Reactive 中完成此操作？

Answer 1

啊，我无法抗拒 - 你多久反驳原始资料的作者一次？ (参见 Benjol 对问题的评论) :)

这是我的尝试(LINQPad-ready):

输出看起来像这样:

At 1/1/0001 12:00:00 AM +00:00, Busy Signal Indicator is now:OFF
Sending BUSY at 1/1/0001 12:00:00 AM +00:00
Sending NOTBUSY at 1/1/0001 12:00:02 AM +00:00

Sending BUSY at 1/1/0001 12:00:03 AM +00:00
Sending BUSY at 1/1/0001 12:00:06 AM +00:00
At 1/1/0001 12:00:08 AM +00:00, Busy Signal Indicator is now:ON
Sending NOTBUSY at 1/1/0001 12:00:09 AM +00:00
At 1/1/0001 12:00:09 AM +00:00, Busy Signal Indicator is now:OFF

这里是我们事件的基本定义:

enum EventType 
{ 
    Busy, 
    NotBusy 
}
class StreamEvent 
{ 
     public EventType Type {get; set;} 
     public StreamEvent(EventType type) { Type = type;}
}

这是查询+测试代码:

void Main()
{
    // our simulated event stream
    var fakeSource = new System.Reactive.Subjects.Subject<StreamEvent>();

    // Let's use a scheduler we actually don't have to wait for
    var theTardis = new System.Reactive.Concurrency.HistoricalScheduler();

    var busySignal = fakeSource
        // Batch up events:
        .Window(
            // Starting batching on a busy signal
            fakeSource.Where(e => e.Type == EventType.Busy),
            // Stop batching on a not busy signal
            (open) => fakeSource.Where(e => e.Type == EventType.NotBusy)
                // but throw a timeout if we exceed 5 seconds per window
                .Timeout(TimeSpan.FromSeconds(5), theTardis))       
        // Unpack the windows
        .Switch()
        // Catch any timeout exception and inject a NULL into the stream        
        .Catch(fakeSource.StartWith((StreamEvent)null))
        // Bool-ify on "did a timeout happen?"
        .Select(e => e == null)
        // Start in an "unbusy" state
        .StartWith(false)
        // And only tell us about transitions
        .DistinctUntilChanged();    

    using(busySignal.Subscribe(signal => 
        Console.WriteLine("At {0}, Busy Signal Indicator is now:{1}",
            theTardis.Now,
            signal ? "ON" : "OFF")))
    {
        // should not generate a busy signal
        Console.WriteLine("Sending BUSY at {0}", theTardis.Now);
        fakeSource.OnNext(new StreamEvent(EventType.Busy));
        theTardis.AdvanceBy(TimeSpan.FromSeconds(2));
        Console.WriteLine("Sending NOTBUSY at {0}", theTardis.Now);
        fakeSource.OnNext(new StreamEvent(EventType.NotBusy));
        theTardis.AdvanceBy(TimeSpan.FromSeconds(1));
        Console.WriteLine();

        // should generate a busy signal
        Console.WriteLine("Sending BUSY at {0}", theTardis.Now);
        fakeSource.OnNext(new StreamEvent(EventType.Busy));
        theTardis.AdvanceBy(TimeSpan.FromSeconds(3));
        Console.WriteLine("Sending BUSY at {0}", theTardis.Now);
        fakeSource.OnNext(new StreamEvent(EventType.Busy));
        theTardis.AdvanceBy(TimeSpan.FromSeconds(3));

        // and this should clear it
        Console.WriteLine("Sending NOTBUSY at {0}", theTardis.Now);
        fakeSource.OnNext(new StreamEvent(EventType.NotBusy));      
        theTardis.AdvanceBy(TimeSpan.FromSeconds(1));
        Console.WriteLine();    
    }
}