initial commit

Pulled the code from firecracker-microvm/src/micro-http.

Signed-off-by: George Pisaltu <gpl@amazon.com>
Signed-off-by: Andreea Florescu <fandree@amazon.com>

Cargo.lock

@@ -0,0 +1,32 @@
+# This file is automatically @generated by Cargo.
+# It is not intended for manual editing.
+[[package]]
+name = "bitflags"
+version = "1.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
+[[package]]
+name = "epoll"
+version = "4.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "bitflags 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.66 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "libc"
+version = "0.2.66"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
+[[package]]
+name = "micro_http"
+version = "0.1.0"
+dependencies = [
+ "epoll 4.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[metadata]
+"checksum bitflags 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "cf1de2fe8c75bc145a2f577add951f8134889b4795d47466a54a5c846d691693"
+"checksum epoll 4.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "990bcfe26bea89669ede68c3f970f61d02568dbc8660317c98d805ea4e710685"
+"checksum libc 0.2.66 (registry+https://github.com/rust-lang/crates.io-index)" = "d515b1f41455adea1313a4a2ac8a8a477634fbae63cc6100e3aebb207ce61558"

Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "micro_http"
+version = "0.1.0"
+authors = ["Amazon Firecracker team <firecracker-devel@amazon.com>"]
+
+[dependencies]
+epoll = ">=4.0.1"

src/common/headers.rs

@@ -0,0 +1,397 @@
+// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::result::Result;
+
+use RequestError;
+
+/// Wrapper over an HTTP Header type.
+#[derive(Debug, Eq, Hash, PartialEq)]
+pub enum Header {
+    /// Header `Content-Length`.
+    ContentLength,
+    /// Header `Content-Type`.
+    ContentType,
+    /// Header `Expect`.
+    Expect,
+    /// Header `Transfer-Encoding`.
+    TransferEncoding,
+    /// Header `Server`.
+    Server,
+}
+
+impl Header {
+    pub fn raw(&self) -> &'static [u8] {
+        match self {
+            Header::ContentLength => b"Content-Length",
+            Header::ContentType => b"Content-Type",
+            Header::Expect => b"Expect",
+            Header::TransferEncoding => b"Transfer-Encoding",
+            Header::Server => b"Server",
+        }
+    }
+
+    fn try_from(string: &[u8]) -> Result<Self, RequestError> {
+        if let Ok(utf8_string) = String::from_utf8(string.to_vec()) {
+            match utf8_string.trim() {
+                "Content-Length" => Ok(Header::ContentLength),
+                "Content-Type" => Ok(Header::ContentType),
+                "Expect" => Ok(Header::Expect),
+                "Transfer-Encoding" => Ok(Header::TransferEncoding),
+                "Server" => Ok(Header::Server),
+                _ => Err(RequestError::InvalidHeader),
+            }
+        } else {
+            Err(RequestError::InvalidRequest)
+        }
+    }
+}
+
+/// Wrapper over the list of headers associated with a Request that we need
+/// in order to parse the request correctly and be able to respond to it.
+///
+/// The only `Content-Type`s supported are `text/plain` and `application/json`, which are both
+/// in plain text actually and don't influence our parsing process.
+///
+/// All the other possible header fields are not necessary in order to serve this connection
+/// and, thus, are not of interest to us. However, we still look for header fields that might
+/// invalidate our request as we don't support the full set of HTTP/1.1 specification.
+/// Such header entries are "Transfer-Encoding: identity; q=0", which means a compression
+/// algorithm is applied to the body of the request, or "Expect: 103-checkpoint".
+#[derive(Debug)]
+pub struct Headers {
+    /// The `Content-Length` header field tells us how many bytes we need to receive
+    /// from the source after the headers.
+    content_length: i32,
+    /// The `Expect` header field is set when the headers contain the entry "Expect: 100-continue".
+    /// This means that, per HTTP/1.1 specifications, we must send a response with the status code
+    /// 100 after we have received the headers in order to receive the body of the request. This
+    /// field should be known immediately after parsing the headers.
+    expect: bool,
+    /// `Chunked` is a possible value of the `Transfer-Encoding` header field and every HTTP/1.1
+    /// server must support it. It is useful only when receiving the body of the request and should
+    /// be known immediately after parsing the headers.
+    chunked: bool,
+}
+
+impl Headers {
+    /// By default Requests are created with no headers.
+    pub fn default() -> Headers {
+        Headers {
+            content_length: 0,
+            expect: false,
+            chunked: false,
+        }
+    }
+
+    /// Expects one header line and parses it, updating the header structure or returning an
+    /// error if the header is invalid.
+    ///
+    /// # Errors
+    /// `UnsupportedHeader` is returned when the parsed header line is not of interest
+    /// to us or when it is unrecognizable.
+    /// `InvalidHeader` is returned when the parsed header is formatted incorrectly or suggests
+    /// that the client is using HTTP features that we do not support in this implementation,
+    /// which invalidates the request.
+    pub fn parse_header_line(&mut self, header_line: &[u8]) -> Result<(), RequestError> {
+        // Headers must be ASCII, so also UTF-8 valid.
+        match std::str::from_utf8(header_line) {
+            Ok(headers_str) => {
+                let entry = headers_str.split(": ").collect::<Vec<&str>>();
+                if entry.len() != 2 {
+                    return Err(RequestError::InvalidHeader);
+                }
+                if let Ok(head) = Header::try_from(entry[0].as_bytes()) {
+                    match head {
+                        Header::ContentLength => {
+                            let try_numeric: Result<i32, std::num::ParseIntError> =
+                                std::str::FromStr::from_str(entry[1].trim());
+                            if let Ok(content_length) = try_numeric {
+                                self.content_length = content_length;
+                                Ok(())
+                            } else {
+                                Err(RequestError::InvalidHeader)
+                            }
+                        }
+                        Header::ContentType => {
+                            match MediaType::try_from(entry[1].trim().as_bytes()) {
+                                Ok(_) => Ok(()),
+                                Err(_) => Err(RequestError::UnsupportedHeader),
+                            }
+                        }
+                        Header::TransferEncoding => match entry[1].trim() {
+                            "chunked" => {
+                                self.chunked = true;
+                                Ok(())
+                            }
+                            "identity; q=0" => Err(RequestError::InvalidHeader),
+                            _ => Err(RequestError::UnsupportedHeader),
+                        },
+                        Header::Expect => match entry[1].trim() {
+                            "100-continue" => {
+                                self.expect = true;
+                                Ok(())
+                            }
+                            _ => Err(RequestError::InvalidHeader),
+                        },
+                        Header::Server => Ok(()),
+                    }
+                } else {
+                    Err(RequestError::UnsupportedHeader)
+                }
+            }
+            _ => Err(RequestError::InvalidHeader),
+        }
+    }
+
+    /// Returns the content length of the body.
+    pub fn content_length(&self) -> i32 {
+        self.content_length
+    }
+
+    /// Returns `true` if the transfer encoding is chunked.
+    #[allow(unused)]
+    pub fn chunked(&self) -> bool {
+        self.chunked
+    }
+
+    /// Returns `true` if the client is expecting the code 100.
+    #[allow(unused)]
+    pub fn expect(&self) -> bool {
+        self.expect
+    }
+
+    #[cfg(test)]
+    pub fn new(content_length: i32, expect: bool, chunked: bool) -> Self {
+        Headers {
+            content_length,
+            expect,
+            chunked,
+        }
+    }
+
+    /// Parses a byte slice into a Headers structure for a HTTP request.
+    ///
+    /// The byte slice is expected to have the following format: </br>
+    ///     * Request Header Lines "<header_line> CRLF"- Optional </br>
+    /// There can be any number of request headers, including none, followed by
+    /// an extra sequence of Carriage Return and Line Feed.
+    /// All header fields are parsed. However, only the ones present in the
+    /// [`Headers`](struct.Headers.html) struct are relevant to us and stored
+    /// for future use.
+    ///
+    /// # Errors
+    /// The function returns `InvalidHeader` when parsing the byte stream fails.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// extern crate micro_http;
+    /// use micro_http::Headers;
+    ///
+    /// let request_headers = Headers::try_from(b"Content-Length: 55\r\n\r\n");
+    /// ```
+    pub fn try_from(bytes: &[u8]) -> Result<Headers, RequestError> {
+        // Headers must be ASCII, so also UTF-8 valid.
+        if let Ok(text) = std::str::from_utf8(bytes) {
+            let mut headers = Headers::default();
+
+            let header_lines = text.split("\r\n");
+            for header_line in header_lines {
+                if header_line.is_empty() {
+                    break;
+                }
+                match headers.parse_header_line(header_line.as_bytes()) {
+                    Ok(_) | Err(RequestError::UnsupportedHeader) => continue,
+                    Err(e) => return Err(e),
+                };
+            }
+            return Ok(headers);
+        }
+        Err(RequestError::InvalidRequest)
+    }
+}
+
+/// Wrapper over supported Media Types.
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub enum MediaType {
+    /// Media Type: "text/plain".
+    PlainText,
+    /// Media Type: "application/json".
+    ApplicationJson,
+}
+
+impl Default for MediaType {
+    fn default() -> Self {
+        MediaType::ApplicationJson
+    }
+}
+
+impl MediaType {
+    fn try_from(bytes: &[u8]) -> Result<Self, RequestError> {
+        if bytes.is_empty() {
+            return Err(RequestError::InvalidRequest);
+        }
+        let utf8_slice =
+            String::from_utf8(bytes.to_vec()).map_err(|_| RequestError::InvalidRequest)?;
+        match utf8_slice.as_str().trim() {
+            "text/plain" => Ok(MediaType::PlainText),
+            "application/json" => Ok(MediaType::ApplicationJson),
+            _ => Err(RequestError::InvalidRequest),
+        }
+    }
+
+    /// Returns a static string representation of the object.
+    pub fn as_str(self) -> &'static str {
+        match self {
+            MediaType::PlainText => "text/plain",
+            MediaType::ApplicationJson => "application/json",
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_default() {
+        let headers = Headers::default();
+        assert_eq!(headers.content_length(), 0);
+        assert_eq!(headers.chunked(), false);
+        assert_eq!(headers.expect(), false);
+    }
+
+    #[test]
+    fn test_try_from_media() {
+        assert_eq!(
+            MediaType::try_from(b"application/json").unwrap(),
+            MediaType::ApplicationJson
+        );
+
+        assert_eq!(
+            MediaType::try_from(b"text/plain").unwrap(),
+            MediaType::PlainText
+        );
+
+        assert_eq!(
+            MediaType::try_from(b"").unwrap_err(),
+            RequestError::InvalidRequest
+        );
+
+        assert_eq!(
+            MediaType::try_from(b"application/json-patch").unwrap_err(),
+            RequestError::InvalidRequest
+        );
+    }
+
+    #[test]
+    fn test_media_as_str() {
+        let media_type = MediaType::ApplicationJson;
+        assert_eq!(media_type.as_str(), "application/json");
+
+        let media_type = MediaType::PlainText;
+        assert_eq!(media_type.as_str(), "text/plain");
+    }
+
+    #[test]
+    fn test_try_from_headers() {
+        // Valid headers.
+        assert_eq!(
+            Headers::try_from(
+                b"Last-Modified: Tue, 15 Nov 1994 12:45:26 GMT\r\nContent-Length: 55\r\n\r\n"
+            )
+            .unwrap()
+            .content_length,
+            55
+        );
+
+        let bytes: [u8; 10] = [130, 140, 150, 130, 140, 150, 130, 140, 150, 160];
+        // Invalid headers.
+        assert!(Headers::try_from(&bytes[..]).is_err());
+    }
+
+    #[test]
+    fn test_parse_header_line() {
+        let mut header = Headers::default();
+
+        // Invalid header syntax.
+        assert_eq!(
+            header.parse_header_line(b"Expect"),
+            Err(RequestError::InvalidHeader)
+        );
+
+        // Invalid content length.
+        assert_eq!(
+            header.parse_header_line(b"Content-Length: five"),
+            Err(RequestError::InvalidHeader)
+        );
+
+        // Invalid transfer encoding.
+        assert_eq!(
+            header.parse_header_line(b"Transfer-Encoding: gzip"),
+            Err(RequestError::UnsupportedHeader)
+        );
+
+        // Invalid expect.
+        assert_eq!(
+            header
+                .parse_header_line(b"Expect: 102-processing")
+                .unwrap_err(),
+            RequestError::InvalidHeader
+        );
+
+        // Unsupported media type.
+        assert_eq!(
+            header
+                .parse_header_line(b"Content-Type: application/json-patch")
+                .unwrap_err(),
+            RequestError::UnsupportedHeader
+        );
+
+        // Invalid input format.
+        let input: [u8; 10] = [130, 140, 150, 130, 140, 150, 130, 140, 150, 160];
+        assert_eq!(
+            header.parse_header_line(&input[..]).unwrap_err(),
+            RequestError::InvalidHeader
+        );
+
+        // Test valid transfer encoding.
+        assert!(header
+            .parse_header_line(b"Transfer-Encoding: chunked")
+            .is_ok());
+        assert!(header.chunked());
+
+        // Test valid expect.
+        assert!(header.parse_header_line(b"Expect: 100-continue").is_ok());
+        assert!(header.expect());
+
+        // Test valid media type.
+        assert!(header
+            .parse_header_line(b"Content-Type: application/json")
+            .is_ok());
+    }
+
+    #[test]
+    fn test_header_try_from() {
+        // Bad header.
+        assert_eq!(
+            Header::try_from(b"Encoding").unwrap_err(),
+            RequestError::InvalidHeader
+        );
+
+        // Invalid encoding.
+        let input: [u8; 10] = [130, 140, 150, 130, 140, 150, 130, 140, 150, 160];
+        assert_eq!(
+            Header::try_from(&input[..]).unwrap_err(),
+            RequestError::InvalidRequest
+        );
+
+        // Test valid headers.
+        let header = Header::try_from(b"Expect").unwrap();
+        assert_eq!(header.raw(), b"Expect");
+
+        let header = Header::try_from(b"Transfer-Encoding").unwrap();
+        assert_eq!(header.raw(), b"Transfer-Encoding");
+    }
+}

src/common/mod.rs

@@ -0,0 +1,355 @@
+// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::fmt::{Display, Error, Formatter};
+
+pub mod headers;
+
+pub mod ascii {
+    pub const CR: u8 = b'\r';
+    pub const COLON: u8 = b':';
+    pub const LF: u8 = b'\n';
+    pub const SP: u8 = b' ';
+    pub const CRLF_LEN: usize = 2;
+}
+
+/// Errors associated with parsing the HTTP Request from a u8 slice.
+#[derive(Debug, PartialEq)]
+pub enum RequestError {
+    /// The HTTP Method is not supported or it is invalid.
+    InvalidHttpMethod(&'static str),
+    /// Request URI is invalid.
+    InvalidUri(&'static str),
+    /// The HTTP Version in the Request is not supported or it is invalid.
+    InvalidHttpVersion(&'static str),
+    /// The header specified may be valid, but is not supported by this HTTP implementation.
+    UnsupportedHeader,
+    /// Header specified is invalid.
+    InvalidHeader,
+    /// The Request is invalid and cannot be served.
+    InvalidRequest,
+}
+
+impl Display for RequestError {
+    fn fmt(&self, f: &mut Formatter) -> Result<(), Error> {
+        match self {
+            RequestError::InvalidHttpMethod(inner) => write!(f, "Invalid HTTP Method: {}", inner),
+            RequestError::InvalidUri(inner) => write!(f, "Invalid URI: {}", inner),
+            RequestError::InvalidHttpVersion(inner) => write!(f, "Invalid HTTP Version: {}", inner),
+            RequestError::UnsupportedHeader => write!(f, "Unsupported header."),
+            RequestError::InvalidHeader => write!(f, "Invalid header."),
+            RequestError::InvalidRequest => write!(f, "Invalid request."),
+        }
+    }
+}
+
+/// Errors associated with a HTTP Connection.
+#[derive(Debug)]
+pub enum ConnectionError {
+    /// The request parsing has failed.
+    ParseError(RequestError),
+    /// Could not perform a stream operation successfully.
+    StreamError(std::io::Error),
+    /// Attempted to read or write on a closed connection.
+    ConnectionClosed,
+    /// Attempted to write on a stream when there was nothing to write.
+    InvalidWrite,
+}
+
+impl Display for ConnectionError {
+    fn fmt(&self, f: &mut Formatter) -> Result<(), Error> {
+        match self {
+            ConnectionError::ParseError(inner) => write!(f, "Parsing error: {}", inner),
+            ConnectionError::StreamError(inner) => write!(f, "Stream error: {}", inner),
+            ConnectionError::ConnectionClosed => write!(f, "Connection closed."),
+            ConnectionError::InvalidWrite => write!(f, "Invalid write attempt."),
+        }
+    }
+}
+
+/// Errors pertaining to `HttpServer`.
+#[derive(Debug)]
+pub enum ServerError {
+    /// Epoll operations failed.
+    IOError(std::io::Error),
+    /// Error from one of the connections.
+    ConnectionError(ConnectionError),
+    /// Server maximum capacity has been reached.
+    ServerFull,
+}
+
+impl Display for ServerError {
+    fn fmt(&self, f: &mut Formatter) -> Result<(), Error> {
+        match self {
+            ServerError::IOError(inner) => write!(f, "IO error: {}", inner),
+            ServerError::ConnectionError(inner) => write!(f, "Connection error: {}", inner),
+            ServerError::ServerFull => write!(f, "Server is full."),
+        }
+    }
+}
+
+/// The Body associated with an HTTP Request or Response.
+///
+/// ## Examples
+/// ```
+/// extern crate micro_http;
+/// use micro_http::Body;
+/// let body = Body::new("This is a test body.".to_string());
+/// assert_eq!(body.raw(), b"This is a test body.");
+/// assert_eq!(body.len(), 20);
+/// ```
+#[derive(Clone, Debug, PartialEq)]
+pub struct Body {
+    /// Body of the HTTP message as bytes.
+    pub body: Vec<u8>,
+}
+
+impl Body {
+    /// Creates a new `Body` from a `String` input.
+    pub fn new<T: Into<Vec<u8>>>(body: T) -> Self {
+        Body { body: body.into() }
+    }
+
+    /// Returns the body as an `u8 slice`.
+    pub fn raw(&self) -> &[u8] {
+        self.body.as_slice()
+    }
+
+    /// Returns the length of the `Body`.
+    pub fn len(&self) -> usize {
+        self.body.len()
+    }
+
+    /// Checks if the body is empty, ie with zero length
+    pub fn is_empty(&self) -> bool {
+        self.body.len() == 0
+    }
+}
+
+/// Supported HTTP Methods.
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub enum Method {
+    /// GET Method.
+    Get,
+    /// PUT Method.
+    Put,
+    /// PATCH Method.
+    Patch,
+}
+
+impl Method {
+    /// Returns a `Method` object if the parsing of `bytes` is successful.
+    ///
+    /// The method is case sensitive. A call to try_from with the input b"get" will return
+    /// an error, but when using the input b"GET", it returns Method::Get.
+    ///
+    /// # Errors
+    /// Returns `RequestError` if the method specified by `bytes` is unsupported.
+    pub fn try_from(bytes: &[u8]) -> Result<Self, RequestError> {
+        match bytes {
+            b"GET" => Ok(Method::Get),
+            b"PUT" => Ok(Method::Put),
+            b"PATCH" => Ok(Method::Patch),
+            _ => Err(RequestError::InvalidHttpMethod("Unsupported HTTP method.")),
+        }
+    }
+
+    /// Returns an `u8 slice` corresponding to the Method.
+    pub fn raw(self) -> &'static [u8] {
+        match self {
+            Method::Get => b"GET",
+            Method::Put => b"PUT",
+            Method::Patch => b"PATCH",
+        }
+    }
+}
+
+/// Supported HTTP Versions.
+///
+/// # Examples
+/// ```
+/// extern crate micro_http;
+/// use micro_http::Version;
+/// let version = Version::try_from(b"HTTP/1.1");
+/// assert!(version.is_ok());
+///
+/// let version = Version::try_from(b"http/1.1");
+/// assert!(version.is_err());
+/// ```
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub enum Version {
+    /// HTTP/1.0
+    Http10,
+    /// HTTP/1.1
+    Http11,
+}
+
+impl Version {
+    /// HTTP Version as an `u8 slice`.
+    pub fn raw(self) -> &'static [u8] {
+        match self {
+            Version::Http10 => b"HTTP/1.0",
+            Version::Http11 => b"HTTP/1.1",
+        }
+    }
+
+    /// Creates a new HTTP Version from an `u8 slice`.
+    ///
+    /// The supported versions are HTTP/1.0 and HTTP/1.1.
+    /// The version is case sensitive and the accepted input is upper case.
+    ///
+    /// # Errors
+    /// Returns a `RequestError` when the version is not supported.
+    pub fn try_from(bytes: &[u8]) -> Result<Self, RequestError> {
+        match bytes {
+            b"HTTP/1.0" => Ok(Version::Http10),
+            b"HTTP/1.1" => Ok(Version::Http11),
+            _ => Err(RequestError::InvalidHttpVersion(
+                "Unsupported HTTP version.",
+            )),
+        }
+    }
+
+    /// Returns the default HTTP version = HTTP/1.1.
+    pub fn default() -> Self {
+        Version::Http11
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    impl PartialEq for ConnectionError {
+        fn eq(&self, other: &Self) -> bool {
+            use self::ConnectionError::*;
+            match (self, other) {
+                (ParseError(_), ParseError(_)) => true,
+                (ConnectionClosed, ConnectionClosed) => true,
+                (StreamError(_), StreamError(_)) => true,
+                (InvalidWrite, InvalidWrite) => true,
+                _ => false,
+            }
+        }
+    }
+
+    #[test]
+    fn test_version() {
+        // Tests for raw()
+        assert_eq!(Version::Http10.raw(), b"HTTP/1.0");
+        assert_eq!(Version::Http11.raw(), b"HTTP/1.1");
+
+        // Tests for try_from()
+        assert_eq!(Version::try_from(b"HTTP/1.0").unwrap(), Version::Http10);
+        assert_eq!(Version::try_from(b"HTTP/1.1").unwrap(), Version::Http11);
+        assert_eq!(
+            Version::try_from(b"HTTP/2.0").unwrap_err(),
+            RequestError::InvalidHttpVersion("Unsupported HTTP version.")
+        );
+
+        // Test for default()
+        assert_eq!(Version::default(), Version::Http11);
+    }
+
+    #[test]
+    fn test_method() {
+        // Test for raw
+        assert_eq!(Method::Get.raw(), b"GET");
+        assert_eq!(Method::Put.raw(), b"PUT");
+        assert_eq!(Method::Patch.raw(), b"PATCH");
+
+        // Tests for try_from
+        assert_eq!(Method::try_from(b"GET").unwrap(), Method::Get);
+        assert_eq!(Method::try_from(b"PUT").unwrap(), Method::Put);
+        assert_eq!(Method::try_from(b"PATCH").unwrap(), Method::Patch);
+        assert_eq!(
+            Method::try_from(b"POST").unwrap_err(),
+            RequestError::InvalidHttpMethod("Unsupported HTTP method.")
+        );
+    }
+
+    #[test]
+    fn test_body() {
+        let body = Body::new("".to_string());
+        // Test for is_empty
+        assert!(body.is_empty());
+        let body = Body::new("This is a body.".to_string());
+        // Test for len
+        assert_eq!(body.len(), 15);
+        // Test for raw
+        assert_eq!(body.raw(), b"This is a body.");
+    }
+
+    #[test]
+    fn test_display_request_error() {
+        assert_eq!(
+            format!("{}", RequestError::InvalidHttpMethod("test")),
+            "Invalid HTTP Method: test"
+        );
+        assert_eq!(
+            format!("{}", RequestError::InvalidUri("test")),
+            "Invalid URI: test"
+        );
+        assert_eq!(
+            format!("{}", RequestError::InvalidHttpVersion("test")),
+            "Invalid HTTP Version: test"
+        );
+        assert_eq!(
+            format!("{}", RequestError::InvalidHeader),
+            "Invalid header."
+        );
+        assert_eq!(
+            format!("{}", RequestError::UnsupportedHeader),
+            "Unsupported header."
+        );
+        assert_eq!(
+            format!("{}", RequestError::InvalidRequest),
+            "Invalid request."
+        );
+    }
+
+    #[test]
+    fn test_display_connection_error() {
+        assert_eq!(
+            format!(
+                "{}",
+                ConnectionError::ParseError(RequestError::InvalidRequest)
+            ),
+            "Parsing error: Invalid request."
+        );
+        assert_eq!(
+            format!(
+                "{}",
+                ConnectionError::StreamError(std::io::Error::from_raw_os_error(11))
+            ),
+            "Stream error: Resource temporarily unavailable (os error 11)"
+        );
+        assert_eq!(
+            format!("{}", ConnectionError::ConnectionClosed),
+            "Connection closed."
+        );
+        assert_eq!(
+            format!("{}", ConnectionError::InvalidWrite),
+            "Invalid write attempt."
+        );
+    }
+
+    #[test]
+    fn test_display_server_error() {
+        assert_eq!(
+            format!(
+                "{}",
+                ServerError::ConnectionError(ConnectionError::ConnectionClosed)
+            ),
+            "Connection error: Connection closed."
+        );
+        assert_eq!(format!("{}", ServerError::ServerFull), "Server is full.");
+        assert_eq!(
+            format!(
+                "{}",
+                ServerError::IOError(std::io::Error::from_raw_os_error(11))
+            ),
+            "IO error: Resource temporarily unavailable (os error 11)"
+        );
+    }
+}

src/connection.rs

@@ -0,0 +1,783 @@
+// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::collections::VecDeque;
+use std::io::{Read, Write};
+
+use common::ascii::{CR, CRLF_LEN, LF};
+use common::Body;
+pub use common::{ConnectionError, RequestError};
+use headers::Headers;
+use request::{find, Request, RequestLine};
+use response::{Response, StatusCode};
+
+const BUFFER_SIZE: usize = 1024;
+
+/// Describes the state machine of an HTTP connection.
+pub enum ConnectionState {
+    WaitingForRequestLine,
+    WaitingForHeaders,
+    WaitingForBody,
+    RequestReady,
+}
+
+/// A wrapper over a HTTP Connection.
+pub struct HttpConnection<T> {
+    /// A partial request that is still being received.
+    pending_request: Option<Request>,
+    /// Stream implementing `Read` and `Write`, capable of sending and
+    /// receiving bytes.
+    stream: T,
+    /// The state of the connection regarding the current request that
+    /// is being processed.
+    state: ConnectionState,
+    /// Buffer where we store the bytes we read from the stream.
+    buffer: [u8; BUFFER_SIZE],
+    /// The index in the buffer from where we have to start reading in
+    /// the next `try_read` call.
+    read_cursor: usize,
+    /// Contains all bytes pertaining to the body of the request that
+    /// is currently being processed.
+    body_vec: Vec<u8>,
+    /// Represents how many bytes from the body of the request are still
+    /// to be read.
+    body_bytes_to_be_read: i32,
+    /// A queue of all requests that have been fully received and parsed.
+    parsed_requests: VecDeque<Request>,
+    /// A queue of requests that are waiting to be sent.
+    response_queue: VecDeque<Response>,
+    /// A buffer containing the bytes of a response that is currently
+    /// being sent.
+    response_buffer: Option<Vec<u8>>,
+}
+
+impl<T: Read + Write> HttpConnection<T> {
+    /// Creates an empty connection.
+    pub fn new(stream: T) -> Self {
+        HttpConnection {
+            pending_request: None,
+            stream,
+            state: ConnectionState::WaitingForRequestLine,
+            buffer: [0; BUFFER_SIZE],
+            read_cursor: 0,
+            body_vec: vec![],
+            body_bytes_to_be_read: 0,
+            parsed_requests: VecDeque::new(),
+            response_queue: VecDeque::new(),
+            response_buffer: None,
+        }
+    }
+
+    /// Tries to read new bytes from the stream and automatically update the request.
+    /// Meant to be used only with non-blocking streams and an `EPOLL` structure.
+    /// Should be called whenever an `EPOLLIN` event is signaled.
+    ///
+    /// # Errors
+    /// `StreamError` is returned when an IO operation fails.
+    /// `ConnectionClosed` is returned when a client prematurely closes the connection.
+    /// `ParseError` is returned when a parsing operation fails.
+    pub fn try_read(&mut self) -> Result<(), ConnectionError> {
+        // Read some bytes from the stream, which will be appended to what is already
+        // present in the buffer from a previous call of `try_read`. There are already
+        // `read_cursor` bytes present in the buffer.
+        let end_cursor = self.read_bytes()?;
+
+        let mut line_start_index = 0;
+        loop {
+            match self.state {
+                ConnectionState::WaitingForRequestLine => {
+                    if !self.parse_request_line(&mut line_start_index, end_cursor)? {
+                        return Ok(());
+                    }
+                }
+                ConnectionState::WaitingForHeaders => {
+                    if !self.parse_headers(&mut line_start_index, end_cursor)? {
+                        return Ok(());
+                    }
+                }
+                ConnectionState::WaitingForBody => {
+                    if !self.parse_body(&mut line_start_index, end_cursor)? {
+                        return Ok(());
+                    }
+                }
+                ConnectionState::RequestReady => {
+                    // This request is ready to be passed for handling.
+                    // Update the state machine to expect a new request and push this request into
+                    // the `parsed_requests` queue.
+                    self.state = ConnectionState::WaitingForRequestLine;
+                    self.body_bytes_to_be_read = 0;
+                    self.parsed_requests
+                        .push_back(self.pending_request.take().unwrap());
+                }
+            };
+        }
+    }
+
+    // Reads a maximum of 1024 bytes from the stream into `buffer`.
+    // The return value represents the end index of what we have just appended.
+    fn read_bytes(&mut self) -> Result<usize, ConnectionError> {
+        // Append new bytes to what we already have in the buffer.
+        let bytes_read = self
+            .stream
+            .read(&mut self.buffer[self.read_cursor..])
+            .map_err(ConnectionError::StreamError)?;
+
+        // If the read returned 0 then the client has closed the connection.
+        if bytes_read == 0 {
+            return Err(ConnectionError::ConnectionClosed);
+        }
+
+        Ok(bytes_read + self.read_cursor)
+    }
+
+    // Parses bytes in `buffer` for a valid request line.
+    // Returns `false` if there are no more bytes to be parsed in the buffer.
+    fn parse_request_line(
+        &mut self,
+        start: &mut usize,
+        end: usize,
+    ) -> Result<bool, ConnectionError> {
+        match find(&self.buffer[*start..end], &[CR, LF]) {
+            Some(line_end_index) => {
+                let line = &self.buffer[*start..(*start + line_end_index)];
+
+                *start = *start + line_end_index + CRLF_LEN;
+                let request_line =
+                    RequestLine::try_from(line).map_err(ConnectionError::ParseError)?;
+
+                // Form the request with a valid request line, which is the bare minimum
+                // for a valid request.
+                self.pending_request = Some(Request {
+                    request_line,
+                    headers: Headers::default(),
+                    body: None,
+                });
+                self.state = ConnectionState::WaitingForHeaders;
+                Ok(true)
+            }
+            None => {
+                // The request line is longer than BUFFER_SIZE bytes, so the request is invalid.
+                if end == BUFFER_SIZE && *start == 0 {
+                    return Err(ConnectionError::ParseError(RequestError::InvalidRequest));
+                } else {
+                    // Move the incomplete request line to the beginning of the buffer and wait
+                    // for the next `try_read` call to complete it.
+                    // This can only happen if another request was sent before this one, as the
+                    // limit for the length of a request line in this implementation is 1024 bytes.
+                    self.shift_buffer_left(*start, end);
+                }
+                Ok(false)
+            }
+        }
+    }
+
+    // Parses bytes in `buffer` for header fields.
+    // Returns `false` if there are no more bytes to be parsed in the buffer.
+    fn parse_headers(
+        &mut self,
+        line_start_index: &mut usize,
+        end_cursor: usize,
+    ) -> Result<bool, ConnectionError> {
+        match find(&self.buffer[*line_start_index..end_cursor], &[CR, LF]) {
+            // We have found the end of the headers.
+            // `line_start_index` points to the end of the most recently found CR LF
+            // sequence. That means that if we found the next CR LF sequence at this index,
+            // they are, in fact, a CR LF CR LF sequence, which marks the end of the header
+            // fields, per HTTP specification.
+            Some(0) => {
+                // If our current state is `WaitingForHeaders`, it means that we already have
+                // a valid request formed from a request line, so it's safe to unwrap.
+                let request = self.pending_request.as_mut().unwrap();
+                if request.headers.content_length() == 0 {
+                    self.state = ConnectionState::RequestReady;
+                } else {
+                    if request.headers.expect() {
+                        // Send expect.
+                        let expect_response =
+                            Response::new(request.http_version(), StatusCode::Continue);
+                        self.response_queue.push_back(expect_response);
+                    }
+
+                    self.body_bytes_to_be_read = request.headers.content_length();
+                    request.body = Some(Body::new(vec![]));
+                    self.state = ConnectionState::WaitingForBody;
+                }
+
+                // Update the index for the next header.
+                *line_start_index += CRLF_LEN;
+                Ok(true)
+            }
+            // We have found the end of a header line.
+            Some(relative_line_end_index) => {
+                let request = self.pending_request.as_mut().unwrap();
+                // The `line_end_index` relative to the whole buffer.
+                let line_end_index = relative_line_end_index + *line_start_index;
+
+                // Get the line slice and parse it.
+                let line = &self.buffer[*line_start_index..line_end_index];
+                match request.headers.parse_header_line(line) {
+                    // If a header is unsupported we ignore it.
+                    Ok(_) | Err(RequestError::UnsupportedHeader) => {}
+                    // If parsing the header invalidates the request, we propagate
+                    // the error.
+                    Err(e) => return Err(ConnectionError::ParseError(e)),
+                };
+
+                // Update the `line_start_index` to where we finished parsing.
+                *line_start_index = line_end_index + CRLF_LEN;
+                Ok(true)
+            }
+            // If we have an incomplete header line.
+            None => {
+                // If we have parsed BUFFER_SIZE bytes and still haven't found the header
+                // line end sequence.
+                if *line_start_index == 0 && end_cursor == BUFFER_SIZE {
+                    // Header line is longer than BUFFER_SIZE bytes, so it is invalid.
+                    return Err(ConnectionError::ParseError(RequestError::InvalidHeader));
+                }
+                // Move the incomplete header line from the end of the buffer to
+                // the beginning, so that we can append the rest of the line and
+                // parse it in the next `try_read` call.
+                self.shift_buffer_left(*line_start_index, end_cursor);
+                Ok(false)
+            }
+        }
+    }
+
+    // Parses bytes in `buffer` to be put into the request body, if there should be one.
+    // Returns `false` if there are no more bytes to be parsed in the buffer.
+    fn parse_body(
+        &mut self,
+        line_start_index: &mut usize,
+        end_cursor: usize,
+    ) -> Result<bool, ConnectionError> {
+        // If what we have just read is not enough to complete the request and
+        // there are more bytes pertaining to the body of the request.
+        if self.body_bytes_to_be_read > end_cursor as i32 - *line_start_index as i32 {
+            // Append everything that we read to our current incomplete body and update
+            // `body_bytes_to_be_read`.
+            self.body_vec
+                .extend_from_slice(&self.buffer[*line_start_index..end_cursor]);
+            self.body_bytes_to_be_read -= end_cursor as i32 - *line_start_index as i32;
+
+            // Clear the buffer and reset the starting index.
+            for i in 0..BUFFER_SIZE {
+                self.buffer[i] = 0;
+            }
+            self.read_cursor = 0;
+
+            return Ok(false);
+        }
+
+        // Append only the remaining necessary bytes to the body of the request.
+        self.body_vec.extend_from_slice(
+            &self.buffer
+                [*line_start_index..(*line_start_index + self.body_bytes_to_be_read as usize)],
+        );
+        *line_start_index += self.body_bytes_to_be_read as usize;
+        self.body_bytes_to_be_read = 0;
+
+        let request = self.pending_request.as_mut().unwrap();
+        // If there are no more bytes to be read for this request.
+        // Assign the body of the request.
+        let placeholder: Vec<_> = self
+            .body_vec
+            .drain(..request.headers.content_length() as usize)
+            .collect();
+        request.body = Some(Body::new(placeholder));
+
+        // If we read more bytes than we should have into the body of the request.
+        if !self.body_vec.is_empty() {
+            return Err(ConnectionError::ParseError(RequestError::InvalidRequest));
+        }
+
+        self.state = ConnectionState::RequestReady;
+        Ok(true)
+    }
+
+    /// Tries to write the first available response to the provided stream.
+    /// Meant to be used only with non-blocking streams and an `EPOLL` structure.
+    /// Should be called whenever an `EPOLLOUT` event is signaled.
+    ///
+    /// # Errors
+    /// `StreamError` is returned when an IO operation fails.
+    /// `ConnectionClosed` is returned when trying to write on a closed connection.
+    /// `InvalidWrite` is returned when trying to write on a connection with an
+    /// empty outgoing buffer.
+    pub fn try_write(&mut self) -> Result<(), ConnectionError> {
+        if self.response_buffer.is_none() {
+            if let Some(response) = self.response_queue.pop_front() {
+                let mut response_buffer_vec: Vec<u8> = Vec::new();
+                response
+                    .write_all(&mut response_buffer_vec)
+                    .map_err(ConnectionError::StreamError)?;
+                self.response_buffer = Some(response_buffer_vec);
+            } else {
+                return Err(ConnectionError::InvalidWrite);
+            }
+        }
+
+        let mut response_fully_written = false;
+        let mut connection_closed = false;
+
+        if let Some(response_buffer_vec) = self.response_buffer.as_mut() {
+            let bytes_to_be_written = response_buffer_vec.len();
+            match self.stream.write(response_buffer_vec.as_slice()) {
+                Ok(0) | Err(_) => {
+                    connection_closed = true;
+                }
+                Ok(bytes_written) => {
+                    if bytes_written != bytes_to_be_written {
+                        response_buffer_vec.drain(..bytes_written);
+                    } else {
+                        response_fully_written = true;
+                    }
+                }
+            }
+        }
+
+        if connection_closed {
+            self.clear_write_buffer();
+            return Err(ConnectionError::ConnectionClosed);
+        } else if response_fully_written {
+            self.response_buffer.take();
+        }
+
+        Ok(())
+    }
+
+    fn clear_write_buffer(&mut self) {
+        self.response_queue.clear();
+        self.response_buffer.take();
+    }
+
+    /// Send a response back to the source of a request.
+    pub fn enqueue_response(&mut self, response: Response) {
+        self.response_queue.push_back(response);
+    }
+
+    fn shift_buffer_left(&mut self, line_start_index: usize, end_cursor: usize) {
+        // We don't want to shift something that is already at the beginning.
+        if line_start_index != 0 {
+            // Move the bytes from `line_start_index` to the beginning of the buffer.
+            for cursor in 0..(end_cursor - line_start_index) {
+                self.buffer[cursor] = self.buffer[line_start_index + cursor];
+            }
+
+            // Clear the rest of the buffer.
+            for cursor in (end_cursor - line_start_index)..end_cursor {
+                self.buffer[cursor] = 0;
+            }
+        }
+
+        // Update `read_cursor`.
+        self.read_cursor = end_cursor - line_start_index;
+    }
+
+    /// Returns the first parsed request in the queue or `None` if the queue
+    /// is empty.
+    pub fn pop_parsed_request(&mut self) -> Option<Request> {
+        self.parsed_requests.pop_front()
+    }
+
+    /// Returns `true` if there are bytes waiting to be written into the stream.
+    pub fn pending_write(&self) -> bool {
+        self.response_buffer.is_some() || !self.response_queue.is_empty()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use common::{Method, Version};
+    use std::os::unix::net::UnixStream;
+
+    #[test]
+    fn test_try_read_expect() {
+        // Test request with `Expect` header.
+        let (mut sender, receiver) = UnixStream::pair().unwrap();
+        receiver.set_nonblocking(true).expect("Can't modify socket");
+        let mut conn = HttpConnection::new(receiver);
+        sender
+            .write_all(
+                b"PATCH http://localhost/home HTTP/1.1\r\n\
+                                 Expect: 100-continue\r\n\
+                                 Content-Length: 26\r\n\
+                                 Transfer-Encoding: chunked\r\n\r\n",
+            )
+            .unwrap();
+        assert!(conn.try_read().is_ok());
+
+        sender.write_all(b"this is not\n\r\na json \nbody").unwrap();
+        conn.try_read().unwrap();
+        let request = conn.pop_parsed_request().unwrap();
+
+        let expected_request = Request {
+            request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11),
+            headers: Headers::new(26, true, true),
+            body: Some(Body::new(b"this is not\n\r\na json \nbody".to_vec())),
+        };
+
+        assert_eq!(request, expected_request);
+    }
+
+    #[test]
+    fn test_try_read_long_headers() {
+        // Long request headers.
+        let (mut sender, receiver) = UnixStream::pair().unwrap();
+        receiver.set_nonblocking(true).expect("Can't modify socket");
+        let mut conn = HttpConnection::new(receiver);
+        sender
+            .write_all(
+                b"PATCH http://localhost/home HTTP/1.1\r\n\
+                                 Expect: 100-continue\r\n\
+                                 Transfer-Encoding: chunked\r\n",
+            )
+            .unwrap();
+
+        for i in 0..90 {
+            sender.write_all(b"Custom-Header-Testing: 1").unwrap();
+            sender.write_all(i.to_string().as_bytes()).unwrap();
+            sender.write_all(b"\r\n").unwrap();
+        }
+        sender
+            .write_all(b"Content-Length: 26\r\n\r\nthis is not\n\r\na json \nbody")
+            .unwrap();
+        assert!(conn.try_read().is_ok());
+        assert!(conn.try_read().is_ok());
+        assert!(conn.try_read().is_ok());
+        let request = conn.pop_parsed_request().unwrap();
+
+        let expected_request = Request {
+            request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11),
+            headers: Headers::new(26, true, true),
+            body: Some(Body::new(b"this is not\n\r\na json \nbody".to_vec())),
+        };
+        assert_eq!(request, expected_request);
+    }
+
+    #[test]
+    fn test_try_read_split_ending() {
+        // Long request with '\r\n' on BUFFER_SIZEth and 1025th positions in the request.
+        let (mut sender, receiver) = UnixStream::pair().unwrap();
+        receiver.set_nonblocking(true).expect("Can't modify socket");
+        let mut conn = HttpConnection::new(receiver);
+        sender
+            .write_all(
+                b"PATCH http://localhost/home HTTP/1.1\r\n\
+                                 Expect: 100-continue\r\n\
+                                 Transfer-Encoding: chunked\r\n",
+            )
+            .unwrap();
+
+        for i in 0..32 {
+            sender.write_all(b"Custom-Header-Testing: 1").unwrap();
+            sender.write_all(i.to_string().as_bytes()).unwrap();
+            sender.write_all(b"\r\n").unwrap();
+        }
+        sender
+            .write_all(b"Head: aaaaa\r\nContent-Length: 26\r\n\r\nthis is not\n\r\na json \nbody")
+            .unwrap();
+        assert!(conn.try_read().is_ok());
+        conn.try_read().unwrap();
+        let request = conn.pop_parsed_request().unwrap();
+        let expected_request = Request {
+            request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11),
+            headers: Headers::new(26, true, true),
+            body: Some(Body::new(b"this is not\n\r\na json \nbody".to_vec())),
+        };
+        assert_eq!(request, expected_request);
+    }
+
+    #[test]
+    fn test_try_read_invalid_request() {
+        // Invalid request.
+        let (mut sender, receiver) = UnixStream::pair().unwrap();
+        receiver.set_nonblocking(true).expect("Can't modify socket");
+        let mut conn = HttpConnection::new(receiver);
+        sender
+            .write_all(
+                b"PATCH http://localhost/home HTTP/1.1\r\n\
+                                 Expect: 100-continue\r\n\
+                                 Transfer-Encoding: chunked\r\n",
+            )
+            .unwrap();
+
+        for i in 0..40 {
+            sender.write_all(b"Custom-Header-Testing: 1").unwrap();
+            sender.write_all(i.to_string().as_bytes()).unwrap();
+            sender.write_all(b"\r\n").unwrap();
+        }
+        sender
+            .write_all(b"Content-Length: alpha\r\n\r\nthis is not\n\r\na json \nbody")
+            .unwrap();
+        assert!(conn.try_read().is_ok());
+        let request_error = conn.try_read().unwrap_err();
+        assert_eq!(
+            request_error,
+            ConnectionError::ParseError(RequestError::InvalidHeader)
+        );
+    }
+
+    #[test]
+    fn test_try_read_long_request_body() {
+        // Long request body.
+        let (mut sender, receiver) = UnixStream::pair().unwrap();
+        receiver.set_nonblocking(true).expect("Can't modify socket");
+        let mut conn = HttpConnection::new(receiver);
+        sender
+            .write_all(
+                b"PATCH http://localhost/home HTTP/1.1\r\n\
+                                 Expect: 100-continue\r\n\
+                                 Transfer-Encoding: chunked\r\n\
+                                 Content-Length: 1400\r\n\r\n",
+            )
+            .unwrap();
+
+        let mut request_body: Vec<u8> = Vec::with_capacity(1400);
+        for _ in 0..100 {
+            request_body.write_all(b"This is a test").unwrap();
+        }
+        sender.write_all(request_body.as_slice()).unwrap();
+        assert!(conn.try_read().is_ok());
+        conn.try_read().unwrap();
+        let request = conn.pop_parsed_request().unwrap();
+        let expected_request = Request {
+            request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11),
+            headers: Headers::new(1400, true, true),
+            body: Some(Body::new(request_body)),
+        };
+        assert_eq!(request, expected_request);
+    }
+
+    #[test]
+    fn test_try_read_large_req_line() {
+        // Request line longer than BUFFER_SIZE bytes.
+        let (mut sender, receiver) = UnixStream::pair().unwrap();
+        receiver.set_nonblocking(true).expect("Can't modify socket");
+        let mut conn = HttpConnection::new(receiver);
+        sender.write_all(b"PATCH http://localhost/home").unwrap();
+
+        let mut request_body: Vec<u8> = Vec::with_capacity(1400);
+        for _ in 0..200 {
+            request_body.write_all(b"/home").unwrap();
+        }
+        sender.write_all(request_body.as_slice()).unwrap();
+        assert_eq!(
+            conn.try_read().unwrap_err(),
+            ConnectionError::ParseError(RequestError::InvalidRequest)
+        );
+    }
+
+    #[test]
+    fn test_try_read_large_header_line() {
+        // Header line longer than BUFFER_SIZE bytes.
+        let (mut sender, receiver) = UnixStream::pair().unwrap();
+        receiver.set_nonblocking(true).expect("Can't modify socket");
+        let mut conn = HttpConnection::new(receiver);
+        sender
+            .write_all(b"PATCH http://localhost/home HTTP/1.1\r\nhead: ")
+            .unwrap();
+
+        let mut request_body: Vec<u8> = Vec::with_capacity(1030);
+        for _ in 0..86 {
+            request_body.write_all(b"abcdefghijkl").unwrap();
+        }
+        request_body.write_all(b"\r\n\r\n").unwrap();
+        sender.write_all(request_body.as_slice()).unwrap();
+        assert!(conn.try_read().is_ok());
+        assert_eq!(
+            conn.try_read().unwrap_err(),
+            ConnectionError::ParseError(RequestError::InvalidHeader)
+        );
+    }
+
+    #[test]
+    fn test_try_read_no_body_request() {
+        // Request without body.
+        let (mut sender, receiver) = UnixStream::pair().unwrap();
+        receiver.set_nonblocking(true).expect("Can't modify socket");
+        let mut conn = HttpConnection::new(receiver);
+        sender
+            .write_all(
+                b"PATCH http://localhost/home HTTP/1.1\r\n\
+                                 Expect: 100-continue\r\n\
+                                 Transfer-Encoding: chunked\r\n\r\n",
+            )
+            .unwrap();
+        conn.try_read().unwrap();
+        let request = conn.pop_parsed_request().unwrap();
+        let expected_request = Request {
+            request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11),
+            headers: Headers::new(0, true, true),
+            body: None,
+        };
+        assert_eq!(request, expected_request);
+    }
+
+    #[test]
+    fn test_try_read_segmented_req_line() {
+        // Segmented request line.
+        let (mut sender, receiver) = UnixStream::pair().unwrap();
+        receiver.set_nonblocking(true).expect("Can't modify socket");
+        let mut conn = HttpConnection::new(receiver);
+        sender.write_all(b"PATCH http://local").unwrap();
+        assert!(conn.try_read().is_ok());
+
+        sender.write_all(b"host/home HTTP/1.1\r\n\r\n").unwrap();
+
+        conn.try_read().unwrap();
+        let request = conn.pop_parsed_request().unwrap();
+        let expected_request = Request {
+            request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11),
+            headers: Headers::new(0, false, false),
+            body: None,
+        };
+        assert_eq!(request, expected_request);
+    }
+
+    #[test]
+    fn test_try_read_long_req_line_b2b() {
+        // Long request line after another request.
+        let (mut sender, receiver) = UnixStream::pair().unwrap();
+        receiver.set_nonblocking(true).expect("Can't modify socket");
+        let mut conn = HttpConnection::new(receiver);
+        // Req line 23 + 10*x + 13 = 36 + 10* x    984 free in first try read
+        sender
+            .write_all(b"PATCH http://localhost/home HTTP/1.1\r\n\r\nPATCH http://localhost/")
+            .unwrap();
+
+        let mut request_line: Vec<u8> = Vec::with_capacity(980);
+        for _ in 0..98 {
+            request_line.write_all(b"localhost/").unwrap();
+        }
+        request_line.write_all(b" HTTP/1.1\r\n\r\n").unwrap();
+        sender.write_all(request_line.as_slice()).unwrap();
+
+        conn.try_read().unwrap();
+        let request = conn.pop_parsed_request().unwrap();
+        let expected_request = Request {
+            request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11),
+            headers: Headers::new(0, false, false),
+            body: None,
+        };
+        assert_eq!(request, expected_request);
+
+        conn.try_read().unwrap();
+        let request = conn.pop_parsed_request().unwrap();
+        let mut expected_request_as_bytes = Vec::new();
+        expected_request_as_bytes
+            .write_all(b"http://localhost/")
+            .unwrap();
+        expected_request_as_bytes.append(request_line.as_mut());
+        let expected_request = Request {
+            request_line: RequestLine::new(
+                Method::Patch,
+                std::str::from_utf8(&expected_request_as_bytes[..997]).unwrap(),
+                Version::Http11,
+            ),
+            headers: Headers::new(0, false, false),
+            body: None,
+        };
+        assert_eq!(request, expected_request);
+    }
+
+    #[test]
+    fn test_try_read_double_request() {
+        // Double request in a single read.
+        let (mut sender, receiver) = UnixStream::pair().unwrap();
+        receiver.set_nonblocking(true).expect("Can't modify socket");
+        let mut conn = HttpConnection::new(receiver);
+        sender
+            .write_all(
+                b"PATCH http://localhost/home HTTP/1.1\r\n\
+                                 Transfer-Encoding: chunked\r\n\
+                                 Content-Length: 26\r\n\r\nthis is not\n\r\na json \nbody",
+            )
+            .unwrap();
+        sender
+            .write_all(
+                b"PUT http://farhost/away HTTP/1.1\r\nContent-Length: 23\r\n\r\nthis is another request",
+            )
+            .unwrap();
+
+        let expected_request_first = Request {
+            request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11),
+            headers: Headers::new(26, false, true),
+            body: Some(Body::new(b"this is not\n\r\na json \nbody".to_vec())),
+        };
+
+        conn.try_read().unwrap();
+        let request_first = conn.pop_parsed_request().unwrap();
+        let request_second = conn.pop_parsed_request().unwrap();
+
+        let expected_request_second = Request {
+            request_line: RequestLine::new(Method::Put, "http://farhost/away", Version::Http11),
+            headers: Headers::new(23, false, false),
+            body: Some(Body::new(b"this is another request".to_vec())),
+        };
+        assert_eq!(request_first, expected_request_first);
+        assert_eq!(request_second, expected_request_second);
+    }
+
+    #[test]
+    fn test_try_read_connection_closed() {
+        // Connection abruptly closed.
+        let (mut sender, receiver) = UnixStream::pair().unwrap();
+        receiver.set_nonblocking(true).expect("Can't modify socket");
+        let mut conn = HttpConnection::new(receiver);
+        sender
+            .write_all(
+                b"PATCH http://localhost/home HTTP/1.1\r\n\
+                                 Transfer-Encoding: chunked\r\n\
+                                 Content-Len",
+            )
+            .unwrap();
+
+        conn.try_read().unwrap();
+        sender.shutdown(std::net::Shutdown::Both).unwrap();
+
+        assert_eq!(
+            conn.try_read().unwrap_err(),
+            ConnectionError::ConnectionClosed
+        );
+    }
+
+    #[test]
+    fn test_enqueue_response() {
+        // Response without body.
+        let (sender, mut receiver) = UnixStream::pair().unwrap();
+        receiver.set_nonblocking(true).expect("Can't modify socket");
+        let mut conn = HttpConnection::new(sender);
+
+        let response = Response::new(Version::Http11, StatusCode::OK);
+        let mut expected_response: Vec<u8> = vec![];
+        response.write_all(&mut expected_response).unwrap();
+
+        conn.enqueue_response(response);
+        assert!(conn.try_write().is_ok());
+
+        let mut response_buffer = vec![0u8; expected_response.len()];
+        receiver.read_exact(&mut response_buffer).unwrap();
+        assert_eq!(response_buffer, expected_response);
+
+        // Response with body.
+        let (sender, mut receiver) = UnixStream::pair().unwrap();
+        receiver.set_nonblocking(true).expect("Can't modify socket");
+        let mut conn = HttpConnection::new(sender);
+        let mut response = Response::new(Version::Http11, StatusCode::OK);
+        let mut body: Vec<u8> = vec![];
+        body.write_all(br#"{ "json": "body", "hello": "world" }"#)
+            .unwrap();
+        response.set_body(Body::new(body));
+        let mut expected_response: Vec<u8> = vec![];
+        response.write_all(&mut expected_response).unwrap();
+
+        conn.enqueue_response(response);
+        assert!(conn.try_write().is_ok());
+
+        let mut response_buffer = vec![0u8; expected_response.len()];
+        receiver.read_exact(&mut response_buffer).unwrap();
+        assert_eq!(response_buffer, expected_response);
+    }
+}

