Merge pull request routerify#8 from seanpianka/master

Update to Tokio 1.0

Author: rousan

Cargo.toml

@@ -12,8 +12,8 @@ license = "MIT"
 edition = "2018"
 
 [dependencies]
-tokio = { version = "0.2", features= [] }
+tokio = { version = "1", features= [] }
 log = "0.4"
 
 [dev-dependencies]
-tokio = { version = "0.2", features = ["full"] }
+tokio = { version = "1", features = ["full"] }

examples/main.rs

@@ -1,5 +1,5 @@
 use async_pipe;
-use tokio::prelude::*;
+use tokio::io::{AsyncWriteExt, AsyncReadExt};
 
 #[tokio::main]
 async fn main() {

src/lib.rs

@@ -5,7 +5,7 @@
 //! ```
 //! # async fn run() {
 //! use async_pipe;
-//! use tokio::prelude::*;
+//! use tokio::io::{AsyncWriteExt, AsyncReadExt};
 //!
 //! let (mut w, mut r) = async_pipe::pipe();
 //!  
@@ -45,12 +45,34 @@ pub fn pipe() -> (PipeWriter, PipeReader) {
     }));
 
     let w = PipeWriter {
-        state: Arc::clone(&shared_state),
+        state: shared_state.clone()
     };
 
     let r = PipeReader {
-        state: Arc::clone(&shared_state),
+        state: shared_state.clone(),
     };
 
     (w, r)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::io::{AsyncWriteExt, AsyncReadExt};
+
+    #[tokio::test]
+    async fn should_read_expected_text() {
+        const EXPECTED: &'static str = "hello world";
+
+        let (mut w, mut r) = pipe();
+
+        tokio::spawn(async move {
+            w.write_all(EXPECTED.as_bytes()).await.unwrap();
+        });
+
+        let mut v = Vec::new();
+        r.read_to_end(&mut v).await.unwrap();
+        let actual = String::from_utf8(v).unwrap();
+        assert_eq!(EXPECTED, actual.as_str());
+    }
+}

src/reader.rs

@@ -3,7 +3,7 @@ use std::pin::Pin;
 use std::ptr;
 use std::sync::{Arc, Mutex};
 use std::task::{Context, Poll};
-use tokio::io::{self, AsyncRead};
+use tokio::io::{self, AsyncRead, ReadBuf};
 
 /// The read half of the pipe which implements [`AsyncRead`](https://docs.rs/tokio/0.2.15/tokio/io/trait.AsyncRead.html).
 pub struct PipeReader {
@@ -55,10 +55,10 @@ impl PipeReader {
         }
     }
 
-    fn copy_data_into_buffer(&self, data: &Data, buf: &mut [u8]) -> usize {
-        let len = data.len.min(buf.len());
+    fn copy_data_into_buffer(&self, data: &Data, buf: &mut ReadBuf) -> usize {
+        let len = data.len.min(buf.capacity());
         unsafe {
-            ptr::copy_nonoverlapping(data.ptr, buf.as_mut_ptr(), len);
+            ptr::copy_nonoverlapping(data.ptr, buf.initialize_unfilled().as_mut_ptr(), len);
         }
         len
     }
@@ -80,8 +80,8 @@ impl AsyncRead for PipeReader {
     fn poll_read(
         self: Pin<&mut Self>,
         cx: &mut Context,
-        buf: &mut [u8],
-    ) -> Poll<io::Result<usize>> {
+        buf: &mut ReadBuf,
+    ) -> Poll<io::Result<()>> {
         let mut state;
         match self.state.lock() {
             Ok(s) => state = s,
@@ -98,7 +98,7 @@ impl AsyncRead for PipeReader {
         }
 
         if state.closed {
-            return Poll::Ready(Ok(0));
+            return Poll::Ready(Ok(()));
         }
 
         return if state.done_cycle {
@@ -115,7 +115,7 @@ impl AsyncRead for PipeReader {
 
                 self.wake_writer_half(&*state);
 
-                Poll::Ready(Ok(copied_bytes_len))
+                Poll::Ready(Ok(()))
             } else {
                 state.reader_waker = Some(cx.waker().clone());
                 Poll::Pending