反应式扩展超时不会停止序列？

我正在尝试做一个IObservable<bool>返回true如果在过去 5 秒内收到 UDP 消息并且发生超时，则返回 false。

到目前为止我有这个：

public IObservable<Boolean> GettingUDPMessages(IPEndPoint localEP)
{
    var udp = BaseComms.UDPBaseStringListener(localEP)
        .Where(msg => msg.Data.Contains("running"))
        .Select(s => true);

    return Observable
        .Timeout(udp, TimeSpan.FromSeconds(5))
        .Catch(Observable.Return(false));
}

与此相关的问题是：-

• 一旦返回 false，序列就会停止
• 我只真正需要true or false关于状态变化。

我可以用一个Subject<T>但我需要小心处理UDPBaseStringListener当没有更多订阅者时可以观察到。

Update

每次我收到 UDP 消息时，我希望它返回一个true。如果我在过去 5 秒内没有收到 UDP 消息，我希望它返回一个false.

正如所指出的Bj0 https://stackoverflow.com/a/19671744/852829，解与BufferWithTime接收到数据点后不会立即返回，并且接收到数据点后缓冲区超时不会重置。

使用 Rx Extensions 2.0，您可以通过接受超时和大小的新缓冲区重载来解决这两个问题：

static IObservable<Boolean> GettingUDPMessages(IPEndPoint localEP)
{
    return BaseComms
        .UDPBaseStringListener(localEP)
        .Where(msg => msg.Data.Contains("running"))
        .Buffer(TimeSpan.FromSeconds(5), 1)
        .Select(s => s.Count > 0)
        .DistinctUntilChanged();
}