src/lib.rs

@@ -0,0 +1,126 @@
+// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+#![deny(missing_docs)]
+//! Minimal implementation of the [HTTP/1.0](https://tools.ietf.org/html/rfc1945)
+//! and [HTTP/1.1](https://www.ietf.org/rfc/rfc2616.txt) protocols.
+//!
+//! HTTP/1.1 has a mandatory header **Host**, but as this crate is only used
+//! for parsing API requests, this header (if present) is ignored.
+//!
+//! This HTTP implementation is stateless thus it does not support chunking or
+//! compression.
+//!
+//! ## Supported Headers
+//! The **micro_http** crate has support for parsing the following **Request**
+//! headers:
+//! - Content-Length
+//! - Expect
+//! - Transfer-Encoding
+//!
+//! The **Response** does not have a public interface for adding headers, but whenever
+//! a write to the **Body** is made, the headers **ContentLength** and **MediaType**
+//! are automatically updated.
+//!
+//! ### Media Types
+//! The supported media types are:
+//! - text/plain
+//! - application/json
+//!
+//! ## Supported Methods
+//! The supported HTTP Methods are:
+//! - GET
+//! - PUT
+//! - PATCH
+//!
+//! ## Supported Status Codes
+//! The supported status codes are:
+//!
+//! - Continue - 100
+//! - OK - 200
+//! - No Content - 204
+//! - Bad Request - 400
+//! - Not Found - 404
+//! - Internal Server Error - 500
+//! - Not Implemented - 501
+//!
+//! ## Example for parsing an HTTP Request from a slice
+//! ```
+//! extern crate micro_http;
+//! use micro_http::{Request, Version};
+//!
+//! let http_request = Request::try_from(b"GET http://localhost/home HTTP/1.0\r\n\r\n").unwrap();
+//! assert_eq!(http_request.http_version(), Version::Http10);
+//! assert_eq!(http_request.uri().get_abs_path(), "/home");
+//! ```
+//!
+//! ## Example for creating an HTTP Response
+//! ```
+//! extern crate micro_http;
+//! use micro_http::{Body, MediaType, Response, StatusCode, Version};
+//!
+//! let mut response = Response::new(Version::Http10, StatusCode::OK);
+//! let body = String::from("This is a test");
+//! response.set_body(Body::new(body.clone()));
+//! response.set_content_type(MediaType::PlainText);
+//!
+//! assert!(response.status() == StatusCode::OK);
+//! assert_eq!(response.body().unwrap(), Body::new(body));
+//! assert_eq!(response.http_version(), Version::Http10);
+//!
+//! let mut response_buf: [u8; 126] = [0; 126];
+//! assert!(response.write_all(&mut response_buf.as_mut()).is_ok());
+//! ```
+//!
+//! `HttpConnection` can be used for automatic data exchange and parsing when
+//! handling a client, but it only supports one stream.
+//!
+//! For handling multiple clients use `HttpServer`, which multiplexes `HttpConnection`s
+//! and offers an easy to use interface. The server can run in either blocking or
+//! non-blocking mode. Non-blocking is achieved by using `epoll` to make sure
+//! `requests` will never block when called.
+//!
+//! ## Example for using the server
+//!
+//! ```
+//! extern crate micro_http;
+//! use micro_http::{HttpServer, Response, StatusCode};
+//!
+//! let path_to_socket = "/tmp/example.sock";
+//! std::fs::remove_file(path_to_socket).unwrap_or_default();
+//!
+//! // Start the server.
+//! let mut server = HttpServer::new(path_to_socket).unwrap();
+//! server.start_server().unwrap();
+//!
+//! // Connect a client to the server so it doesn't block in our example.
+//! let mut socket = std::os::unix::net::UnixStream::connect(path_to_socket).unwrap();
+//!
+//! // Server loop processing requests.
+//! loop {
+//!     for request in server.requests().unwrap() {
+//!         let response = request.process(|request| {
+//!             // Your code here.
+//!             Response::new(request.http_version(), StatusCode::NoContent)
+//!         });
+//!         server.respond(response);
+//!     }
+//!     // Break this example loop.
+//!     break;
+//! }
+//! ```
+
+mod common;
+mod connection;
+mod request;
+mod response;
+mod server;
+use common::ascii;
+use common::headers;
+
+pub use connection::{ConnectionError, HttpConnection};
+pub use request::{Request, RequestError};
+pub use response::{Response, StatusCode};
+pub use server::{HttpServer, ServerError, ServerRequest, ServerResponse};
+
+pub use common::headers::{Headers, MediaType};
+pub use common::{Body, Method, Version};

