Merge pull request #39 from yoshuawuyts/single-io

Unified Reader and Writer

Author: yoshuawuyts

examples/server.rs

@@ -16,7 +16,7 @@ async fn accept(addr: String, stream: TcpStream) -> Result<(), async_h1::Excepti
     // TODO: Delete this line when we implement `Clone` for `TcpStream`.
     let stream = Stream(Arc::new(stream));
 
-    server::accept(&addr, stream.clone(), stream, |req| {
+    server::accept(&addr, stream.clone(), |req| {
         async move {
             dbg!(req.method());
             let mut resp = Response::new(StatusCode::Ok);

src/server.rs

@@ -19,15 +19,9 @@ use crate::{Exception, MAX_HEADERS};
 /// Parse an incoming HTTP connection.
 ///
 /// Supports `KeepAlive` requests by default.
-pub async fn accept<R, W, F, Fut>(
-    addr: &str,
-    reader: R,
-    mut writer: W,
-    endpoint: F,
-) -> Result<(), Exception>
+pub async fn accept<RW, F, Fut>(addr: &str, mut io: RW, endpoint: F) -> Result<(), Exception>
 where
-    R: Read + Unpin + Send + 'static + Clone,
-    W: Write + Unpin,
+    RW: Read + Write + Clone + Send + Unpin + 'static,
     F: Fn(Request) -> Fut,
     Fut: Future<Output = Result<Response, Exception>>,
 {
@@ -37,7 +31,7 @@ where
     let mut num_requests = 0;
 
     // Decode a request. This may be the first of many since the connection is Keep-Alive by default.
-    let r = reader.clone();
+    let r = io.clone();
     let req = decode(addr, r).await?;
     if let Some(mut req) = req {
         loop {
@@ -49,11 +43,11 @@ where
             // TODO: what to do when the endpoint returns Err
             let res = endpoint(req).await?;
             let mut encoder = Encoder::encode(res);
-            io::copy(&mut encoder, &mut writer).await?;
+            io::copy(&mut encoder, &mut io).await?;
 
             // Decode a new request, timing out if this takes longer than the
             // timeout duration.
-            req = match timeout(timeout_duration, decode(addr, reader.clone())).await {
+            req = match timeout(timeout_duration, decode(addr, io.clone())).await {
                 Ok(Ok(Some(r))) => r,
                 Ok(Ok(None)) | Err(TimeoutError { .. }) => break, /* EOF or timeout */
                 Ok(Err(e)) => return Err(e),

tests/common/mod.rs

@@ -1,17 +1,19 @@
 use async_std::fs::File;
-use async_std::io::{self, Read, Write};
+use async_std::fs::OpenOptions;
+use async_std::io::{self, Read, SeekFrom, Write};
 use async_std::path::PathBuf;
 use async_std::sync::Arc;
 use async_std::task::{Context, Poll};
 use std::pin::Pin;
+use std::sync::Mutex;
 
 #[macro_export]
 macro_rules! assert {
     ($actual:expr, $expected_file:expr, $block:expr) => {
         task::block_on(async {
             use async_std::io::prelude::*;
             $block.await.unwrap();
-            let mut actual = std::string::String::from_utf8($actual).unwrap();
+            let mut actual = $actual.to_string().await;
             let mut expected = std::string::String::new();
             $expected_file.read_to_string(&mut expected).await.unwrap();
             match expected.find("{DATE}") {
@@ -38,11 +40,29 @@ pub async fn read_fixture(name: &str) -> TestFile {
     let file = File::open(directory.join(path))
         .await
         .expect("Reading fixture file didn't work");
-    TestFile(Arc::new(file))
+    let temp = std::env::temp_dir().join("foo.txt");
+    let temp = OpenOptions::new()
+        .read(true)
+        .write(true)
+        .open(temp)
+        .await
+        .unwrap();
+    TestFile(Arc::new(file), Arc::new(Mutex::new(temp)))
 }
 
 #[derive(Clone)]
-pub struct TestFile(Arc<File>);
+pub struct TestFile(Arc<File>, Arc<Mutex<File>>);
+
+impl TestFile {
+    pub async fn to_string(self) -> String {
+        use async_std::prelude::*;
+        let mut buf = String::new();
+        let mut file = self.1.lock().unwrap();
+        file.seek(SeekFrom::Start(0)).await.unwrap();
+        dbg!(file.read_to_string(&mut buf).await.unwrap());
+        buf
+    }
+}
 
 impl Read for TestFile {
     fn poll_read(
@@ -56,14 +76,14 @@ impl Read for TestFile {
 
 impl Write for TestFile {
     fn poll_write(self: Pin<&mut Self>, cx: &mut Context, buf: &[u8]) -> Poll<io::Result<usize>> {
-        Pin::new(&mut &*self.0).poll_write(cx, buf)
+        Pin::new(&mut &*self.1.lock().unwrap()).poll_write(cx, buf)
     }
 
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
-        Pin::new(&mut &*self.0).poll_flush(cx)
+        Pin::new(&mut &*self.1.lock().unwrap()).poll_flush(cx)
     }
 
     fn poll_close(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
-        Pin::new(&mut &*self.0).poll_close(cx)
+        Pin::new(&mut &*self.1.lock().unwrap()).poll_close(cx)
     }
 }

tests/server.rs

@@ -7,21 +7,20 @@ use http_types::{Response, StatusCode};
 #[test]
 fn test_basic_request() {
     task::block_on(async {
-        let request = read_fixture("request1").await;
+        let io = read_fixture("request1").await;
         let mut expected = read_fixture("response1").await;
-        let mut actual = Vec::new();
         let addr = "http://example.com";
 
         assert!(
-            actual,
+            io.clone(),
             expected,
-            server::accept(addr, request.clone(), &mut actual, |_req| {
+            server::accept(addr, io.clone(), |_req| {
                 async {
                     let mut resp = Response::new(StatusCode::Ok);
                     resp.set_body("");
                     Ok(resp)
                 }
-           })
+            })
         );
     })
 }