diff --git a/quinn-h3/examples/h3.rs b/quinn-h3/examples/h3.rs index 5e59ba865..e91702b36 100644 --- a/quinn-h3/examples/h3.rs +++ b/quinn-h3/examples/h3.rs @@ -240,7 +240,7 @@ fn client( conn.send_request(request) .map_err(|e| format_err!("send request: {}", e)) .and_then(|response| { - println!("recieved response: {:?}", response.headers); + println!("recieved response: {:?}", response.response()); futures::future::ok(()) }) }); diff --git a/quinn-h3/src/client.rs b/quinn-h3/src/client.rs index ad0e1b4ad..b654736cd 100644 --- a/quinn-h3/src/client.rs +++ b/quinn-h3/src/client.rs @@ -3,10 +3,11 @@ use std::net::SocketAddr; use std::net::ToSocketAddrs; use futures::{try_ready, Async, Future, Poll, Stream}; -use http::{request::Parts, Request}; +use http::{request::Parts, Request, Response}; use quinn::{ Endpoint, EndpointBuilder, EndpointDriver, EndpointError, OpenBi, RecvStream, SendStream, }; +use quinn_proto::StreamId; use slog::{self, o, Logger}; use tokio::io::{self, Shutdown, WriteAll}; @@ -115,7 +116,8 @@ enum SendRequestState { Sending(WriteAll>), Sent(Shutdown), Receiving(FrameStream), - Ready(HeadersFrame), + Decoding(HeadersFrame), + Ready(Header), Finished, } @@ -124,6 +126,7 @@ pub struct SendRequest { body: T, state: SendRequestState, conn: ConnectionRef, + stream_id: Option, recv: Option>, } @@ -144,6 +147,7 @@ impl SendRequest { conn, header: Some(Header::request(method, uri, headers)), state: SendRequestState::Opening(open_bi), + stream_id: None, recv: None, } } @@ -159,10 +163,11 @@ impl Future for SendRequest { SendRequestState::Opening(ref mut o) => { let (send, recv) = try_ready!(o.poll()); self.recv = Some(FrameStream::new(recv)); + self.stream_id = Some(send.id()); let header_block = { let conn = &mut self.conn.h3.lock().unwrap().inner; - let header = self.header.take().ok_or(Error::Internal("header none"))?; + let header = try_take(&mut self.header, "header none")?; conn.encode_header(&send.id(), header)? }; let mut encoded_header = vec![]; @@ -178,16 +183,15 @@ impl Future for SendRequest { } SendRequestState::Sent(ref mut shut) => { try_ready!(shut.poll()); - self.state = match mem::replace(&mut self.recv, None) { - Some(r) => SendRequestState::Receiving(r), - None => return Err(Error::Internal("Invalid receive state")), - } + let recv = try_take(&mut self.recv, "Invalid receive state")?; + self.state = SendRequestState::Receiving(recv); } SendRequestState::Receiving(ref mut frames) => match try_ready!(frames.poll()) { None => return Err(Error::peer("recieved an empty response")), Some(f) => match f { HttpFrame::Headers(headers) => { - match mem::replace(&mut self.state, SendRequestState::Ready(headers)) { + match mem::replace(&mut self.state, SendRequestState::Decoding(headers)) + { SendRequestState::Receiving(frames) => self.recv = Some(frames), _ => unreachable!(), }; @@ -195,13 +199,30 @@ impl Future for SendRequest { _ => return Err(Error::peer("first frame is not headers")), }, }, + SendRequestState::Decoding(ref mut frame) => { + let stream_id = try_take(&mut self.stream_id, "Stream id is none")?; + let result = { + let conn = &mut self.conn.h3.lock().unwrap().inner; + conn.decode_header(&stream_id, frame) + }; + + match result { + Ok(None) => return Ok(Async::NotReady), + Ok(Some(decoded)) => { + self.state = SendRequestState::Ready(decoded); + } + Err(e) => { + return Err(Error::peer(format!("decoding header failed: {:?}", e))) + } + } + } SendRequestState::Ready(_) => { match mem::replace(&mut self.state, SendRequestState::Finished) { SendRequestState::Ready(h) => { - return Ok(Async::Ready(RecvResponse { - headers: h, - frames: mem::replace(&mut self.recv, None).unwrap(), - })) + return Ok(Async::Ready(RecvResponse::build( + h, + try_take(&mut self.recv, "Recv is none")?, + )?)); } _ => unreachable!(), } @@ -212,7 +233,34 @@ impl Future for SendRequest { } } -pub struct RecvResponse { - pub headers: HeadersFrame, - frames: FrameStream, +fn try_take(item: &mut Option, msg: &'static str) -> Result { + mem::replace(item, None).ok_or(Error::Internal(msg)) +} + +pub struct RecvResponse { + response: Response<()>, + recv: FrameStream, +} + +impl RecvResponse { + fn build(header: Header, recv: FrameStream) -> Result { + let (status, headers) = header.into_response_parts()?; + let mut response = Response::builder(); + response.status(status); + response.version(http::version::Version::HTTP_2); // TODO change once available + *response + .headers_mut() + .ok_or(Error::peer("invalid response"))? = headers; + + Ok(Self { + recv, + response: response + .body(()) + .or(Err(Error::Internal("failed to build response")))?, + }) + } + + pub fn response<'a>(&'a self) -> &'a Response<()> { + &self.response + } } diff --git a/quinn-h3/src/proto/connection.rs b/quinn-h3/src/proto/connection.rs index e9d7a1a6b..eb943e26f 100644 --- a/quinn-h3/src/proto/connection.rs +++ b/quinn-h3/src/proto/connection.rs @@ -97,6 +97,7 @@ pub enum Error { InvalidHeaderName(String), InvalidHeaderValue(String), InvalidRequest(String), + InvalidResponse(String), Settings { reason: String }, EncodeError { reason: EncoderError }, DecodeError { reason: DecoderError }, @@ -123,6 +124,7 @@ impl From for Error { headers::Error::InvalidHeaderValue(s) => Error::InvalidHeaderValue(s), headers::Error::InvalidRequest(e) => Error::InvalidRequest(format!("{:?}", e)), headers::Error::MissingMethod => Error::InvalidRequest("missing method".into()), + headers::Error::MissingStatus => Error::InvalidResponse("missing status".into()), } } } diff --git a/quinn-h3/src/proto/headers.rs b/quinn-h3/src/proto/headers.rs index 63cdfcc6a..71896cba1 100644 --- a/quinn-h3/src/proto/headers.rs +++ b/quinn-h3/src/proto/headers.rs @@ -60,6 +60,10 @@ impl Header { )) } + pub fn into_response_parts(self) -> Result<(StatusCode, HeaderMap), Error> { + Ok((self.pseudo.status.ok_or(Error::MissingStatus)?, self.fields)) + } + pub fn len(&self) -> usize { self.pseudo.len() + self.fields.len() } @@ -308,6 +312,7 @@ pub enum Error { InvalidHeaderValue(std::string::String), InvalidRequest(http::Error), MissingMethod, + MissingStatus, } impl Error {