src/request.rs

@@ -0,0 +1,470 @@
+// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::str::from_utf8;
+
+use common::ascii::{CR, CRLF_LEN, LF, SP};
+pub use common::RequestError;
+use common::{Body, Method, Version};
+use headers::Headers;
+
+/// Finds the first occurence of `sequence` in the `bytes` slice.
+///
+/// Returns the starting position of the `sequence` in `bytes` or `None` if the
+/// `sequence` is not found.
+pub fn find(bytes: &[u8], sequence: &[u8]) -> Option<usize> {
+    bytes
+        .windows(sequence.len())
+        .position(|window| window == sequence)
+}
+
+/// Wrapper over HTTP URIs.
+///
+/// The `Uri` can not be used directly and it is only accessible from an HTTP Request.
+#[derive(Clone, Debug, PartialEq)]
+pub struct Uri {
+    string: String,
+}
+
+impl Uri {
+    fn new(slice: &str) -> Self {
+        Uri {
+            string: String::from(slice),
+        }
+    }
+
+    fn try_from(bytes: &[u8]) -> Result<Self, RequestError> {
+        if bytes.is_empty() {
+            return Err(RequestError::InvalidUri("Empty URI not allowed."));
+        }
+        let utf8_slice =
+            from_utf8(bytes).map_err(|_| RequestError::InvalidUri("Cannot parse URI as UTF-8."))?;
+        Ok(Uri::new(utf8_slice))
+    }
+
+    /// Returns the absolute path of the `Uri`.
+    ///
+    /// URIs can be represented in absolute form or relative form. The absolute form includes
+    /// the HTTP scheme, followed by the absolute path as follows:
+    /// "http:" "//" host [ ":" port ] [ abs_path ]
+    /// The relative URIs can be one of net_path | abs_path | rel_path.
+    /// This method only handles absolute URIs and relative URIs specified by abs_path.
+    /// The abs_path is expected to start with '/'.
+    ///
+    /// # Errors
+    /// Returns an empty byte array when the host or the path are empty/invalid.
+    pub fn get_abs_path(&self) -> &str {
+        const HTTP_SCHEME_PREFIX: &str = "http://";
+
+        if self.string.starts_with(HTTP_SCHEME_PREFIX) {
+            let without_scheme = &self.string[HTTP_SCHEME_PREFIX.len()..];
+            if without_scheme.is_empty() {
+                return "";
+            }
+            // The host in this case includes the port and contains the bytes after http:// up to
+            // the next '/'.
+            match without_scheme.bytes().position(|byte| byte == b'/') {
+                Some(len) => &without_scheme[len..],
+                None => "",
+            }
+        } else {
+            if self.string.starts_with('/') {
+                return self.string.as_str();
+            }
+
+            ""
+        }
+    }
+}
+
+/// Wrapper over an HTTP Request Line.
+#[derive(Debug, PartialEq)]
+pub struct RequestLine {
+    method: Method,
+    uri: Uri,
+    http_version: Version,
+}
+
+impl RequestLine {
+    fn parse_request_line(request_line: &[u8]) -> (&[u8], &[u8], &[u8]) {
+        if let Some(method_end) = find(request_line, &[SP]) {
+            let method = &request_line[..method_end];
+
+            let uri_and_version = &request_line[(method_end + 1)..];
+
+            if let Some(uri_end) = find(uri_and_version, &[SP]) {
+                let uri = &uri_and_version[..uri_end];
+
+                let version = &uri_and_version[(uri_end + 1)..];
+
+                return (method, uri, version);
+            }
+
+            return (method, uri_and_version, b"");
+        }
+
+        (b"", b"", b"")
+    }
+
+    /// Tries to parse a byte stream in a request line. Fails if the request line is malformed.
+    pub fn try_from(request_line: &[u8]) -> Result<Self, RequestError> {
+        let (method, uri, version) = RequestLine::parse_request_line(request_line);
+
+        Ok(RequestLine {
+            method: Method::try_from(method)?,
+            uri: Uri::try_from(uri)?,
+            http_version: Version::try_from(version)?,
+        })
+    }
+
+    // Returns the minimum length of a valid request. The request must contain
+    // the method (GET), the URI (minmum 1 character), the HTTP version(HTTP/DIGIT.DIGIT) and
+    // 2 separators (SP).
+    fn min_len() -> usize {
+        Method::Get.raw().len() + 1 + Version::Http10.raw().len() + 2
+    }
+
+    #[cfg(test)]
+    pub fn new(method: Method, uri: &str, http_version: Version) -> Self {
+        RequestLine {
+            method,
+            uri: Uri::new(uri),
+            http_version,
+        }
+    }
+}
+
+/// Wrapper over an HTTP Request.
+#[allow(unused)]
+#[derive(Debug)]
+pub struct Request {
+    /// The request line of the request.
+    pub request_line: RequestLine,
+    /// The headers of the request.
+    pub headers: Headers,
+    /// The body of the request.
+    pub body: Option<Body>,
+}
+
+impl Request {
+    /// Parses a byte slice into a HTTP Request.
+    ///
+    /// The byte slice is expected to have the following format: </br>
+    ///     * Request Line: "GET SP Request-uri SP HTTP/1.0 CRLF" - Mandatory </br>
+    ///     * Request Headers "<headers> CRLF"- Optional </br>
+    ///     * Entity Body - Optional </br>
+    /// The request headers and the entity body is not parsed and None is returned because
+    /// these are not used by the MMDS server.
+    /// The only supported method is GET and the HTTP protocol is expected to be HTTP/1.0
+    /// or HTTP/1.1.
+    ///
+    /// # Errors
+    /// The function returns InvalidRequest when parsing the byte stream fails.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// extern crate micro_http;
+    /// use micro_http::Request;
+    ///
+    /// let http_request = Request::try_from(b"GET http://localhost/home HTTP/1.0\r\n");
+    /// ```
+    pub fn try_from(byte_stream: &[u8]) -> Result<Self, RequestError> {
+        // The first line of the request is the Request Line. The line ending is CR LF.
+        let request_line_end = match find(byte_stream, &[CR, LF]) {
+            Some(len) => len,
+            // If no CR LF is found in the stream, the request format is invalid.
+            None => return Err(RequestError::InvalidRequest),
+        };
+
+        let request_line_bytes = &byte_stream[..request_line_end];
+        if request_line_bytes.len() < RequestLine::min_len() {
+            return Err(RequestError::InvalidRequest);
+        }
+
+        let request_line = RequestLine::try_from(request_line_bytes)?;
+
+        // Find the next CR LF CR LF sequence in our buffer starting at the end on the Request
+        // Line, including the trailing CR LF previously found.
+        match find(&byte_stream[request_line_end..], &[CR, LF, CR, LF]) {
+            // If we have found a CR LF CR LF at the end of the Request Line, the request
+            // is complete.
+            Some(0) => Ok(Request {
+                request_line,
+                headers: Headers::default(),
+                body: None,
+            }),
+            Some(headers_end) => {
+                // Parse the request headers.
+                // Start by removing the leading CR LF from them.
+                let headers_and_body = &byte_stream[(request_line_end + CRLF_LEN)..];
+                let headers_end = headers_end - CRLF_LEN;
+                let headers = Headers::try_from(&headers_and_body[..headers_end])?;
+
+                // Parse the body of the request.
+                // Firstly check if we have a body.
+                let body = match headers.content_length() {
+                    0 => {
+                        // No request body.
+                        None
+                    }
+                    content_length => {
+                        // Headers suggest we have a body, but the buffer is shorter than the specified
+                        // content length.
+                        if headers_and_body.len() - (headers_end + 2 * CRLF_LEN)
+                            < content_length as usize
+                        {
+                            return Err(RequestError::InvalidRequest);
+                        }
+                        let body_as_bytes = &headers_and_body[(headers_end + 2 * CRLF_LEN)..];
+                        // If the actual length of the body is different than the `Content-Length` value
+                        // in the headers then this request is invalid.
+                        if body_as_bytes.len() == content_length as usize {
+                            Some(Body::new(body_as_bytes))
+                        } else {
+                            return Err(RequestError::InvalidRequest);
+                        }
+                    }
+                };
+
+                Ok(Request {
+                    request_line,
+                    headers,
+                    body,
+                })
+            }
+            // If we can't find a CR LF CR LF even though the request should have headers
+            // the request format is invalid.
+            None => Err(RequestError::InvalidRequest),
+        }
+    }
+
+    /// Returns the `Uri` from the parsed `Request`.
+    ///
+    /// The return value can be used to get the absolute path of the URI.
+    pub fn uri(&self) -> &Uri {
+        &self.request_line.uri
+    }
+
+    /// Returns the HTTP `Version` of the `Request`.
+    pub fn http_version(&self) -> Version {
+        self.request_line.http_version
+    }
+
+    /// Returns the HTTP `Method` of the `Request`.
+    pub fn method(&self) -> Method {
+        self.request_line.method
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    impl PartialEq for Request {
+        fn eq(&self, other: &Request) -> bool {
+            // Ignore the other fields of Request for now because they are not used.
+            self.request_line == other.request_line
+                && self.headers.content_length() == other.headers.content_length()
+                && self.headers.expect() == other.headers.expect()
+                && self.headers.chunked() == other.headers.chunked()
+        }
+    }
+
+    #[test]
+    fn test_uri() {
+        let uri = Uri::new("http://localhost/home");
+        assert_eq!(uri.get_abs_path(), "/home");
+
+        let uri = Uri::new("/home");
+        assert_eq!(uri.get_abs_path(), "/home");
+
+        let uri = Uri::new("home");
+        assert_eq!(uri.get_abs_path(), "");
+
+        let uri = Uri::new("http://");
+        assert_eq!(uri.get_abs_path(), "");
+
+        let uri = Uri::new("http://192.168.0.0");
+        assert_eq!(uri.get_abs_path(), "");
+    }
+
+    #[test]
+    fn test_find() {
+        let bytes: &[u8; 13] = b"abcacrgbabsjl";
+        let i = find(&bytes[..], b"ac");
+        assert_eq!(i.unwrap(), 3);
+
+        let i = find(&bytes[..], b"rgb");
+        assert_eq!(i.unwrap(), 5);
+
+        let i = find(&bytes[..], b"ab");
+        assert_eq!(i.unwrap(), 0);
+
+        let i = find(&bytes[..], b"l");
+        assert_eq!(i.unwrap(), 12);
+
+        let i = find(&bytes[..], b"jle");
+        assert!(i.is_none());
+
+        let i = find(&bytes[..], b"asdkjhasjhdjhgsadg");
+        assert!(i.is_none());
+
+        let i = find(&bytes[..], b"abcacrgbabsjl");
+        assert_eq!(i.unwrap(), 0);
+    }
+
+    #[test]
+    fn test_into_request_line() {
+        let expected_request_line = RequestLine {
+            http_version: Version::Http10,
+            method: Method::Get,
+            uri: Uri::new("http://localhost/home"),
+        };
+
+        let request_line = b"GET http://localhost/home HTTP/1.0";
+        assert_eq!(
+            RequestLine::try_from(request_line).unwrap(),
+            expected_request_line
+        );
+
+        let expected_request_line = RequestLine {
+            http_version: Version::Http11,
+            method: Method::Get,
+            uri: Uri::new("http://localhost/home"),
+        };
+
+        // Happy case with request line ending in CRLF.
+        let request_line = b"GET http://localhost/home HTTP/1.1";
+        assert_eq!(
+            RequestLine::try_from(request_line).unwrap(),
+            expected_request_line
+        );
+
+        // Happy case with request line ending in LF instead of CRLF.
+        let request_line = b"GET http://localhost/home HTTP/1.1";
+        assert_eq!(
+            RequestLine::try_from(request_line).unwrap(),
+            expected_request_line
+        );
+
+        // Test for invalid method.
+        let request_line = b"POST http://localhost/home HTTP/1.0";
+        assert_eq!(
+            RequestLine::try_from(request_line).unwrap_err(),
+            RequestError::InvalidHttpMethod("Unsupported HTTP method.")
+        );
+
+        // Test for invalid uri.
+        let request_line = b"GET  HTTP/1.0";
+        assert_eq!(
+            RequestLine::try_from(request_line).unwrap_err(),
+            RequestError::InvalidUri("Empty URI not allowed.")
+        );
+
+        // Test for invalid HTTP version.
+        let request_line = b"GET http://localhost/home HTTP/2.0";
+        assert_eq!(
+            RequestLine::try_from(request_line).unwrap_err(),
+            RequestError::InvalidHttpVersion("Unsupported HTTP version.")
+        );
+
+        // Test for invalid format with no method, uri or version.
+        let request_line = b"nothing";
+        assert_eq!(
+            RequestLine::try_from(request_line).unwrap_err(),
+            RequestError::InvalidHttpMethod("Unsupported HTTP method.")
+        );
+
+        // Test for invalid format with no version.
+        let request_line = b"GET /";
+        assert_eq!(
+            RequestLine::try_from(request_line).unwrap_err(),
+            RequestError::InvalidHttpVersion("Unsupported HTTP version.")
+        );
+    }
+
+    #[test]
+    fn test_into_request() {
+        let expected_request = Request {
+            request_line: RequestLine {
+                http_version: Version::Http10,
+                method: Method::Get,
+                uri: Uri::new("http://localhost/home"),
+            },
+            body: None,
+            headers: Headers::default(),
+        };
+        let request_bytes = b"GET http://localhost/home HTTP/1.0\r\n\
+                                     Last-Modified: Tue, 15 Nov 1994 12:45:26 GMT\r\n\r\n";
+        let request = Request::try_from(request_bytes).unwrap();
+        assert_eq!(request, expected_request);
+        assert_eq!(request.uri(), &Uri::new("http://localhost/home"));
+        assert_eq!(request.http_version(), Version::Http10);
+        assert!(request.body.is_none());
+
+        // Test for invalid Request (length is less than minimum).
+        let request_bytes = b"GET";
+        assert_eq!(
+            Request::try_from(request_bytes).unwrap_err(),
+            RequestError::InvalidRequest
+        );
+
+        // Test for a request with the headers we are looking for.
+        let request = Request::try_from(
+            b"PATCH http://localhost/home HTTP/1.1\r\n\
+                                     Expect: 100-continue\r\n\
+                                     Transfer-Encoding: chunked\r\n\
+                                     Content-Length: 26\r\n\r\nthis is not\n\r\na json \nbody",
+        )
+        .unwrap();
+        assert_eq!(request.uri(), &Uri::new("http://localhost/home"));
+        assert_eq!(request.http_version(), Version::Http11);
+        assert_eq!(request.method(), Method::Patch);
+        assert_eq!(request.headers.chunked(), true);
+        assert_eq!(request.headers.expect(), true);
+        assert_eq!(request.headers.content_length(), 26);
+        assert_eq!(
+            request.body.unwrap().body,
+            String::from("this is not\n\r\na json \nbody")
+                .as_bytes()
+                .to_vec()
+        );
+
+        // Test for an invalid request format.
+        Request::try_from(b"PATCH http://localhost/home HTTP/1.1\r\n").unwrap_err();
+
+        // Test for an invalid encoding.
+        let request = Request::try_from(
+            b"PATCH http://localhost/home HTTP/1.1\r\n\
+                                     Expect: 100-continue\r\n\
+                                     Transfer-Encoding: identity; q=0\r\n\
+                                     Content-Length: 26\r\n\r\nthis is not\n\r\na json \nbody",
+        )
+        .unwrap_err();
+        assert_eq!(request, RequestError::InvalidHeader);
+
+        // Test for an invalid content length.
+        let request = Request::try_from(
+            b"PATCH http://localhost/home HTTP/1.1\r\n\
+                                     Expect: 100-continue\r\n\
+                                     Content-Length: 5000\r\n\r\nthis is a short body",
+        )
+        .unwrap_err();
+        assert_eq!(request, RequestError::InvalidRequest);
+
+        // Test for a request without a body and an optional header.
+        let request = Request::try_from(
+            b"GET http://localhost/ HTTP/1.0\r\n\
+                                     Accept-Encoding: gzip\r\n\r\n",
+        )
+        .unwrap();
+        assert_eq!(request.uri(), &Uri::new("http://localhost/"));
+        assert_eq!(request.http_version(), Version::Http10);
+        assert_eq!(request.method(), Method::Get);
+        assert_eq!(request.headers.chunked(), false);
+        assert_eq!(request.headers.expect(), false);
+        assert_eq!(request.headers.content_length(), 0);
+        assert!(request.body.is_none());
+    }
+}

src/response.rs

@@ -0,0 +1,291 @@
+// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::io::{Error as WriteError, Write};
+
+use ascii::{COLON, CR, LF, SP};
+use common::{Body, Version};
+use headers::{Header, MediaType};
+
+/// Wrapper over a response status code.
+///
+/// The status code is defined as specified in the
+/// [RFC](https://tools.ietf.org/html/rfc7231#section-6).
+#[allow(dead_code)]
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub enum StatusCode {
+    /// 100, Continue
+    Continue,
+    /// 200, OK
+    OK,
+    /// 204, No Content
+    NoContent,
+    /// 400, Bad Request
+    BadRequest,
+    /// 404, Not Found
+    NotFound,
+    /// 500, Internal Server Error
+    InternalServerError,
+    /// 501, Not Implemented
+    NotImplemented,
+}
+
+impl StatusCode {
+    /// Returns the status code as bytes.
+    pub fn raw(self) -> &'static [u8; 3] {
+        match self {
+            StatusCode::Continue => b"100",
+            StatusCode::OK => b"200",
+            StatusCode::NoContent => b"204",
+            StatusCode::BadRequest => b"400",
+            StatusCode::NotFound => b"404",
+            StatusCode::InternalServerError => b"500",
+            StatusCode::NotImplemented => b"501",
+        }
+    }
+}
+
+struct StatusLine {
+    http_version: Version,
+    status_code: StatusCode,
+}
+
+impl StatusLine {
+    fn new(http_version: Version, status_code: StatusCode) -> Self {
+        StatusLine {
+            http_version,
+            status_code,
+        }
+    }
+
+    fn write_all<T: Write>(&self, mut buf: T) -> Result<(), WriteError> {
+        buf.write_all(self.http_version.raw())?;
+        buf.write_all(&[SP])?;
+        buf.write_all(self.status_code.raw())?;
+        buf.write_all(&[SP, CR, LF])?;
+
+        Ok(())
+    }
+}
+
+/// Wrapper over the list of headers associated with a HTTP Response.
+/// When creating a ResponseHeaders object, the content type is initialized to `text/plain`.
+/// The content type can be updated with a call to `set_content_type`.
+pub struct ResponseHeaders {
+    content_length: i32,
+    content_type: MediaType,
+    server: String,
+}
+
+impl Default for ResponseHeaders {
+    fn default() -> Self {
+        ResponseHeaders {
+            content_length: Default::default(),
+            content_type: Default::default(),
+            server: "Firecracker API".to_string(),
+        }
+    }
+}
+
+impl ResponseHeaders {
+    /// Writes the headers to `buf` using the HTTP specification.
+    pub fn write_all<T: Write>(&self, buf: &mut T) -> Result<(), WriteError> {
+        buf.write_all(Header::Server.raw())?;
+        buf.write_all(&[COLON, SP])?;
+        buf.write_all(self.server.as_bytes())?;
+
+        buf.write_all(&[CR, LF])?;
+        buf.write_all(b"Connection: keep-alive")?;
+        buf.write_all(&[CR, LF])?;
+
+        if self.content_length != 0 {
+            buf.write_all(Header::ContentType.raw())?;
+            buf.write_all(&[COLON, SP])?;
+            buf.write_all(self.content_type.as_str().as_bytes())?;
+            buf.write_all(&[CR, LF])?;
+
+            buf.write_all(Header::ContentLength.raw())?;
+            buf.write_all(&[COLON, SP])?;
+            buf.write_all(self.content_length.to_string().as_bytes())?;
+            buf.write_all(&[CR, LF])?;
+        }
+
+        buf.write_all(&[CR, LF])
+    }
+
+    // Sets the content length to be written in the HTTP response.
+    fn set_content_length(&mut self, content_length: i32) {
+        self.content_length = content_length;
+    }
+
+    /// Sets the HTTP response header server.
+    pub fn set_server(&mut self, server: &str) {
+        self.server = String::from(server);
+    }
+
+    /// Sets the content type to be written in the HTTP response.
+    #[allow(unused)]
+    pub fn set_content_type(&mut self, content_type: MediaType) {
+        self.content_type = content_type;
+    }
+}
+
+/// Wrapper over an HTTP Response.
+///
+/// The Response is created using a `Version` and a `StatusCode`. When creating a Response object,
+/// the body is initialized to `None`. The body can be updated with a call to `set_body`.
+pub struct Response {
+    status_line: StatusLine,
+    headers: ResponseHeaders,
+    body: Option<Body>,
+}
+
+impl Response {
+    /// Creates a new HTTP `Response` with an empty body.
+    pub fn new(http_version: Version, status_code: StatusCode) -> Response {
+        Response {
+            status_line: StatusLine::new(http_version, status_code),
+            headers: ResponseHeaders::default(),
+            body: None,
+        }
+    }
+
+    /// Updates the body of the `Response`.
+    ///
+    /// This function has side effects because it also updates the headers:
+    /// - `ContentLength`: this is set to the length of the specified body.
+    pub fn set_body(&mut self, body: Body) {
+        self.headers.set_content_length(body.len() as i32);
+        self.body = Some(body);
+    }
+
+    /// Updates the content type of the `Response`.
+    pub fn set_content_type(&mut self, content_type: MediaType) {
+        self.headers.set_content_type(content_type);
+    }
+
+    /// Sets the HTTP response server.
+    pub fn set_server(&mut self, server: &str) {
+        self.headers.set_server(server);
+    }
+
+    fn write_body<T: Write>(&self, mut buf: T) -> Result<(), WriteError> {
+        if let Some(ref body) = self.body {
+            buf.write_all(body.raw())?;
+        }
+        Ok(())
+    }
+
+    /// Writes the content of the `Response` to the specified `buf`.
+    ///
+    /// # Errors
+    /// Returns an error when the buffer is not large enough.
+    pub fn write_all<T: Write>(&self, mut buf: &mut T) -> Result<(), WriteError> {
+        self.status_line.write_all(&mut buf)?;
+        self.headers.write_all(&mut buf)?;
+        self.write_body(&mut buf)?;
+
+        Ok(())
+    }
+
+    /// Returns the Status Code of the Response.
+    pub fn status(&self) -> StatusCode {
+        self.status_line.status_code
+    }
+
+    /// Returns the Body of the response. If the response does not have a body,
+    /// it returns None.
+    pub fn body(&self) -> Option<Body> {
+        self.body.clone()
+    }
+
+    /// Returns the HTTP Version of the response.
+    pub fn content_length(&self) -> i32 {
+        self.headers.content_length
+    }
+
+    /// Returns the HTTP Version of the response.
+    pub fn content_type(&self) -> MediaType {
+        self.headers.content_type
+    }
+
+    /// Returns the HTTP Version of the response.
+    pub fn http_version(&self) -> Version {
+        self.status_line.http_version
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_write_response() {
+        let mut response = Response::new(Version::Http10, StatusCode::OK);
+        let body = "This is a test";
+        response.set_body(Body::new(body));
+        response.set_content_type(MediaType::PlainText);
+
+        assert!(response.status() == StatusCode::OK);
+        assert_eq!(response.body().unwrap(), Body::new(body));
+        assert_eq!(response.http_version(), Version::Http10);
+        assert_eq!(response.content_length(), 14);
+        assert_eq!(response.content_type(), MediaType::PlainText);
+
+        let expected_response: &'static [u8] = b"HTTP/1.0 200 \r\n\
+            Server: Firecracker API\r\n\
+            Connection: keep-alive\r\n\
+            Content-Type: text/plain\r\n\
+            Content-Length: 14\r\n\r\n\
+            This is a test";
+
+        let mut response_buf: [u8; 126] = [0; 126];
+        assert!(response.write_all(&mut response_buf.as_mut()).is_ok());
+        assert!(response_buf.as_ref() == expected_response);
+
+        // Test write failed.
+        let mut response_buf: [u8; 1] = [0; 1];
+        assert!(response.write_all(&mut response_buf.as_mut()).is_err());
+    }
+
+    #[test]
+    fn test_set_server() {
+        let mut response = Response::new(Version::Http10, StatusCode::OK);
+        let body = "This is a test";
+        let server = "rust-vmm API";
+        response.set_body(Body::new(body));
+        response.set_content_type(MediaType::PlainText);
+        response.set_server(server);
+
+        assert!(response.status() == StatusCode::OK);
+        assert_eq!(response.body().unwrap(), Body::new(body));
+        assert_eq!(response.http_version(), Version::Http10);
+        assert_eq!(response.content_length(), 14);
+        assert_eq!(response.content_type(), MediaType::PlainText);
+
+        let expected_response = format!(
+            "HTTP/1.0 200 \r\n\
+             Server: {}\r\n\
+             Connection: keep-alive\r\n\
+             Content-Type: text/plain\r\n\
+             Content-Length: 14\r\n\r\n\
+             This is a test",
+            server
+        );
+
+        let mut response_buf: [u8; 123] = [0; 123];
+        assert!(response.write_all(&mut response_buf.as_mut()).is_ok());
+        assert!(response_buf.as_ref() == expected_response.as_bytes());
+    }
+
+    #[test]
+    fn test_status_code() {
+        assert_eq!(StatusCode::Continue.raw(), b"100");
+        assert_eq!(StatusCode::OK.raw(), b"200");
+        assert_eq!(StatusCode::NoContent.raw(), b"204");
+        assert_eq!(StatusCode::BadRequest.raw(), b"400");
+        assert_eq!(StatusCode::NotFound.raw(), b"404");
+        assert_eq!(StatusCode::InternalServerError.raw(), b"500");
+        assert_eq!(StatusCode::NotImplemented.raw(), b"501");
+    }
+}

src/server.rs

@@ -0,0 +1,834 @@
+// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+extern crate epoll;
+
+use std::io::{Read, Write};
+use std::os::unix::io::AsRawFd;
+use std::os::unix::io::RawFd;
+use std::os::unix::net::{UnixListener, UnixStream};
+use std::path::Path;
+
+use common::{Body, Version};
+pub use common::{ConnectionError, RequestError, ServerError};
+use connection::HttpConnection;
+use request::Request;
+use response::{Response, StatusCode};
+use std::collections::HashMap;
+
+static SERVER_FULL_ERROR_MESSAGE: &[u8] = b"HTTP/1.1 503\r\n\
+                                            Server: Firecracker API\r\n\
+                                            Connection: close\r\n\
+                                            Content-Length: 40\r\n\r\n{ \"error\": \"Too many open connections\" }";
+const MAX_CONNECTIONS: usize = 10;
+
+type Result<T> = std::result::Result<T, ServerError>;
+
+/// Wrapper over `Request` which adds an identification token.
+pub struct ServerRequest {
+    /// Inner request.
+    pub request: Request,
+    /// Identification token.
+    id: u64,
+}
+
+impl ServerRequest {
+    /// Creates a new `ServerRequest` object from an existing `Request`,
+    /// adding an identification token.
+    pub fn new(request: Request, id: u64) -> Self {
+        ServerRequest { request, id }
+    }
+
+    /// Returns a reference to the inner request.
+    pub fn inner(&self) -> &Request {
+        &self.request
+    }
+
+    /// Calls the function provided on the inner request to obtain the response.
+    /// The response is then wrapped in a `ServerResponse`.
+    ///
+    /// Returns a `ServerResponse` ready for yielding to the server
+    pub fn process<F>(&self, callable: F) -> ServerResponse
+    where
+        F: Fn(&Request) -> Response,
+    {
+        let http_response = callable(self.inner());
+        ServerResponse::new(http_response, self.id)
+    }
+}
+
+/// Wrapper over `Response` which adds an identification token.
+pub struct ServerResponse {
+    /// Inner response.
+    response: Response,
+    /// Identification token.
+    id: u64,
+}
+
+impl ServerResponse {
+    fn new(response: Response, id: u64) -> ServerResponse {
+        ServerResponse { response, id }
+    }
+}
+
+/// Describes the state of the connection as far as data exchange
+/// on the stream is concerned.
+#[derive(PartialOrd, PartialEq)]
+enum ClientConnectionState {
+    AwaitingIncoming,
+    AwaitingOutgoing,
+    Closed,
+}
+
+/// Wrapper over `HttpConnection` which keeps track of yielded
+/// requests and absorbed responses.
+struct ClientConnection<T> {
+    /// The `HttpConnection` object which handles data exchange.
+    connection: HttpConnection<T>,
+    /// The state of the connection in the `epoll` structure.
+    state: ClientConnectionState,
+    /// Represents the difference between yielded requests and
+    /// absorbed responses.
+    /// This has to be `0` if we want to drop the connection.
+    in_flight_response_count: u32,
+}
+
+impl<T: Read + Write> ClientConnection<T> {
+    fn new(connection: HttpConnection<T>) -> Self {
+        ClientConnection {
+            connection,
+            state: ClientConnectionState::AwaitingIncoming,
+            in_flight_response_count: 0,
+        }
+    }
+
+    fn read(&mut self) -> Result<Vec<Request>> {
+        // Data came into the connection.
+        let mut parsed_requests = vec![];
+        match self.connection.try_read() {
+            Err(ConnectionError::ConnectionClosed) => {
+                // Connection timeout.
+                self.state = ClientConnectionState::Closed;
+                // We don't want to propagate this to the server and we will
+                // return no requests and wait for the connection to become
+                // safe to drop.
+                return Ok(vec![]);
+            }
+            Err(ConnectionError::StreamError(inner)) => {
+                // Reading from the connection failed.
+                // We should try to write an error message regardless.
+                let mut internal_error_response =
+                    Response::new(Version::Http11, StatusCode::InternalServerError);
+                internal_error_response.set_body(Body::new(inner.to_string()));
+                self.connection.enqueue_response(internal_error_response);
+            }
+            Err(ConnectionError::ParseError(inner)) => {
+                // An error occurred while parsing the read bytes.
+                // Check if there are any valid parsed requests in the queue.
+                while let Some(_discarded_request) = self.connection.pop_parsed_request() {}
+
+                // Send an error response for the request that gave us the error.
+                let mut error_response = Response::new(Version::Http11, StatusCode::BadRequest);
+                error_response.set_body(Body::new(
+                    format!(
+                        "{{ \"error\": \"{}\nAll previous unanswered requests will be dropped.\" }}",
+                        inner.to_string()
+                    )
+                    .to_string(),
+                ));
+                self.connection.enqueue_response(error_response);
+            }
+            Err(ConnectionError::InvalidWrite) => {
+                // This is unreachable because `HttpConnection::try_read()` cannot return this error variant.
+                unreachable!();
+            }
+            Ok(()) => {
+                while let Some(request) = self.connection.pop_parsed_request() {
+                    // Add all valid requests to `parsed_requests`.
+                    parsed_requests.push(request);
+                }
+            }
+        }
+        self.in_flight_response_count += parsed_requests.len() as u32;
+        // If the state of the connection has changed, we need to update
+        // the event set in the `epoll` structure.
+        if self.connection.pending_write() {
+            self.state = ClientConnectionState::AwaitingOutgoing;
+        }
+
+        Ok(parsed_requests)
+    }
+
+    fn write(&mut self) -> Result<()> {
+        // The stream is available for writing.
+        match self.connection.try_write() {
+            Err(ConnectionError::ConnectionClosed) | Err(ConnectionError::StreamError(_)) => {
+                // Writing to the stream failed so it will be removed.
+                self.state = ClientConnectionState::Closed;
+            }
+            Err(ConnectionError::InvalidWrite) => {
+                // A `try_write` call was performed on a connection that has nothing
+                // to write.
+                return Err(ServerError::ConnectionError(ConnectionError::InvalidWrite));
+            }
+            _ => {
+                // Check if we still have bytes to write for this connection.
+                if !self.connection.pending_write() {
+                    self.state = ClientConnectionState::AwaitingIncoming;
+                }
+            }
+        }
+        Ok(())
+    }
+
+    fn enqueue_response(&mut self, response: Response) {
+        if self.state != ClientConnectionState::Closed {
+            self.connection.enqueue_response(response);
+        }
+        self.in_flight_response_count -= 1;
+    }
+
+    // Returns `true` if the connection is closed and safe to drop.
+    fn is_done(&self) -> bool {
+        self.state == ClientConnectionState::Closed
+            && !self.connection.pending_write()
+            && self.in_flight_response_count == 0
+    }
+}
+
+/// HTTP Server implementation using Unix Domain Sockets and `EPOLL` to
+/// handle multiple connections on the same thread.
+///
+/// The function that does the data exchange is `handle_notifications`.
+/// It can be called in a loop, which will render the thread that the
+/// server runs on incapable of performing other operations, or it can
+/// be used in another `EPOLL` structure, as it provides its `epoll_fd`,
+/// the file descriptor of the epoll structure used within the server,
+/// and it can be added to another one using the `EPOLLIN` flag. Whenever
+/// there is a notification on that fd, `handle_notifications` should be
+/// called once.
+///
+/// # Example
+///
+/// ## Starting and running the server
+///
+/// ```
+/// use micro_http::{HttpServer, Response, StatusCode};
+///
+/// let path_to_socket = "/tmp/example.sock";
+/// std::fs::remove_file(path_to_socket).unwrap_or_default();
+///
+/// // Start the server.
+/// let mut server = HttpServer::new(path_to_socket).unwrap();
+/// server.start_server().unwrap();
+///
+/// // Connect a client to the server so it doesn't block in our example.
+/// let mut socket = std::os::unix::net::UnixStream::connect(path_to_socket).unwrap();
+///
+/// // Server loop processing requests.
+/// loop {
+///     for request in server.requests().unwrap() {
+///         let response = request.process(|request| {
+///             // Your code here.
+///             Response::new(request.http_version(), StatusCode::NoContent)
+///         });
+///         server.respond(response);
+///     }
+///     // Break this example loop.
+///     break;
+/// }
+/// ```
+pub struct HttpServer {
+    /// Socket on which we listen for new connections.
+    socket: UnixListener,
+    /// File descriptor of the server's epoll structure.
+    epoll_fd: RawFd,
+    /// Holds the token-connection pairs of the server.
+    /// Each connection has an associated identification token, which is
+    /// the file descriptor of the underlying stream.
+    /// We use the file descriptor of the stream as the key for mapping
+    /// connections because the 1-to-1 relation is guaranteed by the OS.
+    connections: HashMap<RawFd, ClientConnection<UnixStream>>,
+}
+
+impl HttpServer {
+    /// Constructor for `HttpServer`.
+    ///
+    /// Returns the newly formed `HttpServer`.
+    ///
+    /// # Errors
+    /// Returns an `IOError` when binding or `epoll::create` fails.
+    pub fn new<P: AsRef<Path>>(path_to_socket: P) -> Result<Self> {
+        let socket = UnixListener::bind(path_to_socket).map_err(ServerError::IOError)?;
+        let epoll_fd = epoll::create(true).map_err(ServerError::IOError)?;
+        Ok(HttpServer {
+            socket,
+            epoll_fd,
+            connections: HashMap::new(),
+        })
+    }
+
+    /// Starts the HTTP Server.
+    pub fn start_server(&mut self) -> Result<()> {
+        // Add the socket on which we listen for new connections to the
+        // `epoll` structure.
+        Self::epoll_add(self.epoll_fd, self.socket.as_raw_fd())
+    }
+
+    /// This function is responsible for the data exchange with the clients and should
+    /// be called when we are either notified through `epoll` that we need to exchange
+    /// data with at least a client or when we don't need to perform any other operations
+    /// on this thread and we can afford to call it in a loop.
+    ///
+    /// Note that this function will block the current thread if there are no notifications
+    /// to be handled by the server.
+    ///
+    /// Returns a collection of complete and valid requests to be processed by the user
+    /// of the server. Once processed, responses should be sent using `enqueue_responses()`.
+    ///
+    /// # Errors
+    /// `IOError` is returned when `read`, `write` or `epoll::ctl` operations fail.
+    /// `ServerFull` is returned when a client is trying to connect to the server, but
+    /// full capacity has already been reached.
+    /// `InvalidWrite` is returned when the server attempted to perform a write operation
+    /// on a connection on which it is not possible.
+    pub fn requests(&mut self) -> Result<Vec<ServerRequest>> {
+        let mut parsed_requests: Vec<ServerRequest> = vec![];
+        let mut events = vec![epoll::Event::new(epoll::Events::empty(), 0); MAX_CONNECTIONS];
+
+        // This is a wrapper over the syscall `epoll_wait` and it will block the
+        // current thread until at least one event is received.
+        // The received notifications will then populate the `events` array with
+        // `event_count` elements, where 1 <= event_count <= MAX_CONNECTIONS.
+        let event_count =
+            epoll::wait(self.epoll_fd, -1, &mut events[..]).map_err(ServerError::IOError)?;
+        // We use `take()` on the iterator over `events` as, even though only
+        // `events_count` events have been inserted into `events`, the size of
+        // the array is still `MAX_CONNECTIONS`, so we discard empty elements
+        // at the end of the array.
+        for e in events.iter().take(event_count) {
+            // Check the file descriptor which produced the notification `e`.
+            // It could be that we have a new connection, or one of our open
+            // connections is ready to exchange data with a client.
+            if e.data as RawFd == self.socket.as_raw_fd() {
+                // We have received a notification on the listener socket, which
+                // means we have a new connection to accept.
+                match self.handle_new_connection() {
+                    // If the server is full, we send a message to the client
+                    // notifying them that we will close the connection, then
+                    // we discard it.
+                    Err(ServerError::ServerFull) => {
+                        self.socket
+                            .accept()
+                            .map_err(ServerError::IOError)
+                            .and_then(move |(mut stream, _)| {
+                                stream
+                                    .write(SERVER_FULL_ERROR_MESSAGE)
+                                    .map_err(ServerError::IOError)
+                            })?;
+                    }
+                    // An internal error will compromise any in-flight requests.
+                    Err(error) => return Err(error),
+                    Ok(()) => {}
+                };
+            } else {
+                // We have a notification on one of our open connections.
+                let fd = e.data as RawFd;
+                let client_connection = self.connections.get_mut(&fd).unwrap();
+                if e.events & epoll::Events::EPOLLIN.bits() != 0 {
+                    // We have bytes to read from this connection.
+                    // If our `read` yields `Request` objects, we wrap them with an ID before
+                    // handing them to the user.
+                    parsed_requests.append(
+                        &mut client_connection
+                            .read()?
+                            .into_iter()
+                            .map(|request| ServerRequest::new(request, fd as u64))
+                            .collect(),
+                    );
+                    // If the connection was incoming before we read and we now have to write
+                    // either an error message or an `expect` response, we change its `epoll`
+                    // event set to notify us when the stream is ready for writing.
+                    if client_connection.state == ClientConnectionState::AwaitingOutgoing {
+                        Self::epoll_mod(self.epoll_fd, fd, epoll::Events::EPOLLOUT)?;
+                    }
+                } else if e.events & epoll::Events::EPOLLOUT.bits() != 0 {
+                    // We have bytes to write on this connection.
+                    client_connection.write()?;
+                    // If the connection was outgoing before we tried to write the responses
+                    // and we don't have any more responses to write, we change the `epoll`
+                    // event set to notify us when we have bytes to read from the stream.
+                    if client_connection.state == ClientConnectionState::AwaitingIncoming {
+                        Self::epoll_mod(self.epoll_fd, fd, epoll::Events::EPOLLIN)?;
+                    }
+                }
+            }
+        }
+
+        // Remove dead connections.
+        self.connections
+            .retain(|_, client_connection| !client_connection.is_done());
+
+        Ok(parsed_requests)
+    }
+
+    /// The file descriptor of the `epoll` structure can enable the server to become
+    /// a non-blocking structure in an application.
+    ///
+    /// Returns the file descriptor of the server's internal `epoll` structure.
+    ///
+    /// # Example
+    ///
+    /// ## Non-blocking server
+    /// ```
+    /// extern crate epoll;
+    ///
+    /// use micro_http::{HttpServer, Response, StatusCode};
+    ///
+    /// // Create our epoll manager.
+    /// let epoll_fd = epoll::create(true).unwrap();
+    ///
+    /// let path_to_socket = "/tmp/epoll_example.sock";
+    /// std::fs::remove_file(path_to_socket).unwrap_or_default();
+    ///
+    /// // Start the server.
+    /// let mut server = HttpServer::new(path_to_socket).unwrap();
+    /// server.start_server().unwrap();
+    ///
+    /// // Add our server to the `epoll` manager.
+    /// epoll::ctl(
+    ///     epoll_fd,
+    ///     epoll::ControlOptions::EPOLL_CTL_ADD,
+    ///     server.epoll_fd(),
+    ///     epoll::Event::new(epoll::Events::EPOLLIN, 1234u64),
+    /// )
+    /// .unwrap();
+    ///
+    /// // Connect a client to the server so it doesn't block in our example.
+    /// let mut socket = std::os::unix::net::UnixStream::connect(path_to_socket).unwrap();
+    ///
+    /// // Control loop of the application.
+    /// let mut events = Vec::with_capacity(10);
+    /// loop {
+    ///     let num_ev = epoll::wait(epoll_fd, -1, events.as_mut_slice());
+    ///     for event in events {
+    ///         match event.data {
+    ///             // The server notification.
+    ///             1234 => {
+    ///                 let request = server.requests();
+    ///                 // Process...
+    ///             }
+    ///             // Other `epoll` notifications.
+    ///             _ => {
+    ///                 // Do other computation.
+    ///             }
+    ///         }
+    ///     }
+    ///     // Break this example loop.
+    ///     break;
+    /// }
+    /// ```
+    pub fn epoll_fd(&self) -> RawFd {
+        self.epoll_fd
+    }
+
+    /// Enqueues the provided responses in the outgoing connection.
+    ///
+    /// # Errors
+    /// `IOError` is returned when an `epoll::ctl` operation fails.
+    pub fn enqueue_responses(&mut self, responses: Vec<ServerResponse>) -> Result<()> {
+        for response in responses {
+            self.respond(response)?;
+        }
+
+        Ok(())
+    }
+
+    /// Adds the provided response to the outgoing buffer in the corresponding connection.
+    ///
+    /// # Errors
+    /// `IOError` is returned when an `epoll::ctl` operation fails.
+    pub fn respond(&mut self, response: ServerResponse) -> Result<()> {
+        if let Some(client_connection) = self.connections.get_mut(&(response.id as i32)) {
+            // If the connection was incoming before we enqueue the response, we change its
+            // `epoll` event set to notify us when the stream is ready for writing.
+            if let ClientConnectionState::AwaitingIncoming = client_connection.state {
+                client_connection.state = ClientConnectionState::AwaitingOutgoing;
+                Self::epoll_mod(self.epoll_fd, response.id as RawFd, epoll::Events::EPOLLOUT)?;
+            }
+            client_connection.enqueue_response(response.response);
+        }
+        Ok(())
+    }
+
+    /// Accepts a new incoming connection and adds it to the `epoll` notification structure.
+    ///
+    /// # Errors
+    /// `IOError` is returned when an `epoll::ctl` operation fails.
+    fn handle_new_connection(&mut self) -> Result<()> {
+        if self.connections.len() == MAX_CONNECTIONS {
+            // If we want a replacement policy for connections
+            // this is where we will have it.
+            return Err(ServerError::ServerFull);
+        }
+
+        self.socket
+            .accept()
+            .map_err(ServerError::IOError)
+            .and_then(|(stream, _)| {
+                // `HttpConnection` is supposed to work with non-blocking streams.
+                stream
+                    .set_nonblocking(true)
+                    .map(|_| stream)
+                    .map_err(ServerError::IOError)
+            })
+            .and_then(|stream| {
+                // Add the stream to the `epoll` structure and listen for bytes to be read.
+                Self::epoll_add(self.epoll_fd, stream.as_raw_fd())?;
+                // Then add it to our open connections.
+                self.connections.insert(
+                    stream.as_raw_fd(),
+                    ClientConnection::new(HttpConnection::new(stream)),
+                );
+                Ok(())
+            })
+    }
+
+    /// Changes the event type for a connection to either listen for incoming bytes
+    /// or for when the stream is ready for writing.
+    fn epoll_mod(epoll_fd: RawFd, stream_fd: RawFd, evset: epoll::Events) -> Result<()> {
+        let event = epoll::Event::new(evset, stream_fd as u64);
+        epoll::ctl(
+            epoll_fd,
+            epoll::ControlOptions::EPOLL_CTL_MOD,
+            stream_fd,
+            event,
+        )
+        .map_err(ServerError::IOError)
+    }
+
+    /// Adds a stream to the `epoll` notification structure with the `EPOLLIN` event set.
+    fn epoll_add(epoll_fd: RawFd, stream_fd: RawFd) -> Result<()> {
+        epoll::ctl(
+            epoll_fd,
+            epoll::ControlOptions::EPOLL_CTL_ADD,
+            stream_fd,
+            epoll::Event::new(epoll::Events::EPOLLIN, stream_fd as u64),
+        )
+        .map_err(ServerError::IOError)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::fs;
+    use std::os::unix::net::UnixStream;
+
+    use common::Body;
+    use std::io::Read;
+    use std::io::Write;
+
+    #[test]
+    fn test_wait_one_connection() {
+        let path_to_socket = "/tmp/test_socket_http_server1.sock";
+        fs::remove_file(path_to_socket).unwrap_or_default();
+
+        let mut server = HttpServer::new(path_to_socket.to_string()).unwrap();
+        server.start_server().unwrap();
+
+        // Test one incoming connection.
+        let mut socket = UnixStream::connect(path_to_socket).unwrap();
+        assert!(server.requests().unwrap().is_empty());
+
+        socket
+            .write_all(
+                b"PATCH /machine-config HTTP/1.1\r\n\
+                         Content-Length: 13\r\n\
+                         Content-Type: application/json\r\n\r\nwhatever body",
+            )
+            .unwrap();
+
+        let mut req_vec = server.requests().unwrap();
+        let server_request = req_vec.remove(0);
+
+        server
+            .respond(server_request.process(|_request| {
+                let mut response = Response::new(Version::Http11, StatusCode::OK);
+                let response_body = b"response body";
+                response.set_body(Body::new(response_body.to_vec()));
+                response
+            }))
+            .unwrap();
+        assert!(server.requests().unwrap().is_empty());
+
+        let mut buf: [u8; 1024] = [0; 1024];
+        assert!(socket.read(&mut buf[..]).unwrap() > 0);
+        fs::remove_file(path_to_socket).unwrap();
+    }
+
+    #[test]
+    fn test_wait_concurrent_connections() {
+        let path_to_socket = "/tmp/test_socket_http_server2.sock";
+        fs::remove_file(path_to_socket).unwrap_or_default();
+
+        let mut server = HttpServer::new(path_to_socket.to_string()).unwrap();
+        server.start_server().unwrap();
+
+        // Test two concurrent connections.
+        let mut first_socket = UnixStream::connect(path_to_socket).unwrap();
+        assert!(server.requests().unwrap().is_empty());
+
+        first_socket
+            .write_all(
+                b"PATCH /machine-config HTTP/1.1\r\n\
+                               Content-Length: 13\r\n\
+                               Content-Type: application/json\r\n\r\nwhatever body",
+            )
+            .unwrap();
+        let mut second_socket = UnixStream::connect(path_to_socket).unwrap();
+
+        let mut req_vec = server.requests().unwrap();
+        let server_request = req_vec.remove(0);
+
+        server
+            .respond(server_request.process(|_request| {
+                let mut response = Response::new(Version::Http11, StatusCode::OK);
+                let response_body = b"response body";
+                response.set_body(Body::new(response_body.to_vec()));
+                response
+            }))
+            .unwrap();
+        second_socket
+            .write_all(
+                b"GET /machine-config HTTP/1.1\r\n\
+                                Content-Length: 20\r\n\
+                                Content-Type: application/json\r\n\r\nwhatever second body",
+            )
+            .unwrap();
+
+        let mut req_vec = server.requests().unwrap();
+        let second_server_request = req_vec.remove(0);
+
+        assert_eq!(
+            second_server_request.request,
+            Request::try_from(
+                b"GET /machine-config HTTP/1.1\r\n\
+            Content-Length: 20\r\n\
+            Content-Type: application/json\r\n\r\nwhatever second body"
+            )
+            .unwrap()
+        );
+
+        let mut buf: [u8; 1024] = [0; 1024];
+        assert!(first_socket.read(&mut buf[..]).unwrap() > 0);
+        first_socket.shutdown(std::net::Shutdown::Both).unwrap();
+
+        server
+            .respond(second_server_request.process(|_request| {
+                let mut response = Response::new(Version::Http11, StatusCode::OK);
+                let response_body = b"response second body";
+                response.set_body(Body::new(response_body.to_vec()));
+                response
+            }))
+            .unwrap();
+
+        assert!(server.requests().unwrap().is_empty());
+        let mut buf: [u8; 1024] = [0; 1024];
+        assert!(second_socket.read(&mut buf[..]).unwrap() > 0);
+        second_socket.shutdown(std::net::Shutdown::Both).unwrap();
+        assert!(server.requests().unwrap().is_empty());
+        fs::remove_file(path_to_socket).unwrap();
+    }
+
+    #[test]
+    fn test_wait_expect_connection() {
+        let path_to_socket = "/tmp/test_socket_http_server3.sock";
+        fs::remove_file(path_to_socket).unwrap_or_default();
+
+        let mut server = HttpServer::new(path_to_socket.to_string()).unwrap();
+        server.start_server().unwrap();
+
+        // Test one incoming connection with `Expect: 100-continue`.
+        let mut socket = UnixStream::connect(path_to_socket).unwrap();
+        assert!(server.requests().unwrap().is_empty());
+
+        socket
+            .write_all(
+                b"PATCH /machine-config HTTP/1.1\r\n\
+                         Content-Length: 13\r\n\
+                         Expect: 100-continue\r\n\r\n",
+            )
+            .unwrap();
+        // `wait` on server to receive what the client set on the socket.
+        // This will set the stream direction to `Outgoing`, as we need to send a `100 CONTINUE` response.
+        let req_vec = server.requests().unwrap();
+        assert!(req_vec.is_empty());
+        // Another `wait`, this time to send the response.
+        // Will be called because of an `EPOLLOUT` notification.
+        let req_vec = server.requests().unwrap();
+        assert!(req_vec.is_empty());
+        let mut buf: [u8; 1024] = [0; 1024];
+        assert!(socket.read(&mut buf[..]).unwrap() > 0);
+
+        socket.write_all(b"whatever body").unwrap();
+        let mut req_vec = server.requests().unwrap();
+        let server_request = req_vec.remove(0);
+
+        server
+            .respond(server_request.process(|_request| {
+                let mut response = Response::new(Version::Http11, StatusCode::OK);
+                let response_body = b"response body";
+                response.set_body(Body::new(response_body.to_vec()));
+                response
+            }))
+            .unwrap();
+
+        let req_vec = server.requests().unwrap();
+        assert!(req_vec.is_empty());
+
+        let mut buf: [u8; 1024] = [0; 1024];
+        assert!(socket.read(&mut buf[..]).unwrap() > 0);
+        fs::remove_file(path_to_socket).unwrap();
+    }
+
+    #[test]
+    fn test_wait_many_connections() {
+        let path_to_socket = "/tmp/test_socket_http_server4.sock";
+        fs::remove_file(path_to_socket).unwrap_or_default();
+
+        let mut server = HttpServer::new(path_to_socket.to_string()).unwrap();
+        server.start_server().unwrap();
+
+        let mut sockets: Vec<UnixStream> = Vec::with_capacity(11);
+        for _ in 0..MAX_CONNECTIONS {
+            sockets.push(UnixStream::connect(path_to_socket).unwrap());
+            assert!(server.requests().unwrap().is_empty());
+        }
+
+        sockets.push(UnixStream::connect(path_to_socket).unwrap());
+        assert!(server.requests().unwrap().is_empty());
+        let mut buf: [u8; 120] = [0; 120];
+        sockets[MAX_CONNECTIONS].read_exact(&mut buf).unwrap();
+        assert_eq!(&buf[..], SERVER_FULL_ERROR_MESSAGE);
+
+        fs::remove_file(path_to_socket).unwrap();
+    }
+
+    #[test]
+    fn test_wait_parse_error() {
+        let path_to_socket = "/tmp/test_socket_http_server5.sock";
+        fs::remove_file(path_to_socket).unwrap_or_default();
+
+        let mut server = HttpServer::new(path_to_socket.to_string()).unwrap();
+        server.start_server().unwrap();
+
+        // Test one incoming connection.
+        let mut socket = UnixStream::connect(path_to_socket).unwrap();
+        socket.set_nonblocking(true).unwrap();
+        assert!(server.requests().unwrap().is_empty());
+
+        socket
+            .write_all(
+                b"PATCH /machine-config HTTP/1.1\r\n\
+                         Content-Length: alpha\r\n\
+                         Content-Type: application/json\r\n\r\nwhatever body",
+            )
+            .unwrap();
+
+        assert!(server.requests().unwrap().is_empty());
+        assert!(server.requests().unwrap().is_empty());
+        let mut buf: [u8; 198] = [0; 198];
+        assert!(socket.read(&mut buf[..]).unwrap() > 0);
+        let error_message = b"HTTP/1.1 400 \r\n\
+                              Server: Firecracker API\r\n\
+                              Connection: keep-alive\r\n\
+                              Content-Type: application/json\r\n\
+                              Content-Length: 80\r\n\r\n{ \"error\": \"Invalid header.\n\
+                              All previous unanswered requests will be dropped.\" }";
+        assert_eq!(&buf[..], &error_message[..]);
+
+        fs::remove_file(path_to_socket).unwrap();
+    }
+
+    #[test]
+    fn test_wait_in_flight_responses() {
+        let path_to_socket = "/tmp/test_socket_http_server6.sock";
+        fs::remove_file(path_to_socket).unwrap_or_default();
+
+        let mut server = HttpServer::new(path_to_socket.to_string()).unwrap();
+        server.start_server().unwrap();
+
+        // Test a connection dropped and then a new one appearing
+        // before the user had a chance to send the response to the
+        // first one.
+        let mut first_socket = UnixStream::connect(path_to_socket).unwrap();
+        assert!(server.requests().unwrap().is_empty());
+
+        first_socket
+            .write_all(
+                b"PATCH /machine-config HTTP/1.1\r\n\
+                               Content-Length: 13\r\n\
+                               Content-Type: application/json\r\n\r\nwhatever body",
+            )
+            .unwrap();
+
+        let mut req_vec = server.requests().unwrap();
+        let server_request = req_vec.remove(0);
+
+        first_socket.shutdown(std::net::Shutdown::Both).unwrap();
+        assert!(server.requests().unwrap().is_empty());
+        let mut second_socket = UnixStream::connect(path_to_socket).unwrap();
+        second_socket.set_nonblocking(true).unwrap();
+        assert!(server.requests().unwrap().is_empty());
+
+        server
+            .enqueue_responses(vec![server_request.process(|_request| {
+                let mut response = Response::new(Version::Http11, StatusCode::OK);
+                let response_body = b"response body";
+                response.set_body(Body::new(response_body.to_vec()));
+                response
+            })])
+            .unwrap();
+        assert!(server.requests().unwrap().is_empty());
+        assert_eq!(server.connections.len(), 1);
+        let mut buf: [u8; 1024] = [0; 1024];
+        assert!(second_socket.read(&mut buf[..]).is_err());
+
+        second_socket
+            .write_all(
+                b"GET /machine-config HTTP/1.1\r\n\
+                                Content-Length: 20\r\n\
+                                Content-Type: application/json\r\n\r\nwhatever second body",
+            )
+            .unwrap();
+
+        let mut req_vec = server.requests().unwrap();
+        let second_server_request = req_vec.remove(0);
+
+        assert_eq!(
+            second_server_request.request,
+            Request::try_from(
+                b"GET /machine-config HTTP/1.1\r\n\
+            Content-Length: 20\r\n\
+            Content-Type: application/json\r\n\r\nwhatever second body"
+            )
+            .unwrap()
+        );
+
+        server
+            .respond(second_server_request.process(|_request| {
+                let mut response = Response::new(Version::Http11, StatusCode::OK);
+                let response_body = b"response second body";
+                response.set_body(Body::new(response_body.to_vec()));
+                response
+            }))
+            .unwrap();
+
+        assert!(server.requests().unwrap().is_empty());
+        let mut buf: [u8; 1024] = [0; 1024];
+        assert!(second_socket.read(&mut buf[..]).unwrap() > 0);
+        second_socket.shutdown(std::net::Shutdown::Both).unwrap();
+        assert!(server.requests().is_ok());
+        fs::remove_file(path_to_socket).unwrap();
+    }
+}