process: Misc polish (#1400)

* Denied all warnings in tests, and denied rust_2018_idioms violations
* Bumped the crate version and set publish = false
* Pruned dependencies:
 - Only pull in tokio-sync on windows where it is used
 - Removed unused dev-dependencies
* Switch to Async{Read, Write} traits from tokio-io rather than
futures-io
* Use #[tokio::test] where possible
* Removed deprecated items
* Fix all doc examples

tokio-process/tests/issue_42.rs

@@ -1,18 +1,15 @@
 #![cfg(unix)]
-
-extern crate tokio_process;
+#![deny(warnings, rust_2018_idioms)]
 
 use futures_util::future::FutureExt;
 use futures_util::stream::FuturesOrdered;
 use futures_util::stream::StreamExt;
-use std::future::Future;
-use std::io;
-use std::pin::Pin;
-use std::process::{Command, ExitStatus, Stdio};
+use std::process::{Command, Stdio};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::thread;
 use std::time::Duration;
+use tokio::runtime::current_thread;
 use tokio_process::CommandExt;
 
 mod support;
@@ -22,8 +19,7 @@ fn run_test() {
     let finished_clone = finished.clone();
 
     thread::spawn(move || {
-        let mut futures: FuturesOrdered<Pin<Box<dyn Future<Output = io::Result<ExitStatus>>>>> =
-            FuturesOrdered::new();
+        let mut futures = FuturesOrdered::new();
         for i in 0..2 {
             futures.push(
                 Command::new("echo")
@@ -36,7 +32,10 @@ fn run_test() {
                     .boxed(),
             )
         }
-        support::run_with_timeout(futures.collect::<Vec<io::Result<ExitStatus>>>());
+
+        let mut rt = current_thread::Runtime::new().expect("failed to get runtime");
+        rt.block_on(support::with_timeout(futures.collect::<Vec<_>>()));
+        drop(rt);
 
         finished_clone.store(true, Ordering::SeqCst);
     });

tokio-process/tests/smoke.rs

@@ -1,11 +1,12 @@
-extern crate tokio_process;
+#![deny(warnings, rust_2018_idioms)]
+#![feature(async_await)]
 
 use tokio_process::CommandExt;
 
 mod support;
 
-#[test]
-fn simple() {
+#[tokio::test]
+async fn simple() {
     let mut cmd = support::cmd("exit");
     cmd.arg("2");
 
@@ -14,7 +15,9 @@ fn simple() {
     let id = child.id();
     assert!(id > 0);
 
-    let status = support::run_with_timeout(&mut child).expect("failed to run future");
+    let status = support::with_timeout(&mut child)
+        .await
+        .expect("failed to run future");
     assert_eq!(status.code(), Some(2));
 
     assert_eq!(child.id(), id);

tokio-process/tests/stdio.rs

@@ -1,21 +1,17 @@
+#![deny(warnings, rust_2018_idioms)]
 #![feature(async_await)]
 
 #[macro_use]
 extern crate log;
-extern crate tokio_io;
-extern crate tokio_process;
 
-use std::future::Future;
 use std::io;
-use std::pin::Pin;
 use std::process::{Command, ExitStatus, Stdio};
 
 use futures_util::future;
 use futures_util::future::FutureExt;
-use futures_util::io::AsyncBufReadExt;
-use futures_util::io::AsyncWriteExt;
-use futures_util::io::BufReader;
-use futures_util::stream::{self, StreamExt};
+use futures_util::stream::StreamExt;
+use tokio::codec::{FramedRead, LinesCodec};
+use tokio::io::AsyncWriteExt;
 use tokio_process::{Child, CommandExt};
 
 mod support;
@@ -26,66 +22,66 @@ fn cat() -> Command {
     cmd
 }
 
-fn feed_cat(mut cat: Child, n: usize) -> Pin<Box<dyn Future<Output = io::Result<ExitStatus>>>> {
-    let stdin = cat.stdin().take().unwrap();
+async fn feed_cat(mut cat: Child, n: usize) -> io::Result<ExitStatus> {
+    let mut stdin = cat.stdin().take().unwrap();
     let stdout = cat.stdout().take().unwrap();
 
-    debug!("starting to feed");
     // Produce n lines on the child's stdout.
-    let numbers = stream::iter(0..n);
-    let write = numbers
-        .fold(stdin, move |mut stdin, i| {
-            let fut = async move {
-                debug!("sending line {} to child", i);
-                let bytes = format!("line {}\n", i).into_bytes();
-                AsyncWriteExt::write_all(&mut stdin, &bytes).await.unwrap();
-                stdin
-            };
-            fut
-        })
-        .map(|_| ());
+    let write = async {
+        debug!("starting to feed");
 
-    // Try to read `n + 1` lines, ensuring the last one is empty
-    // (i.e. EOF is reached after `n` lines.
-    let reader = BufReader::new(stdout);
-    let expected_numbers = stream::iter(0..=n);
-    let read = expected_numbers.fold((reader, 0), move |(mut reader, i), _| {
-        let fut = async move {
-            let done = i >= n;
+        for i in 0..n {
+            debug!("sending line {} to child", i);
+            let bytes = format!("line {}\n", i).into_bytes();
+            stdin.write_all(&bytes).await.unwrap();
+        }
+
+        drop(stdin);
+    };
+
+    let read = async {
+        let mut reader = FramedRead::new(stdout, LinesCodec::new());
+        let mut num_lines = 0;
+
+        // Try to read `n + 1` lines, ensuring the last one is empty
+        // (i.e. EOF is reached after `n` lines.
+        loop {
             debug!("starting read from child");
-            let mut vec = Vec::new();
-            AsyncBufReadExt::read_until(&mut reader, b'\n', &mut vec)
+
+            let data = reader
+                .next()
                 .await
-                .unwrap();
+                .unwrap_or_else(|| Ok(String::new()))
+                .expect("failed to read line");
+
+            let num_read = data.len();
+            let done = num_lines >= n;
+
             debug!(
                 "read line {} from child ({} bytes, done: {})",
-                i,
-                vec.len(),
-                done
+                num_lines, num_read, done
             );
-            match (done, vec.len()) {
-                (false, 0) => {
-                    panic!("broken pipe");
-                }
-                (true, n) if n != 0 => {
-                    panic!("extraneous data");
-                }
+
+            match (done, num_read) {
+                (false, 0) => panic!("broken pipe"),
+                (true, n) if n != 0 => panic!("extraneous data"),
                 _ => {
-                    let s = std::str::from_utf8(&vec).unwrap();
-                    let expected = format!("line {}\n", i);
-                    if done || s == expected {
-                        (reader, i + 1)
-                    } else {
-                        panic!("unexpected data");
-                    }
+                    let expected = format!("line {}", num_lines);
+                    assert_eq!(expected, data);
                 }
+            };
+
+            num_lines += 1;
+            if num_lines >= n {
+                break;
             }
-        };
-        fut
-    });
+        }
+    };
 
     // Compose reading and writing concurrently.
-    future::join(write, read).then(|_| cat).boxed()
+    future::join3(write, read, cat)
+        .map(|(_, _, status)| status)
+        .await
 }
 
 /// Check for the following properties when feeding stdin and
@@ -99,42 +95,42 @@ fn feed_cat(mut cat: Child, n: usize) -> Pin<Box<dyn Future<Output = io::Result<
 /// - We read the same lines from the child that we fed it.
 ///
 /// - The child does produce EOF on stdout after the last line.
-#[test]
-fn feed_a_lot() {
+#[tokio::test]
+async fn feed_a_lot() {
     let child = cat().spawn_async().unwrap();
-    let status = support::run_with_timeout(feed_cat(child, 10000)).unwrap();
+    let status = support::with_timeout(feed_cat(child, 10000)).await.unwrap();
     assert_eq!(status.code(), Some(0));
 }
 
-#[test]
-fn wait_with_output_captures() {
+#[tokio::test]
+async fn wait_with_output_captures() {
     let mut child = cat().spawn_async().unwrap();
     let mut stdin = child.stdin().take().unwrap();
 
     let write_bytes = b"1234";
 
     let future = async {
-        AsyncWriteExt::write_all(&mut stdin, write_bytes).await?;
+        stdin.write_all(write_bytes).await?;
         drop(stdin);
         let out = child.wait_with_output();
         out.await
     };
 
-    let ret = support::run_with_timeout(future).unwrap();
-    let output = ret;
+    let output = support::with_timeout(future).await.unwrap();
 
     assert!(output.status.success());
     assert_eq!(output.stdout, write_bytes);
     assert_eq!(output.stderr.len(), 0);
 }
 
-#[test]
-fn status_closes_any_pipes() {
+#[tokio::test]
+async fn status_closes_any_pipes() {
     // Cat will open a pipe between the parent and child.
     // If `status_async` doesn't ensure the handles are closed,
     // we would end up blocking forever (and time out).
     let child = cat().status_async().expect("failed to spawn child");
 
-    support::run_with_timeout(child)
+    support::with_timeout(child)
+        .await
         .expect("time out exceeded! did we get stuck waiting on the child?");
 }

tokio-process/tests/support/mod.rs

@@ -1,4 +1,4 @@
-extern crate tokio;
+#![deny(warnings, rust_2018_idioms)]
 
 use futures_util::future;
 use futures_util::future::FutureExt;
@@ -8,8 +8,6 @@ use std::process::Command;
 use std::time::Duration;
 use tokio::timer::Timeout;
 
-pub use self::tokio::runtime::current_thread::Runtime as CurrentThreadRuntime;
-
 #[allow(dead_code)]
 pub fn cmd(s: &str) -> Command {
     let mut me = env::current_exe().unwrap();
@@ -29,14 +27,3 @@ pub fn with_timeout<F: Future>(future: F) -> impl Future<Output = F::Output> {
         future::ready(r.unwrap())
     })
 }
-
-pub fn run_with_timeout<F>(future: F) -> F::Output
-where
-    F: Future,
-{
-    // NB: Timeout requires a timer registration which is provided by
-    // tokio's `current_thread::Runtime`, but isn't available by just using
-    // tokio's default CurrentThread executor which powers `current_thread::block_on_all`.
-    let mut rt = CurrentThreadRuntime::new().expect("failed to get runtime");
-    rt.block_on(with_timeout(future))
-}