检测 IObservable 上的 IsAlive

Question

我正在编写一个函数 IsAlive 来获取 IObservable 和时间跨度，并返回 IObservable 规范使用案例是检测流媒体服务器是否仍在发送数据。

我为此提出了以下解决方案，但感觉它的工作原理并不是最清楚的。

public static IObservable<bool> IsAlive<T>(this IObservable<T> source, 
                                           TimeSpan timeout, 
                                           IScheduler sched)
{
    return source.Window(timeout, sched)
                 .Select(wind => wind.Any())
                 .SelectMany(a => a)
                 .DistinctUntilChanged();
}

有人有更好的方法吗？

供参考 - 以下是我尝试过的单元测试和现有方法： https://gist.github.com/997003

Answer 1

这应该可行：

public static IObservable<bool> IsAlive<T>(this IObservable<T> source, 
                                           TimeSpan timeout, 
                                           IScheduler sched)
{
    return source.Buffer(timeout, 1, sched)
                 .Select(l => l.Any())
                 .DistinctUntilChanged();
}

这种方法也具有语义意义。每次有一个项目进入时，它都会填充缓冲区，然后传递 true。每次超时，都会创建一个空缓冲区并传递 false。

编辑：

这就是为什么 buffer-1 方法比窗口更好的原因：

var sched = new TestScheduler();
var subj = new Subject<Unit>();

var timeout = TimeSpan.FromTicks(10);

subj
    .Buffer(timeout, 1, sched)
    .Select(Enumerable.Any)
    .Subscribe(x => Console.WriteLine("Buffer(timeout, 1): " + x));

subj
    .Window(timeout, sched)
    .Select(wind => wind.Any())
    .SelectMany(a => a)
    .Subscribe(x => Console.WriteLine("Window(timeout): "+x));

sched.AdvanceTo(5);
subj.OnNext(Unit.Default);
sched.AdvanceTo(16);

产量：

Buffer(timeout, 1): True
Window(timeout): True
Buffer(timeout, 1): False

具体来说，窗口在整个超时时间内打开，并且不会在项目进入后立即关闭和重置这就是缓冲区限制 1 发挥作用的地方。一旦有项目进入，缓冲区及其计时器就会重新启动。

我可以将缓冲区重新实现为窗口，因为缓冲区的实现是一个窗口，但是a）我认为缓冲区具有更好的语义意义，b）我不必选择很多。 Scott 的 Select 和 SelectMany 可以组合成一个 SelectMany(x => x.Any())，但我可以避免整个 lambda 并指定 Enumerable.Any 方法组，无论如何，这都会更快地绑定（微不足道）。

Answer 2

怎么样：

source.Select(_ => true)
    .Timeout(timeout, sched)
    .DistinctUntilChanged()
    .Catch<bool, TimeoutException>)(ex => Observable.Return(false));