使用延迟计时器对反应式扩展方法进行单元测试

Question

我有以下扩展方法。

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 是另一种扩展方法，我可以很好地进行单元测试。上述方法的问题是因为我返回Observable.Timer 断言被调用，然后委托第二次继续。

单元测试方法。

        [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.
        }

AdvanceByMinimal 辅助方法。

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

RetryX 扩展方法的成功单元测试如下。

        [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);
        }

为了清晰起见，下面是 RetryX 扩展方法。

        /// <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

这不是您的问题的答案，而是可能对您有所帮助的东西：我看了一段时间 RetryX，如果您去掉所有您可能应该的 scheduler 东西，它可以减少这个：

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>()));
}

所有调度程序调用都不是“最佳实践”。大多数 Rx 操作员不接受调度程序参数是有原因的（Select、Where、Catch 等）。那些与时间/调度有特定关系的人：Timer、Delay、Join。

有兴趣指定调度程序与无调度程序RetryX 一起使用的人总是可以在传入的参数上指定调度程序。您通常希望线程管理位于顶级调用线程，并指定线程调度不是你想做的。

Answer 2

George 查看 Kent 的 https://github.com/kentcb/Genesis.RetryWithBackoff 以获得一些灵感。

Answer 3

问题是没有将 IScheduler 正确传递给 RetryX 扩展方法以及 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);
        }