add H1 transport

src/server/channel.rs

@@ -1,4 +1,4 @@
-use std::net::{Shutdown, SocketAddr};
+use std::net::Shutdown;
 use std::{io, mem, time};
 
 use bytes::{Buf, BufMut, BytesMut};
@@ -16,7 +16,7 @@ const HTTP2_PREFACE: [u8; 14] = *b"PRI * HTTP/2.0";
 pub(crate) enum HttpProtocol<T: IoStream, H: HttpHandler + 'static> {
     H1(h1::Http1Dispatcher<T, H>),
     H2(h2::Http2<T, H>),
-    Unknown(ServiceConfig<H>, Option<SocketAddr>, T, BytesMut),
+    Unknown(ServiceConfig<H>, T, BytesMut),
     None,
 }
 
@@ -29,7 +29,7 @@ impl<T: IoStream, H: HttpHandler + 'static> HttpProtocol<T, H> {
                 let _ = IoStream::shutdown(io, Shutdown::Both);
             }
             HttpProtocol::H2(ref mut h2) => h2.shutdown(),
-            HttpProtocol::Unknown(_, _, io, _) => {
+            HttpProtocol::Unknown(_, io, _) => {
                 let _ = IoStream::set_linger(io, Some(time::Duration::new(0, 0)));
                 let _ = IoStream::shutdown(io, Shutdown::Both);
             }
@@ -59,9 +59,7 @@ where
     T: IoStream,
     H: HttpHandler + 'static,
 {
-    pub(crate) fn new(
+    pub(crate) fn new(settings: ServiceConfig<H>, io: T) -> HttpChannel<T, H> {
-        settings: ServiceConfig<H>, io: T, peer: Option<SocketAddr>,
-    ) -> HttpChannel<T, H> {
         let ka_timeout = settings.client_timer();
 
         HttpChannel {
@@ -69,7 +67,6 @@ where
             node_reg: false,
             node: Node::new(HttpProtocol::Unknown(
                 settings,
-                peer,
                 io,
                 BytesMut::with_capacity(8192),
             )),
@@ -102,7 +99,7 @@ where
                 Ok(Async::Ready(_)) => {
                     trace!("Slow request timed out, close connection");
                     let proto = mem::replace(self.node.get_mut(), HttpProtocol::None);
-                    if let HttpProtocol::Unknown(settings, _, io, buf) = proto {
+                    if let HttpProtocol::Unknown(settings, io, buf) = proto {
                         *self.node.get_mut() =
                             HttpProtocol::H1(h1::Http1Dispatcher::for_error(
                                 settings,
@@ -125,7 +122,7 @@ where
             let settings = match self.node.get_mut() {
                 HttpProtocol::H1(ref mut h1) => h1.settings().clone(),
                 HttpProtocol::H2(ref mut h2) => h2.settings().clone(),
-                HttpProtocol::Unknown(ref mut settings, _, _, _) => settings.clone(),
+                HttpProtocol::Unknown(ref mut settings, _, _) => settings.clone(),
                 HttpProtocol::None => unreachable!(),
             };
             settings.head().insert(&mut self.node);
@@ -135,7 +132,7 @@ where
         let kind = match self.node.get_mut() {
             HttpProtocol::H1(ref mut h1) => return h1.poll(),
             HttpProtocol::H2(ref mut h2) => return h2.poll(),
-            HttpProtocol::Unknown(_, _, ref mut io, ref mut buf) => {
+            HttpProtocol::Unknown(_, ref mut io, ref mut buf) => {
                 let mut err = None;
                 let mut disconnect = false;
                 match io.read_available(buf) {
@@ -173,13 +170,12 @@ where
 
         // upgrade to specific http protocol
         let proto = mem::replace(self.node.get_mut(), HttpProtocol::None);
-        if let HttpProtocol::Unknown(settings, addr, io, buf) = proto {
+        if let HttpProtocol::Unknown(settings, io, buf) = proto {
             match kind {
                 ProtocolKind::Http1 => {
                     *self.node.get_mut() = HttpProtocol::H1(h1::Http1Dispatcher::new(
                         settings,
                         io,
-                        addr,
                         buf,
                         is_eof,
                         self.ka_timeout.take(),
@@ -190,7 +186,6 @@ where
                     *self.node.get_mut() = HttpProtocol::H2(h2::Http2::new(
                         settings,
                         io,
-                        addr,
                         buf.freeze(),
                         self.ka_timeout.take(),
                     ));
@@ -202,6 +197,70 @@ where
     }
 }
 
+#[doc(hidden)]
+pub struct H1Channel<T, H>
+where
+    T: IoStream,
+    H: HttpHandler + 'static,
+{
+    node: Node<HttpProtocol<T, H>>,
+    node_reg: bool,
+}
+
+impl<T, H> H1Channel<T, H>
+where
+    T: IoStream,
+    H: HttpHandler + 'static,
+{
+    pub(crate) fn new(settings: ServiceConfig<H>, io: T) -> H1Channel<T, H> {
+        H1Channel {
+            node_reg: false,
+            node: Node::new(HttpProtocol::H1(h1::Http1Dispatcher::new(
+                settings,
+                io,
+                BytesMut::with_capacity(8192),
+                false,
+                None,
+            ))),
+        }
+    }
+}
+
+impl<T, H> Drop for H1Channel<T, H>
+where
+    T: IoStream,
+    H: HttpHandler + 'static,
+{
+    fn drop(&mut self) {
+        self.node.remove();
+    }
+}
+
+impl<T, H> Future for H1Channel<T, H>
+where
+    T: IoStream,
+    H: HttpHandler + 'static,
+{
+    type Item = ();
+    type Error = HttpDispatchError;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        if !self.node_reg {
+            self.node_reg = true;
+            let settings = match self.node.get_mut() {
+                HttpProtocol::H1(ref mut h1) => h1.settings().clone(),
+                _ => unreachable!(),
+            };
+            settings.head().insert(&mut self.node);
+        }
+
+        match self.node.get_mut() {
+            HttpProtocol::H1(ref mut h1) => h1.poll(),
+            _ => unreachable!(),
+        }
+    }
+}
+
 pub(crate) struct Node<T> {
     next: Option<*mut Node<T>>,
     prev: Option<*mut Node<T>>,

src/server/h1.rs

@@ -87,9 +87,10 @@ where
     H: HttpHandler + 'static,
 {
     pub fn new(
-        settings: ServiceConfig<H>, stream: T, addr: Option<SocketAddr>, buf: BytesMut,
+        settings: ServiceConfig<H>, stream: T, buf: BytesMut, is_eof: bool,
-        is_eof: bool, keepalive_timer: Option<Delay>,
+        keepalive_timer: Option<Delay>,
     ) -> Self {
+        let addr = stream.peer_addr();
         let (ka_expire, ka_timer) = if let Some(delay) = keepalive_timer {
             (delay.deadline(), Some(delay))
         } else if let Some(delay) = settings.keep_alive_timer() {
@@ -107,12 +108,12 @@ where
         };
 
         Http1Dispatcher {
-            flags,
             stream: H1Writer::new(stream, settings.clone()),
             decoder: H1Decoder::new(),
             payload: None,
             tasks: VecDeque::new(),
             error: None,
+            flags,
             addr,
             buf,
             settings,
@@ -337,9 +338,11 @@ where
     /// read data from the stream
     pub(self) fn poll_io(&mut self) -> Result<bool, HttpDispatchError> {
         if !self.flags.contains(Flags::POLLED) {
-            let updated = self.parse()?;
             self.flags.insert(Flags::POLLED);
-            return Ok(updated);
+            if !self.buf.is_empty() {
+                let updated = self.parse()?;
+                return Ok(updated);
+            }
         }
 
         // read io from socket

src/server/h2.rs

@@ -58,9 +58,9 @@ where
     H: HttpHandler + 'static,
 {
     pub fn new(
-        settings: ServiceConfig<H>, io: T, addr: Option<SocketAddr>, buf: Bytes,
+        settings: ServiceConfig<H>, io: T, buf: Bytes, keepalive_timer: Option<Delay>,
-        keepalive_timer: Option<Delay>,
     ) -> Self {
+        let addr = io.peer_addr();
         let extensions = io.extensions();
         Http2 {
             flags: Flags::empty(),

src/server/incoming.rs

@@ -64,8 +64,6 @@ where
     type Result = ();
 
     fn handle(&mut self, msg: WrapperStream<T>, _: &mut Context<Self>) -> Self::Result {
-        Arbiter::spawn(
+        Arbiter::spawn(HttpChannel::new(self.settings.clone(), msg).map_err(|_| ()));
-            HttpChannel::new(self.settings.clone(), msg, None).map_err(|_| ()),
-        );
     }
 }

src/server/mod.rs

@@ -106,7 +106,7 @@
 //!     let _ = sys.run();
 //!}
 //! ```
-use std::net::Shutdown;
+use std::net::{Shutdown, SocketAddr};
 use std::rc::Rc;
 use std::{io, time};
 
@@ -143,10 +143,13 @@ pub use self::message::Request;
 pub use self::ssl::*;
 
 pub use self::error::{AcceptorError, HttpDispatchError};
-pub use self::settings::{ServerSettings, ServiceConfig, ServiceConfigBuilder};
+pub use self::settings::ServerSettings;
 
 #[doc(hidden)]
-pub use self::service::{HttpService, StreamConfiguration};
+pub use self::settings::{ServiceConfig, ServiceConfigBuilder};
+
+#[doc(hidden)]
+pub use self::service::{H1Service, HttpService, StreamConfiguration};
 
 #[doc(hidden)]
 pub use self::helpers::write_content_length;
@@ -266,6 +269,12 @@ pub trait Writer {
 pub trait IoStream: AsyncRead + AsyncWrite + 'static {
     fn shutdown(&mut self, how: Shutdown) -> io::Result<()>;
 
+    /// Returns the socket address of the remote peer of this TCP connection.
+    fn peer_addr(&self) -> Option<SocketAddr> {
+        None
+    }
+
+    /// Sets the value of the TCP_NODELAY option on this socket.
     fn set_nodelay(&mut self, nodelay: bool) -> io::Result<()>;
 
     fn set_linger(&mut self, dur: Option<time::Duration>) -> io::Result<()>;
@@ -341,6 +350,11 @@ impl IoStream for TcpStream {
         TcpStream::shutdown(self, how)
     }
 
+    #[inline]
+    fn peer_addr(&self) -> Option<SocketAddr> {
+        TcpStream::peer_addr(self).ok()
+    }
+
     #[inline]
     fn set_nodelay(&mut self, nodelay: bool) -> io::Result<()> {
         TcpStream::set_nodelay(self, nodelay)

src/server/service.rs

@@ -5,7 +5,7 @@ use actix_net::service::{NewService, Service};
 use futures::future::{ok, FutureResult};
 use futures::{Async, Poll};
 
-use super::channel::HttpChannel;
+use super::channel::{H1Channel, HttpChannel};
 use super::error::HttpDispatchError;
 use super::handler::HttpHandler;
 use super::settings::ServiceConfig;
@@ -89,7 +89,90 @@ where
     }
 
     fn call(&mut self, req: Self::Request) -> Self::Future {
-        HttpChannel::new(self.settings.clone(), req, None)
+        HttpChannel::new(self.settings.clone(), req)
+    }
+}
+
+/// `NewService` implementation for HTTP1 transport
+pub struct H1Service<H, Io>
+where
+    H: HttpHandler,
+    Io: IoStream,
+{
+    settings: ServiceConfig<H>,
+    _t: PhantomData<Io>,
+}
+
+impl<H, Io> H1Service<H, Io>
+where
+    H: HttpHandler,
+    Io: IoStream,
+{
+    /// Create new `HttpService` instance.
+    pub fn new(settings: ServiceConfig<H>) -> Self {
+        H1Service {
+            settings,
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<H, Io> NewService for H1Service<H, Io>
+where
+    H: HttpHandler,
+    Io: IoStream,
+{
+    type Request = Io;
+    type Response = ();
+    type Error = HttpDispatchError;
+    type InitError = ();
+    type Service = H1ServiceHandler<H, Io>;
+    type Future = FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self) -> Self::Future {
+        ok(H1ServiceHandler::new(self.settings.clone()))
+    }
+}
+
+/// `Service` implementation for HTTP1 transport
+pub struct H1ServiceHandler<H, Io>
+where
+    H: HttpHandler,
+    Io: IoStream,
+{
+    settings: ServiceConfig<H>,
+    _t: PhantomData<Io>,
+}
+
+impl<H, Io> H1ServiceHandler<H, Io>
+where
+    H: HttpHandler,
+    Io: IoStream,
+{
+    fn new(settings: ServiceConfig<H>) -> H1ServiceHandler<H, Io> {
+        H1ServiceHandler {
+            settings,
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<H, Io> Service for H1ServiceHandler<H, Io>
+where
+    H: HttpHandler,
+    Io: IoStream,
+{
+    type Request = Io;
+    type Response = ();
+    type Error = HttpDispatchError;
+    type Future = H1Channel<Io, H>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        Ok(Async::Ready(()))
+    }
+
+    fn call(&mut self, req: Self::Request) -> Self::Future {
+        H1Channel::new(self.settings.clone(), req)
     }
 }
 

src/server/ssl/nativetls.rs

@@ -1,4 +1,4 @@
-use std::net::Shutdown;
+use std::net::{Shutdown, SocketAddr};
 use std::{io, time};
 
 use actix_net::ssl::TlsStream;
@@ -12,6 +12,11 @@ impl<Io: IoStream> IoStream for TlsStream<Io> {
         Ok(())
     }
 
+    #[inline]
+    fn peer_addr(&self) -> Option<SocketAddr> {
+        self.get_ref().get_ref().peer_addr()
+    }
+
     #[inline]
     fn set_nodelay(&mut self, nodelay: bool) -> io::Result<()> {
         self.get_mut().get_mut().set_nodelay(nodelay)

src/server/ssl/openssl.rs

@@ -1,4 +1,4 @@
-use std::net::Shutdown;
+use std::net::{Shutdown, SocketAddr};
 use std::{io, time};
 
 use actix_net::ssl;
@@ -65,6 +65,11 @@ impl<T: IoStream> IoStream for SslStream<T> {
         Ok(())
     }
 
+    #[inline]
+    fn peer_addr(&self) -> Option<SocketAddr> {
+        self.get_ref().get_ref().peer_addr()
+    }
+
     #[inline]
     fn set_nodelay(&mut self, nodelay: bool) -> io::Result<()> {
         self.get_mut().get_mut().set_nodelay(nodelay)

src/server/ssl/rustls.rs

@@ -1,4 +1,4 @@
-use std::net::Shutdown;
+use std::net::{Shutdown, SocketAddr};
 use std::{io, time};
 
 use actix_net::ssl; //::RustlsAcceptor;
@@ -65,6 +65,11 @@ impl<Io: IoStream> IoStream for TlsStream<Io, ServerSession> {
         Ok(())
     }
 
+    #[inline]
+    fn peer_addr(&self) -> Option<SocketAddr> {
+        self.get_ref().0.peer_addr()
+    }
+
     #[inline]
     fn set_nodelay(&mut self, nodelay: bool) -> io::Result<()> {
         self.get_mut().0.set_nodelay(nodelay)

tests/test_custom_pipeline.rs

@@ -0,0 +1,81 @@
+extern crate actix;
+extern crate actix_net;
+extern crate actix_web;
+
+use std::{thread, time};
+
+use actix::System;
+use actix_net::server::Server;
+use actix_net::service::NewServiceExt;
+use actix_web::server::{HttpService, KeepAlive, ServiceConfig, StreamConfiguration};
+use actix_web::{client, test, App, HttpRequest};
+
+#[test]
+fn test_custom_pipeline() {
+    let addr = test::TestServer::unused_addr();
+
+    thread::spawn(move || {
+        Server::new()
+            .bind("test", addr, move || {
+                let app = App::new()
+                    .route("/", http::Method::GET, |_: HttpRequest| "OK")
+                    .finish();
+                let settings = ServiceConfig::build(app)
+                    .keep_alive(KeepAlive::Disabled)
+                    .client_timeout(1000)
+                    .client_shutdown(1000)
+                    .server_hostname("localhost")
+                    .server_address(addr)
+                    .finish();
+
+                StreamConfiguration::new()
+                    .nodelay(true)
+                    .tcp_keepalive(Some(time::Duration::from_secs(10)))
+                    .and_then(HttpService::new(settings))
+            }).unwrap()
+            .run();
+    });
+
+    let mut sys = System::new("test");
+    {
+        let req = client::ClientRequest::get(format!("http://{}/", addr).as_str())
+            .finish()
+            .unwrap();
+        let response = sys.block_on(req.send()).unwrap();
+        assert!(response.status().is_success());
+    }
+}
+
+#[test]
+fn test_h1() {
+    use actix_web::server::H1Service;
+
+    let addr = test::TestServer::unused_addr();
+    thread::spawn(move || {
+        Server::new()
+            .bind("test", addr, move || {
+                let app = App::new()
+                    .route("/", http::Method::GET, |_: HttpRequest| "OK")
+                    .finish();
+                let settings = ServiceConfig::build(app)
+                    .keep_alive(KeepAlive::Disabled)
+                    .client_timeout(1000)
+                    .client_shutdown(1000)
+                    .server_hostname("localhost")
+                    .server_address(addr)
+                    .finish();
+
+                H1Service::new(settings)
+            }).unwrap()
+            .run();
+    });
+
+    let mut sys = System::new("test");
+    {
+        let req = client::ClientRequest::get(format!("http://{}/", addr).as_str())
+            .finish()
+            .unwrap();
+        let response = sys.block_on(req.send()).unwrap();
+        assert!(response.status().is_success());
+    }
+}

tests/test_server.rs

@@ -26,7 +26,6 @@ use std::io::{Read, Write};
 use std::sync::Arc;
 use std::{thread, time};
 
-use actix_net::server::Server;
 #[cfg(feature = "brotli")]
 use brotli2::write::{BrotliDecoder, BrotliEncoder};
 use bytes::{Bytes, BytesMut};
@@ -1223,48 +1222,6 @@ fn test_server_cookies() {
     }
 }
 
-#[test]
-fn test_custom_pipeline() {
-    use actix::System;
-    use actix_net::service::NewServiceExt;
-    use actix_web::server::{
-        HttpService, KeepAlive, ServiceConfig, StreamConfiguration,
-    };
-
-    let addr = test::TestServer::unused_addr();
-
-    thread::spawn(move || {
-        Server::new()
-            .bind("test", addr, move || {
-                let app = App::new()
-                    .route("/", http::Method::GET, |_: HttpRequest| "OK")
-                    .finish();
-                let settings = ServiceConfig::build(app)
-                    .keep_alive(KeepAlive::Disabled)
-                    .client_timeout(1000)
-                    .client_shutdown(1000)
-                    .server_hostname("localhost")
-                    .server_address(addr)
-                    .finish();
-
-                StreamConfiguration::new()
-                    .nodelay(true)
-                    .tcp_keepalive(Some(time::Duration::from_secs(10)))
-                    .and_then(HttpService::new(settings))
-            }).unwrap()
-            .run();
-    });
-
-    let mut sys = System::new("test");
-    {
-        let req = client::ClientRequest::get(format!("http://{}/", addr).as_str())
-            .finish()
-            .unwrap();
-        let response = sys.block_on(req.send()).unwrap();
-        assert!(response.status().is_success());
-    }
-}
-
 #[test]
 fn test_slow_request() {
     use actix::System;