RX观察到如果某些超时它出版的值 [英] Rx observable which publishes a value if certain timeout expires

问题描述

我有一个返回一个可观察的方法。这应该观察到（如果evetything正在右）发布的值每秒。我想这样做是有其发布一些自定义警戒值，如果在一定时间没有输出过去了。

I have a method that returns an observable. This observable should (if evetything is working right) publish a value each second. What I would like to do is have it publish some custom alert value if a certain time has passed with no output.

private IObservable<string> GetStatus()
{
    return statusProvider
                .Subscribe(statusKey)  //returns an IObservable<string>
                .Select(st => st.ToUpper())
                .DistinctUntilChanged()
                .TakeUntil(disposed)
                .Replay(1)
                .RefCount();
}

有我上面的修改一个简单的方法，这样，如果没有状态更新已经来到了30秒，statusProvider出版的坏，然后如果一个更新程序后，随之而来的，它得到出版和往常一样，定时器重新启动，再次30秒？

Is there a simple way for me to modify the above so that if no status update has come in for 30 seconds, the statusProvider publishes "bad" and then if an update does come in after that, it get published as usual and the timer is restarted to 30 secs again?

推荐答案

下面是一种方法。启动一个计时器，将产生坏何时到期。每次你statusProvider产生一个状态，计时器被重置。

Here is a way. Starts a timer which will yield "bad" when it expires. Each time your statusProvider produces a status, the timer gets reset.

var statusSignal = statusProvider
            .Subscribe(statusKey)  //returns an IObservable<string>
            .Select(st => st.ToUpper())
            .Publish()
            .RefCount();

// An observable that produces "bad" after a delay and then "hangs indefinately" after that
var badTimer = Observable
    .Return("bad")
    .Delay(TimeSpan.FromSeconds(30))
    .Concat(Observable.Never<string>());

// A repeating badTimer that resets the timer whenever a good signal arrives.
// The "indefinite hang" in badTimer prevents this from restarting the timer as soon
// as it produces a "bad".  Which prevents you from getting a string of "bad" messages
// if the statusProvider is silent for many minutes.
var badSignal = badTimer.TakeUntil(statusSignal).Repeat();

// listen to both good and bad signals.
return Observable
    .Merge(statusSignal, badSignal)
    .DistinctUntilChanged()
    .Replay(1)
    .RefCount();

这篇关于RX观察到如果某些超时它出版的值的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！