calculate response parameters

src/h1/codec.rs

@@ -10,7 +10,7 @@ use body::Body;
 use error::ParseError;
 use helpers;
 use http::header::{HeaderValue, CONNECTION, CONTENT_LENGTH, DATE, TRANSFER_ENCODING};
-use http::Version;
+use http::{Method, Version};
 use httpresponse::HttpResponse;
 use request::RequestPool;
 use server::output::{ResponseInfo, ResponseLength};
@@ -27,6 +27,8 @@ pub enum OutMessage {
 pub struct Codec {
     decoder: H1Decoder,
     encoder: H1Writer,
+    head: bool,
+    version: Version,
 }
 
 impl Codec {
@@ -40,6 +42,8 @@ impl Codec {
         Codec {
             decoder: H1Decoder::with_pool(pool),
             encoder: H1Writer::new(),
+            head: false,
+            version: Version::HTTP_11,
         }
     }
 }
@@ -49,7 +53,17 @@ impl Decoder for Codec {
     type Error = ParseError;
 
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
-        self.decoder.decode(src)
+        let res = self.decoder.decode(src);
+
+        match res {
+            Ok(Some(InMessage::Message(ref req)))
+            | Ok(Some(InMessage::MessageWithPayload(ref req))) => {
+                self.head = req.inner.method == Method::HEAD;
+                self.version = req.inner.version;
+            }
+            _ => (),
+        }
+        res
     }
 }
 
