Unit test a reactive extension method with delay timer

Question

I have the below extension method.

public static IObservable<T> RetryWithCount<T>(this IObservable<T> source, 
            int retryCount, int delayMillisecondsToRetry, IScheduler executeScheduler = null,
            IScheduler retryScheduler = null)
        {
            var retryAgain = retryCount + 1;
            return source
                .RetryX(
                    (retry, exception) =>
                        retry == retryAgain
                            ? Observable.Throw<bool>(exception)
                            : Observable.Timer(TimeSpan.FromMilliseconds(delayMillisecondsToRetry))
                                .Select(_ => true));
        }

RetryX is another extension method and I can unit test fine. The problem with the above method is because I return Observable.Timer the assertion is invoked and then the delegate is continuing for the second time.

The Unit Test method.

        [Test]
        public void should_retry_with_count()
        {
            // Arrange
            var tries = 0;
            var scheduler = new TestScheduler();
            IObservable<Unit> source = Observable.Defer(() =>
            {
                ++tries;
                return Observable.Throw<Unit>(new Exception());
            });

            // Act
            var subscription = source.RetryWithCount(1, 100, scheduler, scheduler)
                .Subscribe(
                    _ => { },
                    ex => { });
            scheduler.AdvanceByMinimal(); //How to make sure that it is completed?

            // Assert
            Assert.IsTrue(tries == 2); // Assert is invoked before the source has completed.
        }

The AdvanceByMinimal helper method.

public static void AdvanceMinimal(this TestScheduler @this) => @this.AdvanceBy(TimeSpan.FromMilliseconds(1));

A successful unit test for the RetryX extensions method is below.

        [Test]
        public void should_retry_once()
        {
            // Arrange
            var tries = 0;
            var scheduler = new TestScheduler();
            var source = Observable
                .Defer(
                    () =>
                    {
                        ++tries;
                        return Observable.Throw<Unit>(new Exception());
                    });
            var retryAgain = 2;

            // Act
            source.RetryX(
                (retry, exception) =>
                {
                    var a = retry == retryAgain
                        ? Observable.Return(false)
                        : Observable.Return(true);

                    return a;
                }, scheduler, scheduler)
                .Subscribe(
                    _ => { },
                    ex => { });
            scheduler.AdvanceMinimal();

            // Assert
            Assert.IsTrue(tries == retryAgain);
        }

And for the clarity of overall picture below is the RetryX extension method.

        /// <summary>
        /// Retry the source using a separate Observable to determine whether to retry again or not.
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="source"></param>
        /// <param name="retryObservable">The observable factory used to determine whether to retry again or not. Number of retries & exception provided as parameters</param>
        /// <param name="executeScheduler">The scheduler to be used to observe the source on. If non specified MainThreadScheduler used</param>
        /// <param name="retryScheduler">The scheduler to use for the retry to be observed on. If non specified MainThreadScheduler used.</param>
        /// <returns></returns>
        public static IObservable<T> RetryX<T>(this IObservable<T> source,
            Func<int, Exception, IObservable<bool>> retryObservable, IScheduler executeScheduler = null,
            IScheduler retryScheduler = null)
        {
            if (retryObservable == null)
            {
                throw new ArgumentNullException(nameof(retryObservable));
            }

            if (executeScheduler == null)
            {
                executeScheduler = MainScheduler;
            }

            if (retryScheduler == null)
            {
                retryScheduler = MainScheduler;
            }

            // so, we need to subscribe to the sequence, if we get an error, then we do that again...
            return Observable.Create<T>(o =>
            {
                // whilst we are supposed to be running, we need to execute this
                var trySubject = new Subject<Exception>();

                // record number of times we retry
                var retryCount = 0;

                return trySubject.
                    AsObservable().
                    ObserveOn(retryScheduler).
                    SelectMany(e => Observable.Defer(() => retryObservable(retryCount, e))). // select the retry logic
                    StartWith(true). // prime the pumps to ensure at least one execution
                    TakeWhile(shouldTry => shouldTry). // whilst we should try again
                    ObserveOn(executeScheduler).
                    Select(g => Observable.Defer(source.Materialize)). // get the result of the selector
                    Switch(). // always take the last one
                    Do((v) =>
                    {
                        switch (v.Kind)
                        {
                            case NotificationKind.OnNext:
                                o.OnNext(v.Value);
                                break;

                            case NotificationKind.OnError:
                                ++retryCount;
                                trySubject.OnNext(v.Exception);
                                break;

                            case NotificationKind.OnCompleted:
                                trySubject.OnCompleted();
                                break;
                        }
                    }
                    ).Subscribe(_ => { }, o.OnError, o.OnCompleted);
            });
        }

Answer 1

George check out https://github.com/kentcb/Genesis.RetryWithBackoff by Kent for some inspiration.

Answer 2

This isn't an answer to your question, but rather something that may help you: I looked at that RetryX for a while, and if you strip out all the scheduler stuff, which you probably should, it can be reduced down this:

public static IObservable<T> RetryX<T>(this IObservable<T> source, Func<int, Exception, IObservable<bool>> retryObservable)
{
    return source.Catch((Exception e) => retryObservable(1, e)
        .Take(1)
        .SelectMany(b => b ? source.RetryX((count, ex) => retryObservable(count + 1, ex)) : Observable.Empty<T>()));
}

All the scheduler calls aren't "best practice". There's a reason most Rx operators don't accept scheduler parameters (Select, Where, Catch, etc.). The ones that do, have something specifically to do with timing/scheduling: Timer, Delay, Join.

Someone who is interested in specifying the scheduler to use with a scheduler-less RetryX could always specify the scheduler on the parameters passed in. You generally want thread-management to be at the top-level calling thread, and specifying thread scheduling isn't where you want to do it.

Answer 3

The problem was not passing the IScheduler correctly down to the RetryX extension method and also to Observable.Timer.

public static IObservable<T> RetryWithCount<T>(this IObservable<T> source, 
            int retryCount, int delayMillisecondsToRetry, IScheduler executeScheduler = null,
            IScheduler retryScheduler = null)
        {
            if (executeScheduler == null)
            {
                executeScheduler = MainScheduler;
            }
            var retryAgain = retryCount + 1;
            return source
                .RetryX(
                    (retry, exception) =>
                    {
                        return retry == retryAgain
                            ? Observable.Throw<bool>(exception, executeScheduler)
                            : Observable.Timer(TimeSpan.FromMilliseconds(delayMillisecondsToRetry), executeScheduler)
                                .Select(_ => true);
                    },
                    retryScheduler,
                    executeScheduler);
        }