c# - 当没有值传入时，是否有 Rx 方法定期重复以前的值？

Question

我遇到过的一个用例，我怀疑我不是唯一的一个，是一个像这样的方法:

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

这将从内部可观察对象返回所有 future 的项目，而且，如果内部可观察对象在一段时间内(maxQuietPeriod)没有调用 OnNext，它只会重复最后一个值(当然直到内部调用 OnCompleted或 OnError)。

一个合理的理由是服务定期 ping 出定期状态更新。例如:

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

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

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

谢谢，亚历克斯

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

Answer 1

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

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。