@@ -62,7 +76,7 @@ impl Encoder for Codec {
     ) -> Result<(), Self::Error> {
         match item {
             OutMessage::Response(res) => {
-                self.encoder.encode(res, dst)?;
+                self.encoder.encode(res, dst, self.head, self.version)?;
             }
             OutMessage::Payload(bytes) => {
                 dst.extend_from_slice(&bytes);
@@ -87,6 +101,7 @@ struct H1Writer {
     flags: Flags,
     written: u64,
     headers_size: u32,
+    info: ResponseInfo,
 }
 
 impl H1Writer {
@@ -95,6 +110,7 @@ impl H1Writer {
             flags: Flags::empty(),
             written: 0,
             headers_size: 0,
+            info: ResponseInfo::default(),
         }
     }
 
@@ -116,10 +132,11 @@ impl H1Writer {
     }
 
     fn encode(
-        &mut self, mut msg: HttpResponse, buffer: &mut BytesMut,
+        &mut self, mut msg: HttpResponse, buffer: &mut BytesMut, head: bool,
+        version: Version,
     ) -> io::Result<()> {
         // prepare task
-        let info = ResponseInfo::new(false); // req.inner.method == Method::HEAD);
+        self.info.update(&mut msg, head, version);
 
         //if msg.keep_alive().unwrap_or_else(|| req.keep_alive()) {
         //self.flags = Flags::STARTED | Flags::KEEPALIVE;
@@ -166,7 +183,7 @@ impl H1Writer {
             buffer.extend_from_slice(reason);
 
             // content length
-            match info.length {
+            match self.info.length {
                 ResponseLength::Chunked => {
                     buffer.extend_from_slice(b"\r\ntransfer-encoding: chunked\r\n")
                 }
@@ -183,11 +200,6 @@ impl H1Writer {
                 }
                 ResponseLength::None => buffer.extend_from_slice(b"\r\n"),
             }
-            if let Some(ce) = info.content_encoding {
-                buffer.extend_from_slice(b"content-encoding: ");
-                buffer.extend_from_slice(ce.as_ref());
-                buffer.extend_from_slice(b"\r\n");
-            }
 
             // write headers
             let mut pos = 0;
@@ -197,7 +209,7 @@ impl H1Writer {
             for (key, value) in msg.headers() {
                 match *key {
                     TRANSFER_ENCODING => continue,
-                    CONTENT_LENGTH => match info.length {
+                    CONTENT_LENGTH => match self.info.length {
                         ResponseLength::None => (),
                         _ => continue,
                     },

src/h1/dispatcher.rs

@@ -1,7 +1,7 @@
 // #![allow(unused_imports, unused_variables, dead_code)]
 use std::collections::VecDeque;
 use std::fmt::{Debug, Display};
-// use std::time::{Duration, Instant};
+use std::time::Instant;
 
 use actix_net::service::Service;
 
@@ -9,7 +9,7 @@ use futures::{Async, AsyncSink, Future, Poll, Sink, Stream};
 use tokio_codec::Framed;
 // use tokio_current_thread::spawn;
 use tokio_io::{AsyncRead, AsyncWrite};
-// use tokio_timer::Delay;
+use tokio_timer::Delay;
 
 use error::{ParseError, PayloadError};
 use payload::{Payload, PayloadStatus, PayloadWriter};
@@ -47,12 +47,14 @@ where
     flags: Flags,
     framed: Framed<T, Codec>,
     error: Option<DispatchError<S::Error>>,
+    config: ServiceConfig,
 
     state: State<S>,
     payload: Option<PayloadType>,
     messages: VecDeque<Request>,
 
-    config: ServiceConfig,
+    ka_expire: Instant,
+    ka_timer: Option<Delay>,
 }
 
 enum State<S: Service> {
@@ -81,9 +83,28 @@ where
 {
     /// Create http/1 dispatcher.
     pub fn new(stream: T, config: ServiceConfig, service: S) -> Self {
-        let flags = Flags::FLUSHED;
+        Dispatcher::with_timeout(stream, config, None, service)
+    }
+
+    /// Create http/1 dispatcher with slow request timeout.
+    pub fn with_timeout(
+        stream: T, config: ServiceConfig, timeout: Option<Delay>, service: S,
+    ) -> Self {
+        let flags = if config.keep_alive_enabled() {
+            Flags::KEEPALIVE | Flags::KEEPALIVE_ENABLED | Flags::FLUSHED
+        } else {
+            Flags::FLUSHED
+        };
         let framed = Framed::new(stream, Codec::new());
 
+        let (ka_expire, ka_timer) = if let Some(delay) = timeout {
+            (delay.deadline(), Some(delay))
+        } else if let Some(delay) = config.keep_alive_timer() {
+            (delay.deadline(), Some(delay))
+        } else {
+            (config.now(), None)
+        };
+
         Dispatcher {
             payload: None,
             state: State::None,
@@ -93,6 +114,8 @@ where
             flags,
             framed,
             config,
+            ka_expire,
+            ka_timer,
         }
     }
 
@@ -358,8 +381,64 @@ where
             }
         }
 
+        if self.ka_timer.is_some() && updated {
+            if let Some(expire) = self.config.keep_alive_expire() {
+                self.ka_expire = expire;
+            }
+        }
         Ok(updated)
     }
+
+    /// keep-alive timer
+    fn poll_keepalive(&mut self) -> Result<(), DispatchError<S::Error>> {
+        if let Some(ref mut timer) = self.ka_timer {
+            match timer.poll() {
+                Ok(Async::Ready(_)) => {
+                    if timer.deadline() >= self.ka_expire {
+                        // check for any outstanding request handling
+                        if self.state.is_empty() && self.messages.is_empty() {
+                            // if we get timer during shutdown, just drop connection
+                            if self.flags.contains(Flags::SHUTDOWN) {
+                                return Err(DispatchError::DisconnectTimeout);
+                            } else if !self.flags.contains(Flags::STARTED) {
+                                // timeout on first request (slow request) return 408
+                                trace!("Slow request timeout");
+                                self.flags
+                                    .insert(Flags::STARTED | Flags::READ_DISCONNECTED);
+                                self.state =
+                                    State::SendResponse(Some(OutMessage::Response(
+                                        HttpResponse::RequestTimeout().finish(),
+                                    )));
+                            } else {
+                                trace!("Keep-alive timeout, close connection");
+                                self.flags.insert(Flags::SHUTDOWN);
+
+                                // start shutdown timer
+                                if let Some(deadline) =
+                                    self.config.client_disconnect_timer()
+                                {
+                                    timer.reset(deadline)
+                                } else {
+                                    return Ok(());
+                                }
+                            }
+                        } else if let Some(deadline) = self.config.keep_alive_expire() {
+                            timer.reset(deadline)
+                        }
+                    } else {
+                        timer.reset(self.ka_expire)
+                    }
+                }
+                Ok(Async::NotReady) => (),
+                Err(e) => {
+                    error!("Timer error {:?}", e);
+                    return Err(DispatchError::Unknown);
+                }
+            }
+        }
+
+        Ok(())
+    }
 }
 
 impl<T, S> Future for Dispatcher<T, S>
@@ -373,6 +452,8 @@ where
 
     #[inline]
     fn poll(&mut self) -> Poll<(), Self::Error> {
+        self.poll_keepalive()?;
+
         // shutdown
         if self.flags.contains(Flags::SHUTDOWN) {
             if self.flags.contains(Flags::WRITE_DISCONNECTED) {