如何不断重试反应式方法直到成功？

Question

这是我的 WebClient 异步下载反应式扩展。 一次又一次调用“DownloadStringAsync”直到操作成功的最佳方法是什么？

像这样但以反应方式：

while (true)
{
  var result = DownloadStringAsync();
  if (result)
  {
    return;
  }
}

---

我的代码：

[Serializable]
public class WebClientException : Exception
{
    public WebClientResponse Response { get; set; }

    public WebClientException()
    {
    }

    public WebClientException(string message)
        : base(message)
    {
    }

    public WebClientException(string message, Exception innerException)
        : base(message, innerException)
    {
    }

    protected WebClientException(SerializationInfo info, StreamingContext context)
        : base(info, context)
    {
    }
}

public class WebClientResponse
{
    public WebHeaderCollection Headers { get; set; }
    public HttpStatusCode StatusCode { get; set; }
    public string Result { get; set; }
    public WebException Exception { get; set; }
}

public static IObservable<WebClientResponse> DownloadStringAsync(this WebClient webClient, Uri address, WebHeaderCollection requestHeaders)
{
    var asyncResult =
        Observable.FromEventPattern<DownloadStringCompletedEventHandler, DownloadStringCompletedEventArgs>
            (ev => webClient.DownloadStringCompleted += ev, ev => webClient.DownloadStringCompleted -= ev)
            .ObserveOn(Scheduler.TaskPool)
            .Select(o =>
                        {
                            var ex = o.EventArgs.Error as WebException;

                            if (ex == null)
                            {
                                var wc = (WebClient) o.Sender;

                                return new WebClientResponse {Headers = wc.ResponseHeaders, Result = o.EventArgs.Result};
                            }

                            var wcr = new WebClientResponse {Exception = ex};

                            var r = ex.Response as HttpWebResponse;
                            if (r != null)
                            {
                                wcr.Headers = r.Headers;
                                wcr.StatusCode = r.StatusCode;

                                var s = r.GetResponseStream();
                                if (s != null)
                                {
                                    using (TextReader tr = new StreamReader(s))
                                    {
                                        wcr.Result = tr.ReadToEnd();
                                    }
                                }
                            }

                            throw new WebClientException {Response = wcr};
                        })
            .Take(1);

    if (requestHeaders != null)
    {
        foreach (var key in requestHeaders.AllKeys)
        {
            webClient.Headers.Add(key, requestHeaders[key]);
        }
    }

    webClient.DownloadStringAsync(address);

    return asyncResult;
}

Answer 1

您的方法会生成一个热可观察对象，这意味着它在返回时已经开始加载，并且每个新订阅都不会向 Web 服务器创建新请求。您需要将您的方法包装在另一个方法中并使用 Observable.Create （为了创建一个冷 observable，它会在每个订阅上创建一个新请求）：

public static IObservable<WebClientResponse> DownloadStringAsync(this WebClient webClient, Uri address, WebHeaderCollection requestHeaders)
{
    return Observable
        .Create(observer => 
        {
            DownloadStringAsyncImpl(webClient, address, requestHeaders)
                .Subscribe(observer);
            return () => { webClient.CancelAsync(); };
        });
}

这里，DownloadStringAsyncImpl 是您以前的 DownloadStringAsync 实现，而公共方法已被替换。

现在您可以重试异步方法，直到成功，如下所示：

myWebClient
    .DownloadStringAsync( /* args... */)
    .Retry()
    .Subscribe(result => { 
         /* now I've got a result! */
    });

Answer 2

我认为你至少有一个像样的“这里是一些代码”答案，所以我将重点关注更一般的手持方式。

我首先要看的是Rx 设计指南。这是一个简短的（34 页）PDF 文档，有助于将范式从拉动“订阅”转变为推送，或从 IEnumerable 转变为 IObservable。

如果您想更进一步，可以使用 .NET 的 PDF HOL（实践实验室） 和 JavaScript。您可以在 Rx 页面上找到其他资源（从这里开始）。

Answer 3

如果是异步函数。进行重复检查意味着您将其变成了同步函数调用。这是你真正想做的事情吗？

您可以让一个专用线程调用此异步函数，并在调用此函数后阻止自身。创建此线程时，向其传递一个委托，该委托应在异步函数返回后调用。完成后，使用错误代码调用委托。

希望这能回答您的问题。