单元测试带有延迟计时器的反应式扩展方法

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 是另一种扩展方法，我可以很好地进行单元测试。上述方法的问题是因为我 return 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

问题是没有将 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);
        }

Answer 2

这不是您问题的答案，而是可以帮助您的东西：我看了 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>()));
}

并非所有调度程序调用都是 "best practice"。大多数 Rx 操作员不接受调度程序参数（Select、Where、Catch 等）是有原因的。那些确实与 timing/scheduling 有特定关系的：Timer、Delay、Join.

有兴趣指定与 scheduler-less RetryX 一起使用的调度程序的人总是可以在传入的参数上指定调度程序。您通常希望 thread-management 在top-level 调用线程，指定线程调度不是你想要的。

Answer 3

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

单元测试带有延迟计时器的反应式扩展方法

Unit test a reactive extension method with delay timer

c#

unit-testing

system.reactive

reactiveui

reactivex