system.reactive - 在 IObservable 上检测 IsAlive

Question

我正在写一个函数 IsAlive采取IObservable<T> , 和时间跨度，并返回 IObservable<bool>规范用例是检测流服务器是否仍在发送数据。

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

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)我不必 SelectMany。 Scott 的 Select 和 SelectMany 可以组合成一个 SelectMany(x => x.Any())，但我可以避免使用整个 lambda 并指定 Enumerable.Any 方法组，这无论如何都会更快(微不足道)绑定(bind)。