Rx重試（）未按預期工作

我正在嘗試使用Reactive Extensions來獲取一堆RSS項目。我基於Tim Greenfield的博客文章：Silverlight Rx DataClient within MVVM。Rx重試（）未按預期工作

我在桌面應用程序中使用它，但代碼類似。

我遇到的問題是瞭解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錯誤，正如人們所期望的那樣，但它只報告一次。我預料它會兩次顯示這個錯誤。

因此，看起來Retry()函數沒有在我的web請求中運行，但看起來它實際上應用於傳遞給Subscribe()的函數。雖然我可能在這裏錯了。

如何確保Retry()調用適用於Web請求？

額外的代碼：

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

來源

2012-10-18 Jensen

爲序列對RX語法定義爲：

OnNext *（的OnError | OnCompleted）？

在接受一個OnError或OnCompleted信號被預期管道上的序列和訂閱的端部將被拆除。

在運營商的情況下：

observable.Retry(n)是：當接收到OnError重新訂閱observable，高達n次。

observable.Finally(action)是：在接收OnError|OnCompleted

重試是指用冷的觀測使用執行action（Lee Campbell具有a good post on this）其中訂閱本質上使源啓動。

同理Repeat與Retry完全相同，只是它在收到OnCompleted後重新訂閱。

爲了看到這一點，我們可以創建一個observable，它將在前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異步模式，以澄清你的理解閱讀起來。

來源

2012-10-18 16:35:46 Asti

你的答案提供了一些很好的見解，也讓我有導致http://social.msdn.microsoft.com/Forums/da-DK/rx/thread/96a06e27-9c02-4177一些特定的關鍵字搜索互聯網-ae6a-04b8a7f966e5，這讓我更加了解可觀察工作的方式。 – Jensen

@JensenSomers很高興我能幫上忙。抱歉，關於你的問題的確切原因不是非常具體。有關應用Rx的文檔很少，我希望這些常規答案可能對未來嘗試學習Rx的人有用。 – Asti

如果每個人都會像我一樣對Rx退出，我相信在不久的將來會出現適當的文檔和更大的用例示例。 :-) – Jensen

阿斯蒂的答案是當場上。我只是想添加一些額外的信息，以防您想知道如何爲單個邏輯序列公開多個錯誤。

由於阿斯蒂所指出的，你只能終止一次的序列。該終止可以是錯誤或完成（OnError | OnCompleted）。

但是沒有什麼能夠阻止你已經嵌套觀察序列！如果您確實想看到多個錯誤消息，請考慮您返回IObservable<IObservable<T>>的情況。內部序列是數據序列（您當前擁有的序列）。當這個序列錯誤時，它不能再使用，所以外部序列可以產生一個新的內部數據序列。

這似乎有點奇怪，但它是在一個的Rx支持概念作爲像合併運算符和交換機已經滿足這些嵌套序列。這種Rx的風格在我的書Nested Sequences段的IntroToRx中有所涉及，然後在Sequences of Coincidence章

中再次詳細介紹，我希望這可以幫助您瞭解將來如何使用Rx的其他可能性。

來源

2012-10-22 09:00:06

Rx重試（）未按預期工作

回答

相關問題