在 Windows Phone 7 上创建 WebRequestFactory

Question

我正在尝试创建一个基于Rx的WebRequestFactory来使用任何通用API。

我有一个基本的 GET 请求半排序

public static IObservable<T> GetData<T>(Uri uri, Func<string, T> generator)
{
    System.Diagnostics.Debug.WriteLine(uri);
    return (from request in Observable.Return(CreateWebRequest(uri))
            from response in Observable.FromAsyncPattern<WebResponse>(request.BeginGetResponse, request.EndGetResponse)()
            select generator(new StreamReader(response.GetResponseStream()).ReadToEnd())).ObserveOnDispatcher();
}
private static WebRequest CreateWebRequest(Uri uri)
{
    var ret = (HttpWebRequest)WebRequest.Create(uri);
    ret.AllowReadStreamBuffering = false;
    return ret;
}

哪个效果很好，但我不确定异常处理是如何在 Rx 中完成的，它只在 Subscribe（） 方法中完成吗？ 即订阅（onNext： response => handleResponse（response）， onError： error => handleError（error） 或者我可以更早地抓住它？

我也很难获得 POST 版本的 GetData，我目前有：

public static IObservable<T> PostData<T>(Uri uri, Func<string, T> generator, String postData)
{
    System.Diagnostics.Debug.WriteLine(uri);
    byte[] buffer = Encoding.UTF8.GetBytes(postData);
    var request = CreatePOSTWebRequest(uri);
    var obs1 = Observable.FromAsyncPattern<WebResponse>(request.BeginGetResponse, request.EndGetResponse)();
    return null;
}
private static WebRequest CreatePOSTWebRequest(Uri uri)
{
    var ret = (HttpWebRequest)WebRequest.Create(uri);
    ret.Method = "POST";
    ret.AllowReadStreamBuffering = false;
    return ret;
}

但是我不知道如何将POST数据事件的发送与服务器响应的检索联系起来。

这是我第一次使用异步HTTPRequests，所以任何帮助将不胜感激。

编辑：（我希望这是堆栈溢出推荐的方式来添加到我的问题中）

我想我已经通过使用WebClient解决了它：

public static IObservable<T> PostData<T>(Uri uri, Func<string, T> generator, String postData)
{
    System.Diagnostics.Debug.WriteLine(uri);
    WebClient wc = CreatePostWebClient(uri, postData);
    return (from e in Observable.FromEvent<UploadStringCompletedEventArgs>(wc, "UploadStringCompleted")
            select generator(e.EventArgs.Result)).ObserveOnDispatcher();
}
public static WebClient CreatePostWebClient(Uri uri, string postData)
{
    var wc = new WebClient();
    wc.AllowReadStreamBuffering = false;
    wc.UploadStringAsync(uri, postData);
    return wc;
}

但我有点担心它给我的竞争条件氛围，就像我在观察响应之前开始上传一样。

Answer 1

您是对的，错误处理是通过向管道发送OnError消息并从源中取消订阅来完成的。有些运算符可以更改此行为，例如Catch可以提供要使用的新序列（例如，使用 Observable.Return 返回默认值）。

就发布数据而言，除非您计划实际异步缓冲请求数据以免阻塞客户端（在这种情况下，它变得更加复杂），否则应该不会太困难。只要它只是发布数据，它应该只是这样：

return (from request in Observable.Return(CreateWebRequest(uri))
        from requestStream in Observable.FromAsyncPattern<WebResponse>(
                                  request.BeginGetRequest, request.EndGetRequest)()
        from response in PostAndGetResponse(request, requestStream)
        select generator(new StreamReader(response.GetResponseStream()).ReadToEnd())
       ).ObserveOnDispatcher();
private IObservable<WebResponse> PostAndGetResponse(
    WebRequest request, Stream requestStream)
{
    // Write data to requestStream
    return Observable.FromAsyncPattern<WebResponse>(
        request.BeginGetResponse, request.EndGetResponse)()
}