Можно ли вернуть результат метода POST HttpWebRequest через интерфейс IObservable

Question

Я пытаюсь одним методом вернуть результат метода POST HttpWebRequest через Reactive интерфейс IObservable. Мне удалось сделать это для метода 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 просто подписывается на результат получения ответа.

Я должен признать, что нахожу этот пример, расширяющий концепцию Observable. Я думаю, что вы получите лучшие результаты и более читаемый код, используя Task s, и, возможно, даже более хорошие результаты, используя новую асинхронную поддержку в C # 5 (но она недоступна, за исключением CPT).

Answer 3

obs1.Concat(obs2) возможно?