是否可以通过 IObservable 接口返回 HttpWebRequest POST 方法的结果

Question

我正在尝试使用单个方法通过 Reactive 接口 IObservable 返回 HttpWebRequest POST 方法的结果。我已经设法使用下面的代码对 GET 方法执行此操作：

var request = (HttpWebRequest)WebRequestCreator.ClientHttp.Create(url);
  request.Method = method;
  request.Accept = GetHttpType();
  request.CookieContainer = new CookieContainer();

   return Observable.FromAsyncPattern(request.BeginGetResponse, ar => ProcessResponse(method, ar, request))()
                    .Select(r => r);

但我不确定如何将写入请求流的异步观察器与读取 HTTP POST 操作所需的响应流链接在一起。如何连接以下变量obs1 & obs2 在一起以便我可以返回 obs2？

var request = (HttpWebRequest)WebRequestCreator.ClientHttp.Create(url);
var type = GetHttpType();
request.Method = method;
request.Accept = type;
request.ContentType = type;
request.CookieContainer = new CookieContainer();

var data = Serialize(requestResource);

var obs1 = Observable.FromAsyncPattern(request.BeginGetRequestStream, ar1 => ProcessRequest(method, data, ar1, request))

var obs2 = Observable.FromAsyncPattern(request.BeginGetResponse, ar2 => ProcessResponse(method, ar2, request))();

// How do I connect obs1 And obs2 together...

return obs2;

我本以为以下应该可行，但“ProcessResponse”方法从未被调用，有人知道为什么吗？

 var obs1 = Observable.FromAsyncPattern(request.BeginGetRequestStream, ar1 => ProcessRequest(method, data, ar1, request))();
 var obs2 = Observable.FromAsyncPattern(request.BeginGetResponse, ar2 => ProcessResponse(method, ar2, request));

 return obs1.SelectMany(a => obs2(), (a, b) => b);

Answer 1

FromAsyncPattern 期望 BeginAction 和 EndAction 方法作为参数。我认为您正在寻找的是这样的：

HttpWebRequest httpRequest = 
    (HttpWebRequest)HttpWebRequest.Create("http://www.stackoverflow.com");

httpRequest.Method = "POST";

var requestAsync = Observable.FromAsyncPattern<Stream>(
    httpRequest.BeginGetRequestStream, httpRequest.EndGetRequestStream);

var responseAsync = Observable.FromAsyncPattern<WebResponse>(
    httpRequest.BeginGetResponse, httpRequest.EndGetResponse);

requestAsync()
    .SelectMany(requestStream =>
    {
        // requestStream.Write

        return responseAsync();
    })
    .Subscribe(webResponse =>
    {
        // complete
    });

让我知道您对请求流的写入是否是异步的，我可以相应地更新示例。

Answer 2

好的，我现在明白了。首先是代码，然后是解释：

byte[] buffer1 = Encoding.UTF8.GetBytes(data); // assume post data is utf-8 encoded

var obs1 = Observable.FromAsyncPattern(request.BeginGetRequestStream, ar1 => request.EndGetRequestStream(ar1));

return from stream in obs1()
       let obs2 = Observable.FromAsyncPattern((ac, state) => stream.BeginWrite(buffer1, 0, buffer1.Length, ac, state), ar2 => stream.EndWrite(ar2))
       from _ in obs2()
       let obs3 = Observable.FromAsyncPattern(request.BeginGetResponse, ar2 => request.EndGetResponse(ar2))
       from response in obs3()
       select response;

您使用 SelectMany 的方法是正确的，但我发现常规 LINQ 语法更具可读性。我还通过异步发布改进了您的示例，即以异步方式写入上传流。基本上，第一个 from 为我们提供了一个流，用于第一个 let 中的异步发布，第三个 from 将忽略上传的结果（因此奇怪地命名为变量“_”），最后一个from 只是订阅 get 响应的结果。

我必须承认，我发现这个示例扩展了 Observable 的概念。我认为通过使用 Task 可以得到更好的结果和更易读的代码，并且使用 C# 5 中新的异步支持可能会得到更好的结果（但除了 CPT 之外，这是不可用的）。

Answer 3

也许是 obs1.Concat(obs2) ？