RxNET で RetryWhen #C#

RxNET に RetryWhen が見当たりません...¹
条件付きリトライができずに困ったので自作してみました。

拡張メソッド

IObservable を返すC#の拡張メソッドとやらを実装してメソッドチェーンを維持します。

RetryWhenExtension.cs

using System;
using System.Reactive.Linq;

namespace RxNET.Extensions
{
    public static class RetryWhenExtension
    {
        public static IObservable<T> RetryWhen<T>(
            this IObservable<T> source,
            Func<IObservable<Exception>, IObservable<T>> predicate)
        {
            return RetryWhenRecursive(source, predicate);
        }

        private static IObservable<T> RetryWhenRecursive<T>(
            IObservable<T> source,
            Func<IObservable<Exception>, IObservable<T>> predicate)
        {
            return source.Catch((Exception e) =>
            {
                //
                // シーケンスから例外が Throw された場合、
                // predicate でリトライするかどうかを判定する。
                //
                return
                     predicate(Observable.Return(e))
                    .Catch((Exception ee) =>
                     {
                        //
                        // predicate から例外が Throw された場合、
                        // 後続のシーケンスに例外を流す（＝リトライせずオブザーバーに例外を捕捉させる）。
                        //
                        return Observable.Throw<T>(e);
                     })
                    .SelectMany(_ =>
                     {
                        //
                        // predicate から例外が Throw されなかった場合、
                        // 再度、起点となったシーケンスに繋げる（＝リトライ）。
                        //
                        return RetryWhenRecursive(source, predicate);
                     });
            });
        }
    }
}

試してみる

5回ごとにしか成功しないメソッドを、1秒おきに最大10回リトライします。

Progran.cs

using System;
using System.Linq;
using System.Reactive.Linq;

using RxNET.Extensions;

namespace RetryApp
{
    class MainClass
    {
        const int RETRY_MAX  = 10;
        const int RETRY_WAIT = 1;

        public static void Main(string[] args)
        {
            var service = new GreetingService();

            int retryAttempt = 0;

            service.Greet()
                   .RetryWhen(errors => {
                        return errors.SelectMany(error => {
                            if (error is NeedToRetryException && ++retryAttempt <= RETRY_MAX) {
                                return Observable.Return(String.Empty).Delay(TimeSpan.FromSeconds(RETRY_WAIT));
                            }
                            return Observable.Throw<string>(error);
                        });
                    })
                   .Subscribe(
                        greet => Console.WriteLine("OnNext: {0}", greet),
                        error => Console.WriteLine("OnError: {0}", error.Message),
                        ()    => Console.WriteLine("OnCompleted:")
                  );

            Console.ReadKey();
        }
    }

    class GreetingService
    {
        int attempt = 0;

        public IObservable<string> Greet()
        {
            return Observable.Defer(() => {
                Console.WriteLine("Defer [{0}]", ++attempt);
                if (attempt % 5 == 0) {
                    return Observable.Return("Hello");
                }
                return Observable.Throw<string>(new NeedToRetryException("Zzz..."));
            });
        }
    }

    class NeedToRetryException : Exception
    {
        public NeedToRetryException() : base() { }
        public NeedToRetryException(string message) : base(message) { }
        public NeedToRetryException(string message, Exception inner) : base(message, inner) { }
    }
}

これを実行してみた結果が↓です。
5回目（リトライ4回目）で成功して、処理完了しています。

Defer [1]
Defer [2]
Defer [3]
Defer [4]
Defer [5]
OnNext: Hello
OnCompleted:

ちなみに RETRY_MAX を 3 にしてみると、

Defer [1]
Defer [2]
Defer [3]
Defer [4]
OnError: Zzz...

4回目（リトライ3回目）で、成功を待たずに、エラー終了しています。

どうやら期待通りに動いている感じです。

とりあえずは以上です。

今回作成したコード

こちらを参考にしました

http://ufcpp.net/study/csharp/sp3_extension.html
http://safx-dev.blogspot.jp/2015/12/rxswiftretryretrywhen.html

ReactiveX - Retry operator ↩