Rx IsAlive UnitTests

IsAlives.cs

        // doesn't detect Never()
        public static IObservable<bool> IsAlive_Merge<T>(this IObservable<T> source, TimeSpan timeout, IScheduler sched)
        {
            return source.Select(_ => true)
            .Merge(source.Select(_ => false).Throttle(timeout, sched))
            .DistinctUntilChanged();
        }

        // Delay detection of Alive till timespan has expired
        public static IObservable<bool> IsAlive_Buffer<T>(this IObservable<T> source, TimeSpan timeout, IScheduler sched)
        {
            return source.Buffer(timeout, sched).Select(list => list.Count > 0).DistinctUntilChanged();
        }

        // via http://enumeratethis.com/2010/10/  doesn't detect Never()
        public static IObservable<bool> IsAlive_Switch<T>(this IObservable<T> source, TimeSpan timeout, IScheduler sched)
        {
            return Observable.Switch(
                                    from hb in source
                                    select Observable.Return(true).Concat(Observable.Return(false).Delay(timeout, sched))
                                    )
                                    .DistinctUntilChanged();
        }

        // works, but isn't the most clear
        public static IObservable<bool> IsAlive_Window<T>(this IObservable<T> source, TimeSpan timeout, IScheduler sched)
        {
            return source.Window(timeout, sched)
                         .Select(wind => wind.Any())
                         .SelectMany(a => a)
                         .DistinctUntilChanged();
        }

RxIsAliveUnitTests.cs

        [Fact]
        public void IsAlive_Is_False_On_NeverStream()
        {
            var source = Observable.Never<Unit>();

            var actual = _testScheduler.Start( () => source.IsAlive(TimeSpan.FromTicks(50), _testScheduler));

            actual.Messages.AssertEqual(OnNext(250, false));
        }

        [Fact]
        public void IsAlive_Is_Alive_On_Heartbeat()
        {
            var source = _testScheduler.CreateHotObservable(OnNext(600,1));

            var actual = _testScheduler.Start(() => source.IsAlive(TimeSpan.FromTicks(500), _testScheduler));

            actual.Messages.AssertEqual(OnNext(600, true));
        }

        [Fact]
        public void IsAlive_Detects_FirstHeartbeatAfterTimeout()
        {
            var source = _testScheduler.CreateHotObservable(OnNext(901, 1));

            var actual = _testScheduler.Start(() => source.IsAlive(TimeSpan.FromTicks(100), _testScheduler));

            actual.Messages.AssertEqual(OnNext(300, false), OnNext(901, true));
        }


        [Fact]
        public void IsAlive_Detects_StoppedHeartbeat()
        {
            var source = _testScheduler.CreateHotObservable(OnNext(300, 1));

            var actual = _testScheduler.Start(() => source.IsAlive(TimeSpan.FromTicks(100), _testScheduler));

            actual.Messages.AssertEqual(OnNext(300, true), OnNext(400, false));
        }

        [Fact]
        public void IsAlive_Detects_RestartedHeartbeat()
        {
            var source = _testScheduler.CreateHotObservable(OnNext(300, 1), OnNext(901, 1));

            var actual = _testScheduler.Start(() => source.IsAlive(TimeSpan.FromTicks(100), _testScheduler));

            actual.Messages.AssertEqual(OnNext(300, true), OnNext(400, false), OnNext(901, true));
        }