add TakeRequest service; update ws test case

src/h1/mod.rs

@@ -7,4 +7,4 @@ mod service;
 
 pub use self::codec::{Codec, InMessage, OutMessage};
 pub use self::dispatcher::Dispatcher;
-pub use self::service::{H1Service, H1ServiceHandler};
+pub use self::service::{H1Service, H1ServiceHandler, TakeRequest};

src/h1/service.rs

@@ -1,15 +1,17 @@
 use std::fmt::{Debug, Display};
 use std::marker::PhantomData;
 
+use actix_net::codec::Framed;
 use actix_net::service::{IntoNewService, NewService, Service};
-use futures::{Async, Future, Poll};
+use futures::{future, Async, Future, Poll, Stream};
 use tokio_io::{AsyncRead, AsyncWrite};
 
 use config::ServiceConfig;
-use error::DispatchError;
+use error::{DispatchError, ParseError};
 use request::Request;
 use response::Response;
 
+use super::codec::{Codec, InMessage};
 use super::dispatcher::Dispatcher;
 
 /// `NewService` implementation for HTTP1 transport
@@ -121,3 +123,78 @@ where
         Dispatcher::new(req, self.cfg.clone(), self.srv.clone())
     }
 }
+
+/// `NewService` that implements, read one request from framed object feature.
+pub struct TakeRequest<T> {
+    _t: PhantomData<T>,
+}
+
+impl<T> TakeRequest<T> {
+    /// Create new `TakeRequest` instance.
+    pub fn new() -> Self {
+        TakeRequest { _t: PhantomData }
+    }
+}
+
+impl<T> NewService for TakeRequest<T>
+where
+    T: AsyncRead + AsyncWrite,
+{
+    type Request = Framed<T, Codec>;
+    type Response = (Option<InMessage>, Framed<T, Codec>);
+    type Error = ParseError;
+    type InitError = ();
+    type Service = TakeRequestService<T>;
+    type Future = future::FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self) -> Self::Future {
+        future::ok(TakeRequestService { _t: PhantomData })
+    }
+}
+
+/// `NewService` that implements, read one request from framed object feature.
+pub struct TakeRequestService<T> {
+    _t: PhantomData<T>,
+}
+
+impl<T> Service for TakeRequestService<T>
+where
+    T: AsyncRead + AsyncWrite,
+{
+    type Request = Framed<T, Codec>;
+    type Response = (Option<InMessage>, Framed<T, Codec>);
+    type Error = ParseError;
+    type Future = TakeRequestServiceResponse<T>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        Ok(Async::Ready(()))
+    }
+
+    fn call(&mut self, framed: Self::Request) -> Self::Future {
+        TakeRequestServiceResponse {
+            framed: Some(framed),
+        }
+    }
+}
+
+pub struct TakeRequestServiceResponse<T>
+where
+    T: AsyncRead + AsyncWrite,
+{
+    framed: Option<Framed<T, Codec>>,
+}
+
+impl<T> Future for TakeRequestServiceResponse<T>
+where
+    T: AsyncRead + AsyncWrite,
+{
+    type Item = (Option<InMessage>, Framed<T, Codec>);
+    type Error = ParseError;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        match self.framed.as_mut().unwrap().poll()? {
+            Async::Ready(item) => Ok(Async::Ready((item, self.framed.take().unwrap()))),
+            Async::NotReady => Ok(Async::NotReady),
+        }
+    }
+}

src/ws/mod.rs

@@ -3,7 +3,6 @@
 //! To setup a `WebSocket`, first do web socket handshake then on success
 //! convert `Payload` into a `WsStream` stream and then use `WsWriter` to
 //! communicate with the peer.
-//! ```
 use std::io;
 
 use error::ResponseError;

tests/test_ws.rs

@@ -9,8 +9,9 @@ use std::{io, thread};
 
 use actix::System;
 use actix_net::codec::Framed;
+use actix_net::framed::IntoFramed;
 use actix_net::server::Server;
-use actix_net::service::IntoNewService;
+use actix_net::service::NewServiceExt;
 use actix_web::{test, ws as web_ws};
 use bytes::Bytes;
 use futures::future::{ok, Either};
@@ -18,7 +19,7 @@ use futures::{Future, Sink, Stream};
 
 use actix_http::{h1, ws, ResponseError};
 
-fn ws_handler(req: ws::Message) -> impl Future<Item = ws::Message, Error = io::Error> {
+fn ws_service(req: ws::Message) -> impl Future<Item = ws::Message, Error = io::Error> {
     match req {
         ws::Message::Ping(msg) => ok(ws::Message::Pong(msg)),
         ws::Message::Text(text) => ok(ws::Message::Text(text)),
@@ -34,13 +35,9 @@ fn test_simple() {
     thread::spawn(move || {
         Server::new()
             .bind("test", addr, move || {
-                (|io| {
+                IntoFramed::new(|| h1::Codec::new(false))
-                    // read http request
+                    .and_then(h1::TakeRequest::new().map_err(|_| ()))
-                    let framed = Framed::new(io, h1::Codec::new(false));
+                    .and_then(|(req, framed): (_, Framed<_, _>)| {
-                    framed
-                        .into_future()
-                        .map_err(|_| ())
-                        .and_then(|(req, framed)| {
                         // validate request
                         if let Some(h1::InMessage::MessageWithPayload(req)) = req {
                             match ws::handshake(&req) {
@@ -57,14 +54,13 @@ fn test_simple() {
                                 Ok(mut resp) => Either::B(
                                     // send response
                                     framed
-                                            .send(h1::OutMessage::Response(
+                                        .send(h1::OutMessage::Response(resp.finish()))
-                                                resp.finish(),
+                                        .map_err(|_| ())
-                                            )).map_err(|_| ())
                                         .and_then(|framed| {
                                             // start websocket service
                                             let framed =
                                                 framed.into_framed(ws::Codec::new());
-                                                ws::Transport::with(framed, ws_handler)
+                                            ws::Transport::with(framed, ws_service)
                                                 .map_err(|_| ())
                                         }),
                                 ),
@@ -73,7 +69,6 @@ fn test_simple() {
                             panic!()
                         }
                     })
-                }).into_new_service()
             }).unwrap()
             .run();
     });