Merge branch 'release/v0.1.0'

.gitignore

@@ -41,4 +41,8 @@ Cargo.lock
 # These are backup files generated by rustfmt
 **/*.rs.bk
 
+# Others
+async-pipe-rs.iws
+
+
 # End of https://www.gitignore.io/api/rust,macos

Cargo.toml

@@ -1,9 +1,19 @@
 [package]
-name = "async-pipe-rs"
+name = "async-pipe"
 version = "0.1.0"
+description = "Creates an asynchronous piped reader and writer pair using tokio.rs"
+homepage = "https://github.com/rousan/async-pipe-rs"
+repository = "https://github.com/rousan/async-pipe-rs"
+keywords = ["pipe", "future", "async", "reader", "writer"]
+categories = ["asynchronous"]
 authors = ["Rousan Ali <[email protected]>"]
+readme = "README.md"
+license = "MIT"
 edition = "2018"
 
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
 [dependencies]
+tokio = { version = "0.2", features= [] }
+log = "0.4"
+
+[dev-dependencies]
+tokio = { version = "0.2", features = ["full"] }

README.md

@@ -1,5 +1,33 @@
-# pipe-rs
+# async-pipe-rs
 
-```Rust
-pipe<W: AsyncWrite, R: AsyncRead>() -> (W, R)
+[![crates.io](https://img.shields.io/crates/v/async-rs.svg)](https://crates.io/crates/async-rs)
+[![Documentation](https://docs.rs/async-rs/badge.svg)](https://docs.rs/async-rs)
+[![MIT](https://img.shields.io/crates/l/async-rs.svg)](./LICENSE)
+
+Creates an asynchronous piped reader and writer pair using `tokio.rs`.
+
+[Docs](https://docs.rs/async-rs)
+
+## Example
+
+```rust
+use async_pipe;
+use tokio::prelude::*;
+
+#[tokio::main]
+async fn main() {
+    let (mut w, mut r) = async_pipe::pipe();
+
+    tokio::spawn(async move {
+        w.write_all(b"hello world").await.unwrap();
+    });
+
+    let mut v = Vec::new();
+    r.read_to_end(&mut v).await.unwrap();
+    println!("Received: {:?}", String::from_utf8(v));
+}
 ```
+
+## Contributing
+
+Your PRs and stars are always welcome.

async-pipe-rs.iml

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="RUST_MODULE" version="4">
+  <component name="NewModuleRootManager" inherit-compiler-output="true">
+    <exclude-output />
+    <content url="file://$MODULE_DIR$">
+      <sourceFolder url="file://$MODULE_DIR$/src" isTestSource="false" />
+      <sourceFolder url="file://$MODULE_DIR$/examples" isTestSource="false" />
+      <sourceFolder url="file://$MODULE_DIR$/tests" isTestSource="true" />
+      <sourceFolder url="file://$MODULE_DIR$/benches" isTestSource="true" />
+      <excludeFolder url="file://$MODULE_DIR$/target" />
+    </content>
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>

examples/main.rs

@@ -0,0 +1,15 @@
+use async_pipe;
+use tokio::prelude::*;
+
+#[tokio::main]
+async fn main() {
+    let (mut w, mut r) = async_pipe::pipe();
+
+    tokio::spawn(async move {
+        w.write_all(b"hello world").await.unwrap();
+    });
+
+    let mut v = Vec::new();
+    r.read_to_end(&mut v).await.unwrap();
+    println!("Received: {:?}", String::from_utf8(v));
+}

src/lib.rs

@@ -1,7 +1,56 @@
-#[cfg(test)]
-mod tests {
-    #[test]
-    fn it_works() {
-        assert_eq!(2 + 2, 4);
-    }
+//! Creates an asynchronous piped reader and writer pair using `tokio.rs`.
+//!
+//! # Examples
+//!
+//! ```
+//! # async fn run() {
+//! use async_pipe;
+//! use tokio::prelude::*;
+//!
+//! let (mut w, mut r) = async_pipe::pipe();
+//!  
+//! tokio::spawn(async move {
+//!     w.write_all(b"hello world").await.unwrap();
+//! });
+//!  
+//! let mut v = Vec::new();
+//! r.read_to_end(&mut v).await.unwrap();
+//!
+//! println!("Received: {:?}", String::from_utf8(v));
+//! # }
+//!
+//! tokio::runtime::Runtime::new().unwrap().block_on(run());
+//! ```
+
+use state::State;
+use std::sync::{Arc, Mutex};
+
+pub use self::reader::PipeReader;
+pub use self::writer::PipeWriter;
+
+mod reader;
+mod state;
+mod writer;
+
+/// Creates a piped pair of an [`AsyncWrite`](https://docs.rs/tokio/0.2.16/tokio/io/trait.AsyncWrite.html) and an [`AsyncRead`](https://docs.rs/tokio/0.2.15/tokio/io/trait.AsyncRead.html).
+pub fn pipe() -> (PipeWriter, PipeReader) {
+    let shared_state = Arc::new(Mutex::new(State {
+        reader_waker: None,
+        writer_waker: None,
+        data: None,
+        done_reading: false,
+        read: 0,
+        done_cycle: true,
+        closed: false,
+    }));
+
+    let w = PipeWriter {
+        state: Arc::clone(&shared_state),
+    };
+
+    let r = PipeReader {
+        state: Arc::clone(&shared_state),
+    };
+
+    (w, r)
 }

src/reader.rs

@@ -0,0 +1,106 @@
+use crate::state::{Data, State};
+use std::pin::Pin;
+use std::ptr;
+use std::sync::{Arc, Mutex};
+use std::task::{Context, Poll};
+use tokio::io::{self, AsyncRead};
+
+/// The read half of the pipe which implements [`AsyncRead`](https://docs.rs/tokio/0.2.15/tokio/io/trait.AsyncRead.html).
+pub struct PipeReader {
+    pub(crate) state: Arc<Mutex<State>>,
+}
+
+impl PipeReader {
+    /// Closes the pipe, any further read will return EOF and any further write will raise an error.
+    pub fn close(&self) -> io::Result<()> {
+        match self.state.lock() {
+            Ok(mut state) => {
+                state.closed = true;
+                self.wake_writer_half(&*state);
+                Ok(())
+            }
+            Err(err) => Err(io::Error::new(
+                io::ErrorKind::Other,
+                format!(
+                    "{}: PipeReader: Failed to lock the channel state: {}",
+                    env!("CARGO_PKG_NAME"),
+                    err
+                ),
+            )),
+        }
+    }
+
+    fn wake_writer_half(&self, state: &State) {
+        if let Some(ref waker) = state.writer_waker {
+            waker.clone().wake();
+        }
+    }
+
+    fn copy_data_into_buffer(&self, data: &Data, buf: &mut [u8]) -> usize {
+        let len = data.len.min(buf.len());
+        unsafe {
+            ptr::copy_nonoverlapping(data.ptr, buf.as_mut_ptr(), len);
+        }
+        len
+    }
+}
+
+impl Drop for PipeReader {
+    fn drop(&mut self) {
+        if let Err(err) = self.close() {
+            log::warn!(
+                "{}: PipeReader: Failed to close the channel on drop: {}",
+                env!("CARGO_PKG_NAME"),
+                err
+            );
+        }
+    }
+}
+
+impl AsyncRead for PipeReader {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        let mut state;
+        match self.state.lock() {
+            Ok(s) => state = s,
+            Err(err) => {
+                return Poll::Ready(Err(io::Error::new(
+                    io::ErrorKind::Other,
+                    format!(
+                        "{}: PipeReader: Failed to lock the channel state: {}",
+                        env!("CARGO_PKG_NAME"),
+                        err
+                    ),
+                )))
+            }
+        }
+
+        if state.closed {
+            return Poll::Ready(Ok(0));
+        }
+
+        return if state.done_cycle {
+            state.reader_waker = Some(cx.waker().clone());
+            Poll::Pending
+        } else {
+            if let Some(ref data) = state.data {
+                let copied_bytes_len = self.copy_data_into_buffer(data, buf);
+
+                state.data = None;
+                state.read = copied_bytes_len;
+                state.done_reading = true;
+                state.reader_waker = None;
+
+                self.wake_writer_half(&*state);
+
+                Poll::Ready(Ok(copied_bytes_len))
+            } else {
+                state.reader_waker = Some(cx.waker().clone());
+                Poll::Pending
+            }
+        };
+    }
+}

src/state.rs

@@ -0,0 +1,18 @@
+use std::task::Waker;
+
+pub(crate) struct State {
+    pub(crate) reader_waker: Option<Waker>,
+    pub(crate) writer_waker: Option<Waker>,
+    pub(crate) data: Option<Data>,
+    pub(crate) done_reading: bool,
+    pub(crate) read: usize,
+    pub(crate) done_cycle: bool,
+    pub(crate) closed: bool,
+}
+
+pub(crate) struct Data {
+    pub(crate) ptr: *const u8,
+    pub(crate) len: usize,
+}
+
+unsafe impl Send for Data {}