アクションを再試行する方法OnError reactive X IObservableを複数回実行しても成功する場合

Question

次の問題があります。新しいイベントを通知するためにOnNextメソッドを使用する件名があります。ただし、OnErrorというメソッド呼び出しでコード内でシミュレートされる例外が発生することがあります。エラーの原因は時間に依存しますので、同じアクションをもう一度やり直すと数回は成功します。ですから、新しいシーケンスを生成する関数を受け入れるcatchメソッドを使用して、アクションを5回繰り返します。すべての繰り返し試行が失敗した場合、最終的な例外がスローされ、少なくとも1つが成功した場合は、OnNext（5）およびOnNext（6）を続行する必要があります。

例外的に、その後のOnError呼び出しを防止するのが理想的ですが、私の場合はそうではありません。

リトライ、コンカット、キャッチなどの方法で複数のシナリオを試しましたが、私が望むように何もできませんでした。

Subject<int> sub = new Subject<int>(); 

var seq = sub.Select(x => 
{ 
    //time dependent operation 
    Console.WriteLine(x); 
    return x; 
}). 
Catch((SeqException<int> ex) => 
{ 
    return Observable.Empty(0); // what sequece to return to achieve the desired behaviour 
}); 
seq.Subscribe(); 

sub.OnNext(1); 
sub.OnNext(2); 
sub.OnNext(3); 
sub.OnError(new SeqException<int>{ Value = 4}); 
sub.OnNext(5); 
sub.OnNext(6); 

seq.Wait(); 

ありがとうございます。

Answer 1

ケース1：Select内の時間依存操作が再試行する必要がある場合。

Observable.Startを使用すると、サンクを観測可能なものに変換できます。その後、すべての回復演算子を使用して行動を宣言します。

Subject<int> sub = new Subject<int>();  

var seq = sub.SelectMany(x =>    
     Observable.Start(() => 
     { 
      //time dependent failure 
      if (DateTime.Now.Second % 2 == 0) 
       throw new Exception(); 

      Console.WriteLine(x); 
      return x; 
     }) 
     .Retry(5) 
     .Catch(Observable.Return(x)) 
    ); 

//for testing 
Observable.Interval(TimeSpan.FromSeconds(1)).Select(x => (int)x).Subscribe(sub); 

seq.Wait(); 

---

ケース2：あなたが実際にSubjectを通じて知らさなっています。

Rxコントラクトでは、OnComplete | OnErrorの後に通知がなくなる必要があります。そのエラーを受信すると、サブジェクトはすべてのサブスクリプションを破棄し、パイプラインは切断されます。それを動作させるには、ケース1に持ち込む必要があります。

Answer 2

いつでもリトライブロックで何かをラップすることができます。 1つの引数を持つアクションの例です。

Action<T> RetryAction<T>(Action<T> action, int retries) 
{ 
    return arg => { 
     int count = 0; 
     while (true) 
     { 
      try 
      { 
       action(arg); 
       return; 
      } 
      catch 
      { 
       if (count == retries) 
        throw; 
       count++; 
      } 
     } 
    } 
} 

したがって、シーケンス処理のロジックを変更する必要はありません。