当没有值传入时，是否有 Rx 方法来定期重复先前的值? [英] Is there a Rx method to repeat the previous value periodically when no values are incoming?

问题描述

我遇到过的一个用例，我怀疑我不可能是唯一的用例，它是用于如下方法:

A use case which I have encountered, and I suspect I can't be the only one, is for a method like:

IObservable<T> Observable.RepeatLastValueDuringSilence(this IObservable<T> inner, TimeSpan maxQuietPeriod);

这将从内部 observable 返回所有未来的项目，而且，如果内部 observable 在一段时间内(maxQuietPeriod)没有调用 OnNext，它只会重复最后一个值(直到内部调用 OnCompleted或 OnError).

which would return all the future items from the inner observable, but also, if the inner observable doesn't call OnNext for a certain period of time (maxQuietPeriod), it just repeats the last value (until of course inner calls OnCompleted or OnError).

服务定期发出定期状态更新的理由是.例如:

A justification would be for a service to periodically ping out a periodic status update. For example:

var myStatus = Observable.FromEvent(
    h=>this.StatusUpdate+=h,
    h=>this.StatusUpdate-=h);

var messageBusStatusPinger = myStatus
    .RepeatLastValueDuringSilence(TimeSpan.FromSeconds(1))
    .Subscribe(update => _messageBus.Send(update));

这样的东西存在吗?还是我高估了它的用处?

Does something like this exist? Or am I over-estimating it's usefulness?

谢谢，亚历克斯

PS:对于任何不正确的术语/语法，我深表歉意，因为我只是第一次探索 Rx.

PS: I apologise for any incorrect terminology/syntax, as I'm only just exploring Rx for the first time.

推荐答案

与 Matthew 的解决方案类似，但这里的计时器在源中收到每个元素后启动，我认为这更正确(但是差异不太可能重要):

Similar solution to Matthew's, but here the timer starts after each element is received in the source, which I think is more correct (however the differences are unlikely to matter):

public static IObservable<T> RepeatLastValueDuringSilence<T>(this IObservable<T> inner, TimeSpan maxQuietPeriod)
{    
    return inner.Select(x => 
        Observable.Interval(maxQuietPeriod)
                  .Select(_ => x)
                  .StartWith(x)
    ).Switch();
}

和测试:

var source = Observable.Interval(TimeSpan.FromMilliseconds(100)).Take(5).Select(_ => "1")
                       .Concat(Observable.Interval(TimeSpan.FromSeconds(1)).Take(5).Select(_ => "2"))
                       .Concat(Observable.Interval(TimeSpan.FromMilliseconds(100)).Take(5).Select(_ => "3"));

source.RepeatLastValueDuringSilence(TimeSpan.FromMilliseconds(200)).Subscribe(Console.WriteLine);

您应该看到 1 打印了 10 次(5 次来自源代码，5 次在静音期间重复)，然后是大量 2 从源代码中获得的，还有 4 次来自每个之间的沉默，然后是无限的3.

You should see 1 printed 10 times (5 from source, 5 repeated during silence), then lots of 2 as you get the one from source and 4 more from silence between each, followed by infinite 3.

这篇关于当没有值传入时，是否有 Rx 方法来定期重复先前的值?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！