keep alive basics

Author: rylev

src/body.rs

@@ -49,6 +49,12 @@ impl<R: AsyncRead> Body<R> {
     }
 }
 
+impl<R: AsyncRead + Unpin> Body<R> {
+    pub fn into_reader(self) -> Option<R> {
+        self.reader
+    }
+}
+
 impl Body<io::Cursor<Vec<u8>>> {
     /// Create a new instance from a string.
     #[inline]

src/client.rs

@@ -64,7 +64,6 @@ pub async fn encode<R: AsyncRead>(req: Request<Body<R>>) -> Result<Encoder<R>, s
         // See: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Transfer-Encoding
         //      https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Trailer
     }
-
     for (header, value) in req.headers() {
         write!(
             &mut buf,

src/server.rs

@@ -1,21 +1,57 @@
 //! Process HTTP connections on the server.
 
+use async_std::future::{timeout, Future};
 use async_std::io::{self, BufReader};
+use async_std::io::{Read, Write};
 use async_std::prelude::*;
 use async_std::task::{Context, Poll};
 use futures_core::ready;
-use futures_io::AsyncRead;
 use http::{Request, Response, Version};
 
 use std::pin::Pin;
 
 use crate::{Body, Exception, MAX_HEADERS};
 
+pub async fn connect<'a, F, Fut, B>(
+    reader: &'a mut B,
+    writer: &'a mut B,
+    callback: F,
+) -> Result<(), Exception>
+where
+    F: Fn(&mut Request<Body<BufReader<&'a mut B>>>) -> Fut,
+    Fut: Future<Output = Result<Response<Body<&'a mut B>>, Exception>>,
+    B: Read + Write + Unpin + Send + 'static,
+{
+    let decoder = Decoder::new(reader);
+    let req = decoder.decode().await?;
+    if let Some(mut req) = req {
+        // TODO: parse Keep Alive header
+
+        let handle = async {
+            loop {
+                let mut res = encode(callback(&mut req).await?).await.unwrap();
+                io::copy(&mut res, writer).await?;
+                let reader = res.body.into_reader().unwrap();
+                let decoder = Decoder::new(reader);
+                match decoder.decode().await? {
+                    // TODO: no unwrap
+                    Some(r) => req = r,
+                    None => break,
+                }
+            }
+
+            Ok::<(), Exception>(())
+        };
+    }
+
+    Ok(())
+}
+
 /// A streaming HTTP encoder.
 ///
 /// This is returned from [`encode`].
 #[derive(Debug)]
-pub struct Encoder<R: AsyncRead> {
+pub struct Encoder<R: Read> {
     /// Keep track how far we've indexed into the headers + body.
     cursor: usize,
     /// HTTP headers to be sent.
@@ -30,7 +66,7 @@ pub struct Encoder<R: AsyncRead> {
     body_bytes_read: usize,
 }
 
-impl<R: AsyncRead> Encoder<R> {
+impl<R: Read> Encoder<R> {
     /// Create a new instance.
     pub(crate) fn new(headers: Vec<u8>, body: Body<R>) -> Self {
         Self {
@@ -44,7 +80,7 @@ impl<R: AsyncRead> Encoder<R> {
     }
 }
 
-impl<R: AsyncRead + Unpin> AsyncRead for Encoder<R> {
+impl<R: Read + Unpin> Read for Encoder<R> {
     fn poll_read(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
@@ -81,7 +117,7 @@ impl<R: AsyncRead + Unpin> AsyncRead for Encoder<R> {
 // TODO: return a reader in the response
 pub async fn encode<R>(res: Response<Body<R>>) -> io::Result<Encoder<R>>
 where
-    R: AsyncRead,
+    R: Read,
 {
     let mut buf: Vec<u8> = vec![];
 
@@ -114,66 +150,77 @@ where
     Ok(Encoder::new(buf, res.into_body()))
 }
 
-/// Decode an HTTP request on the server.
-pub async fn decode<R>(reader: R) -> Result<Option<Request<Body<BufReader<R>>>>, Exception>
-where
-    R: AsyncRead + Unpin + Send,
-{
-    let mut reader = BufReader::new(reader);
-    let mut buf = Vec::new();
-    let mut headers = [httparse::EMPTY_HEADER; MAX_HEADERS];
-    let mut httparse_req = httparse::Request::new(&mut headers);
-
-    // Keep reading bytes from the stream until we hit the end of the stream.
-    loop {
-        let bytes_read = reader.read_until(b'\n', &mut buf).await?;
-        // No more bytes are yielded from the stream.
-        if bytes_read == 0 {
-            return Ok(None);
-        }
+struct Decoder<R> {
+    reader: BufReader<R>,
+}
 
-        // We've hit the end delimiter of the stream.
-        let idx = buf.len() - 1;
-        if idx >= 3 && &buf[idx - 3..=idx] == b"\r\n\r\n" {
-            break;
+impl<R: Read + Unpin + Send> Decoder<R> {
+    fn new(reader: R) -> Self {
+        Decoder {
+            reader: BufReader::new(reader),
         }
     }
 
-    // Convert our header buf into an httparse instance, and validate.
-    let status = httparse_req.parse(&buf)?;
-    if status.is_partial() {
-        dbg!(String::from_utf8(buf).unwrap());
-        return Err("Malformed HTTP head".into());
-    }
+    /// Decode an HTTP request on the server.
+    pub async fn decode(mut self) -> Result<Option<Request<Body<BufReader<R>>>>, Exception>
+    where
+        R: Read + Unpin + Send,
+    {
+        let mut buf = Vec::new();
+        let mut headers = [httparse::EMPTY_HEADER; MAX_HEADERS];
+        let mut httparse_req = httparse::Request::new(&mut headers);
+
+        // Keep reading bytes from the stream until we hit the end of the stream.
+        loop {
+            let bytes_read = self.reader.read_until(b'\n', &mut buf).await?;
+            // No more bytes are yielded from the stream.
+            if bytes_read == 0 {
+                return Ok(None);
+            }
 
-    // Convert httparse headers + body into a `http::Request` type.
-    let mut req = Request::builder();
-    for header in httparse_req.headers.iter() {
-        req.header(header.name, header.value);
-    }
-    if let Some(method) = httparse_req.method {
-        req.method(method);
-    }
-    if let Some(path) = httparse_req.path {
-        req.uri(path);
-    }
-    if let Some(version) = httparse_req.version {
-        req.version(match version {
-            1 => Version::HTTP_11,
-            _ => return Err("Unsupported HTTP version".into()),
-        });
-    }
+            // We've hit the end delimiter of the stream.
+            let idx = buf.len() - 1;
+            if idx >= 3 && &buf[idx - 3..=idx] == b"\r\n\r\n" {
+                break;
+            }
+        }
 
-    // Process the body if `Content-Length` was passed.
-    let body = match httparse_req
-        .headers
-        .iter()
-        .find(|h| h.name == "Content-Length")
-    {
-        Some(_header) => Body::new(reader), // TODO: use the header value
-        None => Body::empty(),
-    };
+        // Convert our header buf into an httparse instance, and validate.
+        let status = httparse_req.parse(&buf)?;
+        if status.is_partial() {
+            dbg!(String::from_utf8(buf).unwrap());
+            return Err("Malformed HTTP head".into());
+        }
 
-    // Return the request.
-    Ok(Some(req.body(body)?))
+        // Convert httparse headers + body into a `http::Request` type.
+        let mut req = Request::builder();
+        for header in httparse_req.headers.iter() {
+            req.header(header.name, header.value);
+        }
+        if let Some(method) = httparse_req.method {
+            req.method(method);
+        }
+        if let Some(path) = httparse_req.path {
+            req.uri(path);
+        }
+        if let Some(version) = httparse_req.version {
+            req.version(match version {
+                1 => Version::HTTP_11,
+                _ => return Err("Unsupported HTTP version".into()),
+            });
+        }
+
+        // Process the body if `Content-Length` was passed.
+        let body = match httparse_req
+            .headers
+            .iter()
+            .find(|h| h.name == "Content-Length")
+        {
+            Some(_header) => Body::new(self.reader), // TODO: use the header value
+            None => Body::empty(),
+        };
+
+        // Return the request.
+        Ok(Some(req.body(body)?))
+    }
 }