Implemented async IntoWs and WsUpgrade, only server is left.

Author: illegalprime

src/codec/http.rs

@@ -66,13 +66,12 @@ impl Decoder for HttpClientCodec {
 pub struct HttpServerCodec;
 
 impl Encoder for HttpServerCodec {
-	type Item = Incoming<RawStatus>;
+	type Item = Incoming<StatusCode>;
 	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 response = format!("{} {}\r\n{}\r\n", item.version, item.subject, item.headers);
 		let byte_len = response.as_bytes().len();
 		if byte_len > dst.remaining_mut() {
 			dst.reserve(byte_len);

src/codec/ws.rs

@@ -15,6 +15,9 @@ use ws::message::Message as MessageTrait;
 use ws::util::header::read_header;
 use result::WebSocketError;
 
+// TODO: IMPORTANT: check if frame_size is correct,
+// do not call .reserve with the entire size
+
 /**************
  * Dataframes *
  **************/

src/server/mod.rs

@@ -5,7 +5,7 @@ use std::convert::Into;
 #[cfg(feature="ssl")]
 use native_tls::{TlsStream, TlsAcceptor};
 use stream::Stream;
-use self::upgrade::{WsUpgrade, IntoWs, Buffer};
+use self::upgrade::{SyncWsUpgrade, IntoWs, Buffer};
 pub use self::upgrade::{Request, HyperIntoWsError};
 
 pub mod upgrade;
@@ -35,7 +35,7 @@ pub struct InvalidConnection<S>
 /// Either the stream was established and it sent a websocket handshake
 /// which represents the `Ok` variant, or there was an error (this is the
 /// `Err` variant).
-pub type AcceptResult<S> = Result<WsUpgrade<S>, InvalidConnection<S>>;
+pub type AcceptResult<S> = Result<SyncWsUpgrade<S>, InvalidConnection<S>>;
 
 /// Marker struct for a struct not being secure
 #[derive(Clone)]

src/server/upgrade/async.rs

@@ -1,33 +1,81 @@
-use super::{Buffer, HyperIntoWsError, WsUpgrade, Request, validate};
+use super::{HyperIntoWsError, WsUpgrade, Request, validate};
 use std::io::{self, ErrorKind};
-use tokio_io::codec::FramedParts;
+use tokio_io::codec::{Framed, FramedParts};
 use hyper::header::Headers;
+use hyper::http::h1::Incoming;
+use hyper::status::StatusCode;
 use stream::AsyncStream;
-use futures::{Stream, Future};
+use futures::{Stream, Sink, Future};
+use futures::sink::Send;
 use codec::http::HttpServerCodec;
+use codec::ws::{MessageCodec, Context};
 use bytes::BytesMut;
 use client::async::ClientNew;
 
-impl<S> WsUpgrade<S>
-    where S: AsyncStream
+pub type AsyncWsUpgrade<S> = WsUpgrade<S, BytesMut>;
+
+impl<S> AsyncWsUpgrade<S>
+    where S: AsyncStream + 'static
 {
-	pub fn async_accept(self) -> Result<ClientNew<S>, (S, io::Error)> {
-		unimplemented!();
+	pub fn async_accept(self) -> ClientNew<S> {
+		self.internal_async_accept(None)
+	}
+
+	pub fn async_accept_with(self, custom_headers: &Headers) -> ClientNew<S> {
+		self.internal_async_accept(Some(custom_headers))
+	}
+
+	fn internal_async_accept(mut self, custom_headers: Option<&Headers>) -> ClientNew<S> {
+		let status = self.prepare_headers(custom_headers);
+		let WsUpgrade { headers, stream, request, buffer } = self;
+
+		let duplex = Framed::from_parts(FramedParts {
+		                                    inner: stream,
+		                                    readbuf: buffer,
+		                                    writebuf: BytesMut::with_capacity(0),
+		                                },
+		                                HttpServerCodec);
+
+		let future = duplex.send(Incoming {
+		                             version: request.version,
+		                             subject: status,
+		                             headers: headers.clone(),
+		                         })
+		                   .map(move |s| {
+			                        let codec = MessageCodec::default(Context::Client);
+			                        let client = Framed::from_parts(s.into_parts(), codec);
+			                        (client, headers)
+			                       })
+		                   .map_err(|e| e.into());
+		Box::new(future)
 	}
 
-	pub fn async_accept_with(
-		self,
-		custom_headers: &Headers,
-	) -> Result<ClientNew<S>, (S, io::Error)> {
-		unimplemented!();
+	pub fn async_reject(self) -> Send<Framed<S, HttpServerCodec>> {
+		self.internal_async_reject(None)
 	}
 
-	pub fn async_reject(self) -> Result<S, (S, io::Error)> {
-		unimplemented!();
+	pub fn async_reject_with(self, headers: &Headers) -> Send<Framed<S, HttpServerCodec>> {
+		self.internal_async_reject(Some(headers))
 	}
 
-	pub fn async_reject_with(self, headers: &Headers) -> Result<S, (S, io::Error)> {
-		unimplemented!();
+	fn internal_async_reject(
+		mut self,
+		headers: Option<&Headers>,
+	) -> Send<Framed<S, HttpServerCodec>> {
+		if let Some(custom) = headers {
+			self.headers.extend(custom.iter());
+		}
+		let duplex = Framed::from_parts(FramedParts {
+		                                    inner: self.stream,
+		                                    readbuf: self.buffer,
+		                                    writebuf: BytesMut::with_capacity(0),
+		                                },
+		                                HttpServerCodec);
+		duplex.send(Incoming {
+		                version: self.request.version,
+		                subject: StatusCode::BadRequest,
+		                headers: self.headers,
+		            })
 	}
 }
 
@@ -44,46 +92,42 @@ pub trait AsyncIntoWs {
 	///
 	/// 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>>;
+	fn into_ws(self) -> Box<Future<Item = AsyncWsUpgrade<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);
+	type Error = (S, Option<Request>, BytesMut, HyperIntoWsError);
 
-	fn into_ws(self) -> Box<Future<Item = WsUpgrade<Self::Stream>, Error = Self::Error>> {
+	fn into_ws(self) -> Box<Future<Item = AsyncWsUpgrade<Self::Stream>, Error = Self::Error>> {
 		let future = self.framed(HttpServerCodec)
           .into_future()
           .map_err(|(e, s)| {
               let FramedParts { inner, readbuf, .. } = s.into_parts();
-              (inner, None, Some(readbuf), e.into())
+              (inner, None, readbuf, e.into())
           })
           .and_then(|(m, s)| {
               let FramedParts { inner, readbuf, .. } = s.into_parts();
               if let Some(msg) = m {
                   match validate(&msg.subject.0, &msg.version, &msg.headers) {
                       Ok(()) => Ok((msg, inner, readbuf)),
-                      Err(e) => Err((inner, None, Some(readbuf), e)),
+                      Err(e) => Err((inner, None, readbuf, e)),
                   }
               } else {
                   let err = HyperIntoWsError::Io(io::Error::new(
                       ErrorKind::ConnectionReset,
                   "Connection dropped before handshake could be read"));
-                  Err((inner, None, Some(readbuf), err))
+                  Err((inner, None, readbuf, err))
               }
           })
           .map(|(m, stream, buffer)| {
               WsUpgrade {
                   headers: Headers::new(),
                   stream: stream,
                   request: m,
-                  buffer: Some(Buffer {
-                      buf: unimplemented!(),
-                      pos: 0,
-                      cap: buffer.capacity(),
-                  }),
+                  buffer: buffer,
               }
           });
 		Box::new(future)

src/server/upgrade/from_hyper.rs

@@ -36,7 +36,7 @@
 //! ```
 
 use hyper::net::NetworkStream;
-use super::{IntoWs, WsUpgrade, Buffer};
+use super::{IntoWs, SyncWsUpgrade, Buffer};
 
 pub use hyper::http::h1::Incoming;
 pub use hyper::method::Method;
@@ -58,7 +58,7 @@ impl<'a, 'b> IntoWs for HyperRequest<'a, 'b> {
 	type Stream = &'a mut &'b mut NetworkStream;
 	type Error = (Request<'a, 'b>, HyperIntoWsError);
 
-	fn into_ws(self) -> Result<WsUpgrade<Self::Stream>, Self::Error> {
+	fn into_ws(self) -> Result<SyncWsUpgrade<Self::Stream>, Self::Error> {
 		if let Err(e) = validate(&self.0.method, &self.0.version, &self.0.headers) {
 			return Err((self.0, e));
 		}
@@ -70,7 +70,7 @@ impl<'a, 'b> IntoWs for HyperRequest<'a, 'b> {
 		let (buf, pos, cap) = reader.take_buf();
 		let stream = reader.get_mut();
 
-		Ok(WsUpgrade {
+		Ok(SyncWsUpgrade {
 		       headers: Headers::new(),
 		       stream: stream,
 		       buffer: Some(Buffer {