Rx再試行（）が正常に動作しない

Question

Reactive Extensionsを試していて、RSSアイテムをたくさん取得しています。ティム・グリーンフィールドのブログ記事：Silverlight Rx DataClient within MVVMに基づいています。

私はデスクトップアプリケーション内で使用していますが、コードは似ています。

私が抱えている問題は、Retry()の機能を理解することです。それは私が期待していることをやっているとは思わないし、私が期待していることにもあるようだ。

var items = new List<RssItem>(); 
WebHelper.DownloadXmlFileAsync<RssItem>(new Uri(URI), "item") 
    .Retry(2) 
    .Finally(PublishResults) 
    .Subscribe(items.Add, ProcessError,() => ProcessCompleted(items)); 

有効なURIを渡すと、問題なく動作します。私がURIにタイプミスをすると、ProcessError()関数を介して404エラーが報告されますが、それは一度だけ報告されます。私はこのエラーを2回表示することを期待していました。

したがって、Retry()機能は私のWebリクエストでは動作していないようですが、実際にはSubscribe()に渡される機能に適用されるようです。私はここで間違っている可能性があります。

Retry()コールがウェブリクエストに適用されることを確認するにはどうすればよいですか？

エクストラコード：

public static class WebHelper 
{ 
    public static HttpWebRequest CreateHttp(Uri uri) 
    { 
     return CreateHttp(uri, "GET"); 
    } 

    public static HttpWebRequest CreateHttp(Uri uri, string method) 
    { 
     if (uri.Scheme != Uri.UriSchemeHttp && uri.Scheme != Uri.UriSchemeHttps) 
     { 
      throw new ArgumentException("The specified URI does not use HTTP or HTTPS.", "uri"); 
     } 

     var request = (HttpWebRequest)WebRequest.Create(uri); 
     request.Method = method; 

     return request; 
    } 

    public static IObservable<T> DownloadXmlFileAsync<T>(Uri uri, string elementName) where T : class 
    { 
     return (from request in Observable.Return(CreateHttp(uri)) 
       from response in Observable.FromAsyncPattern<WebResponse>(request.BeginGetResponse, request.EndGetResponse)() 
       let stream = response.GetResponseStream() 
       where stream != null 
       from item in XmlReader.Create(stream).GetXmlItem<T>(elementName).ToObservable() 
       select item); 
    } 
} 

public static class XmlExtensions 
{ 
    public static IEnumerable<T> GetXmlItem<T>(this XmlReader reader, string elementName) where T : class 
    { 
     var serializer = new XmlSerializer(typeof (T)); 
     while (reader.GoToElement(elementName)) 
     { 
      yield return serializer.Deserialize(reader) as T; 
     } 
    } 

    public static bool GoToElement(this XmlReader reader, string elementName) 
    { 
     do 
     { 
      if (reader.NodeType == XmlNodeType.Element && reader.Name == elementName) 
      { 
       return true; 
      } 
     } while (reader.Read()); 

     return false; 
    } 
} 

XmlRoot("item")] 
public class RssItem 
{ 
    [XmlElement("description")] 
    public string Description { get; set; } 

    [XmlElement("link")] 
    public string Link { get; set; } 

    [XmlElement("pubDate")] 
    public string PublishDate { get; set; } 

    [XmlElement("title")] 
    public string Title { get; set; } 

    public override string ToString() 
    { 
     return string.Format("Title: {0}", Title); 
    } 
} 

Answer 1

シーケンスのためのRx文法は次のように定義されます

OnNext *（ONERROR | OnCompleted）？パイプライン上OnError又はOnCompleted信号シーケンス及びサブスクリプションのいずれかの端部を受ける

は解体されることが期待されます。オペレータの文脈において

：

observable.Retry(n)である：OnErrorがn倍点で最大、受信されたときobservableに加入再。

observable.Finally(action)である：再試行は、サブスクリプションは、本質的にソースを開始させる冷観測（Lee Campbellがa good post on thisを有している）と共に使用することを意味するOnError|OnCompleted

を受信するactionを実行します。

同様にRepeatは、OnCompletedを受信したことを再申請する以外は、正確にRetryと同じです。

これを実際に見るには、最初のn回「失敗」し、その後成功する観測可能なものを作成することができます。 は現在、いくつかのコードのために：

private static IObservable<int> ErrorProducer(int i) 
    { 
     int count = 0; 
     return Observable.Create<int>(observer => 
     { 
      Console.WriteLine("Doing work"); 

      if (count++ < i) 
      { 
       Console.WriteLine("Failed"); 
       observer.OnError(new Exception()); 
      } 
      else 
      { 
       Console.WriteLine("Done"); 
       observer.OnNext(count); 
       observer.OnCompleted();      
      } 
      return Disposable.Empty; 
     }); 
    } 

常に失敗プロデューサーの場合：

 print(ErrorProducer(3).Retry(2)); 

与える：最終的には成功したプロデューサーのために

Doing work <-- Subscription 
Failed 
Doing work <-- Resubscription 
Failed 
OnError(System.Exception) 
Finally 

を：

print(ErrorProducer(2).Retry(3)); 

Doing work 
Failed 
Doing work 
Failed 
Doing work 
Done 
OnNext(3) <-- Succeeded 
OnCompleted() 
Finally 

あなたがy私たちのプロセスエラー関数は何度も呼び出され、再試行されるので、Retryの前に置く必要があります。

すなわち、seq.Do(value => { }, exception => { }).Retry(n)

あなたは、ホット/コールド観測を使用し、ご理解を明確にするのRxとの非同期パターンを使用して上に読むことができます。

Answer 2

アスティの答えはスポット上にあります。単一の論理シーケンスに対して複数のエラーを公開する方法を知りたい場合に備えて、追加情報を追加したかっただけです。

Astiが指摘するように、シーケンスは1回だけ終了できます。この終了は、エラーまたは完了（OnError | OnCompleted）のいずれかです。

しかし、ネストされた観測可能なシーケンスを停止することはありません！複数のエラーメッセージを表示したい場合は、IObservable<IObservable<T>>を返すシナリオを検討してください。内側のシーケンスはデータシーケンス（現在持っているシーケンス）です。このシーケンスがエラーになると、それはもはや使用することができないので、外部シーケンスは新しい内部データシーケンスを生成することができる。

これはちょっと奇妙に思えるかもしれませんが、MergeとSwitchのような演算子が既にこれらのネストしたシーケンスに対応しているので、Rxではサポートされている概念です。 RxのこのスタイルはSequences of Coincidence章

で詳しくNested Sequences段落で、その後、再び私の本、IntroToRxにで触れている私は、これはあなたが将来的にはRXする方法の他の可能性を確認するのに役立ちます願っています。