Created client/server http codecs, added async intows trait.

Author: illegalprime

examples/client.rs

@@ -74,7 +74,7 @@ fn main() {
 					let _ = tx_1.send(OwnedMessage::Close(None));
 					return;
 				}
-				  OwnedMessage::Ping(data) => {
+				OwnedMessage::Ping(data) => {
 					match tx_1.send(OwnedMessage::Pong(data)) {
 						// Send a pong in response
 						Ok(()) => (),

src/client/builder.rs

@@ -551,7 +551,7 @@ impl<'u> ClientBuilder<'u> {
 			key_set: self.key_set,
 		};
 		let resource = builder.build_request();
-		let framed = stream.framed(::codec::http::HttpCodec);
+		let framed = stream.framed(::codec::http::HttpClientCodec);
 		let request = Incoming {
 			version: builder.version,
 			headers: builder.headers.clone(),
@@ -563,7 +563,7 @@ impl<'u> ClientBuilder<'u> {
           .send(request).map_err(::std::convert::Into::into)
 
           // wait for a response
-		      .and_then(|stream| stream.into_future().map_err(|e| e.0))
+		      .and_then(|stream| stream.into_future().map_err(|e| e.0.into()))
 
           // validate
 		      .and_then(move |(message, stream)| {

src/codec/http.rs

@@ -1,20 +1,24 @@
+use std::io::{self, Write};
+use std::error::Error;
+use std::fmt::{self, Formatter, Display};
 use hyper;
 use hyper::http::h1::Incoming;
 use hyper::http::h1::parse_response;
+use hyper::http::h1::parse_request;
 use hyper::http::RawStatus;
+use hyper::status::StatusCode;
 use hyper::method::Method;
 use hyper::uri::RequestUri;
 use hyper::buffer::BufReader;
-use std::io::{self, Write};
 use tokio_io::codec::{Decoder, Encoder};
 use bytes::BytesMut;
 use bytes::BufMut;
 use result::WebSocketError;
 
 #[derive(Copy, Clone, Debug)]
-pub struct HttpCodec;
+pub struct HttpClientCodec;
 
-impl Encoder for HttpCodec {
+impl Encoder for HttpClientCodec {
 	type Item = Incoming<(Method, RequestUri)>;
 	type Error = io::Error;
 
@@ -33,9 +37,9 @@ impl Encoder for HttpCodec {
 	}
 }
 
-impl Decoder for HttpCodec {
+impl Decoder for HttpClientCodec {
 	type Item = Incoming<RawStatus>;
-	type Error = WebSocketError;
+	type Error = HttpCodecError;
 
 	fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
 		// check if we get a request from hyper
@@ -58,3 +62,88 @@ impl Decoder for HttpCodec {
 		Ok(Some(response))
 	}
 }
+
+#[derive(Copy, Clone, Debug)]
+pub struct HttpServerCodec;
+
+impl Encoder for HttpServerCodec {
+	type Item = Incoming<RawStatus>;
+	type Error = io::Error;
+
+	fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> {
+		// TODO: optomize this!
+		let status = StatusCode::from_u16(item.subject.0);
+		let response = format!("{} {}\r\n{}\r\n", item.version, status, item.headers);
+		let byte_len = response.as_bytes().len();
+		if byte_len > dst.remaining_mut() {
+			dst.reserve(byte_len);
+		}
+		dst.writer().write(response.as_bytes()).map(|_| ())
+	}
+}
+
+impl Decoder for HttpServerCodec {
+	type Item = Incoming<(Method, RequestUri)>;
+	type Error = HttpCodecError;
+
+	fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+		// check if we get a request from hyper
+		// TODO: this is ineffecient, but hyper does not give us a better way to parse
+		let (response, bytes_read) = {
+			let mut reader = BufReader::new(&*src as &[u8]);
+			let res = match parse_request(&mut reader) {
+				Err(hyper::Error::Io(ref e)) if e.kind() == io::ErrorKind::UnexpectedEof => {
+					return Ok(None)
+				}
+				Err(hyper::Error::TooLarge) => return Ok(None),
+				Err(e) => return Err(e.into()),
+				Ok(r) => r,
+			};
+			let (_, _, pos, _) = reader.into_parts();
+			(res, pos)
+		};
+
+		src.split_to(bytes_read);
+		Ok(Some(response))
+	}
+}
+
+#[derive(Debug)]
+pub enum HttpCodecError {
+	Io(io::Error),
+	Http(hyper::Error),
+}
+
+impl Display for HttpCodecError {
+	fn fmt(&self, fmt: &mut Formatter) -> Result<(), fmt::Error> {
+		fmt.write_str(self.description())
+	}
+}
+
+impl Error for HttpCodecError {
+	fn description(&self) -> &str {
+		match *self {
+			HttpCodecError::Io(ref e) => e.description(),
+			HttpCodecError::Http(ref e) => e.description(),
+		}
+	}
+
+	fn cause(&self) -> Option<&Error> {
+		match *self {
+			HttpCodecError::Io(ref error) => Some(error),
+			HttpCodecError::Http(ref error) => Some(error),
+		}
+	}
+}
+
+impl From<io::Error> for HttpCodecError {
+	fn from(err: io::Error) -> HttpCodecError {
+		HttpCodecError::Io(err)
+	}
+}
+
+impl From<hyper::Error> for HttpCodecError {
+	fn from(err: hyper::Error) -> HttpCodecError {
+		HttpCodecError::Http(err)
+	}
+}

src/result.rs

@@ -138,6 +138,16 @@ impl From<Utf8Error> for WebSocketError {
 	}
 }
 
+#[cfg(feature="async")]
+impl From<::codec::http::HttpCodecError> for WebSocketError {
+	fn from(src: ::codec::http::HttpCodecError) -> Self {
+		match src {
+			::codec::http::HttpCodecError::Io(e) => WebSocketError::IoError(e),
+			::codec::http::HttpCodecError::Http(e) => WebSocketError::HttpError(e),
+		}
+	}
+}
+
 impl From<WSUrlErrorKind> for WebSocketError {
 	fn from(err: WSUrlErrorKind) -> WebSocketError {
 		WebSocketError::WebSocketUrlError(err)

src/server/upgrade/async.rs

@@ -0,0 +1,65 @@
+use super::{Buffer, HyperIntoWsError, WsUpgrade, Request, validate};
+use std::io::{self, ErrorKind};
+use hyper::header::Headers;
+use stream::AsyncStream;
+use futures::{Stream, Future};
+use codec::http::HttpServerCodec;
+use bytes::BytesMut;
+
+pub trait AsyncIntoWs {
+	/// The type of stream this upgrade process is working with (TcpStream, etc.)
+	type Stream: AsyncStream;
+	/// An error value in case the stream is not asking for a websocket connection
+	/// or something went wrong. It is common to also include the stream here.
+	type Error;
+	/// Attempt to read and parse the start of a Websocket handshake, later
+	/// with the  returned `WsUpgrade` struct, call `accept to start a
+	/// websocket client, and `reject` to send a handshake rejection response.
+	///
+	/// Note: this is the asynchronous version, meaning it will not block when
+	/// trying to read a request.
+	fn into_ws(self) -> Box<Future<Item = WsUpgrade<Self::Stream>, Error = Self::Error>>;
+}
+
+impl<S> AsyncIntoWs for S
+    where S: AsyncStream + 'static
+{
+	type Stream = S;
+	type Error = (S, Option<Request>, Option<BytesMut>, HyperIntoWsError);
+
+	fn into_ws(self) -> Box<Future<Item = WsUpgrade<Self::Stream>, Error = Self::Error>> {
+		let future = self.framed(HttpServerCodec)
+          .into_future()
+          .map_err(|(e, s)| {
+              let (stream, buffer) = s.into_parts();
+              (stream, None, Some(buffer), e.into())
+          })
+          .and_then(|(m, s)| {
+              let (stream, buffer) = s.into_parts();
+              if let Some(msg) = m {
+                  match validate(&msg.subject.0, &msg.version, &msg.headers) {
+                      Ok(()) => Ok((msg, stream, buffer)),
+                      Err(e) => Err((stream, None, Some(buffer), e)),
+                  }
+              } else {
+                  let err = HyperIntoWsError::Io(io::Error::new(
+                      ErrorKind::ConnectionReset,
+                  "Connection dropped before handshake could be read"));
+                  Err((stream, None, Some(buffer), err))
+              }
+          })
+          .map(|(m, stream, buffer)| {
+              WsUpgrade {
+                  headers: Headers::new(),
+                  stream: stream,
+                  request: m,
+                  buffer: Some(Buffer {
+                      buf: unimplemented!(),
+                      pos: 0,
+                      cap: buffer.capacity(),
+                  }),
+              }
+          });
+		Box::new(future)
+	}
+}

src/server/upgrade/mod.rs

@@ -24,6 +24,9 @@ use hyper::header::{Headers, Upgrade, Protocol, ProtocolName, Connection, Connec
 
 pub mod from_hyper;
 
+#[cfg(feature="async")]
+pub mod async;
+
 /// This crate uses buffered readers to read in the handshake quickly, in order to
 /// interface with other use cases that don't use buffered readers the buffered readers
 /// is deconstructed when it is returned to the user and given as the underlying
@@ -43,6 +46,14 @@ pub struct Buffer {
 	pub cap: usize,
 }
 
+/// A typical request from hyper
+pub type Request = Incoming<(Method, RequestUri)>;
+
+/// If you have your requests separate from your stream you can use this struct
+/// to upgrade the connection based on the request given
+/// (the request should be a handshake).
+pub struct RequestStreamPair<S: Stream>(pub S, pub Request);
+
 /// Intermediate representation of a half created websocket session.
 /// Should be used to examine the client's handshake
 /// accept the protocols requested, route the path, etc.
@@ -246,14 +257,6 @@ pub trait IntoWs {
 	fn into_ws(self) -> Result<WsUpgrade<Self::Stream>, Self::Error>;
 }
 
-
-/// A typical request from hyper
-pub type Request = Incoming<(Method, RequestUri)>;
-/// If you have your requests separate from your stream you can use this struct
-/// to upgrade the connection based on the request given
-/// (the request should be a handshake).
-pub struct RequestStreamPair<S: Stream>(pub S, pub Request);
-
 impl<S> IntoWs for S
     where S: Stream
 {
@@ -381,6 +384,16 @@ impl From<::hyper::error::Error> for HyperIntoWsError {
 	}
 }
 
+#[cfg(feature="async")]
+impl From<::codec::http::HttpCodecError> for HyperIntoWsError {
+	fn from(src: ::codec::http::HttpCodecError) -> Self {
+		match src {
+			::codec::http::HttpCodecError::Io(e) => HyperIntoWsError::Io(e),
+			::codec::http::HttpCodecError::Http(e) => HyperIntoWsError::Parsing(e),
+		}
+	}
+}
+
 fn validate(
 	method: &Method,
 	version: &HttpVersion,