c# - 不停止序列的 react 性扩展超时？

Question

我正在尝试制作一个 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或 false关于状态变化。

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

更新

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

Answer 1

正如 Bj0 指出的那样，带有 BufferWithTime 的解决方案不会在接收到数据点后立即返回数据点，并且在接收到数据点后不会重置缓冲区超时。

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

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();
}