signal: Replace ctrl_c with a CtrlC struct (#1273)

* Add a new `CtrlC` struct which will represent a stream of SIGINT
signals on Unix or the CTRL_C event on Windows
* `CtrlC` implements `Stream<Output = ()>` rather than `IoSteam` as
previously

tokio-signal/examples/ctrl-c.rs

@@ -1,7 +1,6 @@
 #![deny(warnings, rust_2018_idioms)]
 #![feature(async_await)]
 
-use futures_util::future;
 use futures_util::stream::StreamExt;
 
 /// how many signals to handle before exiting
@@ -12,11 +11,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // tokio_signal provides a convenience builder for Ctrl+C
     // this even works cross-platform: linux and windows!
     //
-    // `fn ctrl_c()` produces a `Future` of the actual stream-initialisation
+    // `CtrlC::new()` produces a `Future` of the actual stream-initialisation
     // so first we await until the signal is ready.
-    let endless_stream = tokio_signal::ctrl_c().await?;
+    let endless_stream = tokio_signal::CtrlC::new().await?;
     // don't keep going forever: convert the endless stream to a bounded one.
-    let limited_stream = endless_stream.take(STOP_AFTER);
+    let mut limited_stream = endless_stream.take(STOP_AFTER);
 
     // how many Ctrl+C have we received so far?
     let mut counter = 0;
@@ -28,29 +27,19 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         STOP_AFTER
     );
 
-    // Stream::for_each is a powerful primitive provided by the Futures crate.
-    // It turns a Stream into a Future that completes after all stream-items
-    // have been completed, or the first time the closure returns an error
-    let future = limited_stream
-        .map(|result| result.expect("failed to get event"))
-        .for_each(|()| {
-            // Note how we manipulate the counter without any fancy synchronisation.
-            // The borrowchecker realises there can't be any conflicts, so the closure
-            // can just capture it.
-            counter += 1;
-            println!(
-                "Ctrl+C received {} times! {} more before exit",
-                counter,
-                STOP_AFTER - counter
-            );
-
-            // return a result to continue handling the stream
-            future::ready(())
-        });
-
     // Up until now, we haven't really DONE anything, just prepared
-    // now it's time to actually the results!
-    future.await;
+    // our futures, now it's time to actually await the results!
+    while let Some(_) = limited_stream.next().await {
+        // Note how we manipulate the counter without any fancy synchronisation.
+        // The borrowchecker realises there can't be any conflicts, so the closure
+        // can just capture it.
+        counter += 1;
+        println!(
+            "Ctrl+C received {} times! {} more before exit",
+            counter,
+            STOP_AFTER - counter
+        );
+    }
 
     println!("Stream ended, quiting the program.");
     Ok(())

tokio-signal/examples/sighup-example.rs

@@ -6,15 +6,13 @@ use std::error::Error;
 // A trick to not fail build on non-unix platforms when using unix-specific features.
 #[cfg(unix)]
 mod platform {
-
-    use futures_util::future;
     use futures_util::stream::StreamExt;
     use std::error::Error;
     use tokio_signal::unix::{Signal, SIGHUP};
 
     pub async fn main() -> Result<(), Box<dyn Error>> {
         // on Unix, we can listen to whatever signal we want, in this case: SIGHUP
-        let stream = Signal::new(SIGHUP).await?;
+        let mut stream = Signal::new(SIGHUP).await?;
 
         println!("Waiting for SIGHUPS (Ctrl+C to quit)");
         println!(
@@ -23,22 +21,15 @@ mod platform {
              (i.e. this binary)"
         );
 
-        // for_each is a powerful primitive provided by the Futures crate
-        // it turns a Stream into a Future that completes after all stream-items
-        // have been completed.
-        let future = stream.for_each(|the_signal| {
+        // Up until now, we haven't really DONE anything, just prepared
+        // our futures, now it's time to actually await the results!
+        while let Some(the_signal) = stream.next().await {
             println!(
                 "*Got signal {:#x}* I should probably reload my config \
                  or something",
                 the_signal
             );
-
-            future::ready(())
-        });
-
-        // Up until now, we haven't really DONE anything, just prepared
-        // now it's time to actually the results!
-        future.await;
+        }
 
         Ok(())
     }

tokio-signal/src/ctrl_c.rs

@@ -0,0 +1,55 @@
+#[cfg(unix)]
+use crate::unix::Signal as Inner;
+#[cfg(windows)]
+use crate::windows::Event as Inner;
+use crate::IoFuture;
+use futures_core::stream::Stream;
+use futures_util::future::FutureExt;
+use futures_util::try_future::TryFutureExt;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio_reactor::Handle;
+
+/// Represents a stream which receives "ctrl-c" notifications sent to the process.
+///
+/// In general signals are handled very differently across Unix and Windows, but
+/// this is somewhat cross platform in terms of how it can be handled. A ctrl-c
+/// event to a console process can be represented as a stream for both Windows
+/// and Unix.
+///
+/// Note that there are a number of caveats listening for signals, and you may
+/// wish to read up on the documentation in the `unix` or `windows` module to
+/// take a peek.
+#[must_use = "streams do nothing unless polled"]
+#[derive(Debug)]
+pub struct CtrlC {
+    inner: Inner,
+}
+
+impl CtrlC {
+    /// Creates a new stream which receives "ctrl-c" notifications sent to the
+    /// process.
+    ///
+    /// This function binds to the default reactor.
+    pub fn new() -> IoFuture<Self> {
+        Self::with_handle(&Handle::default())
+    }
+
+    /// Creates a new stream which receives "ctrl-c" notifications sent to the
+    /// process.
+    ///
+    /// This function binds to reactor specified by `handle`.
+    pub fn with_handle(handle: &Handle) -> IoFuture<CtrlC> {
+        Inner::ctrl_c(handle).map_ok(|inner| Self { inner }).boxed()
+    }
+}
+
+impl Stream for CtrlC {
+    type Item = ();
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        Pin::new(&mut self.inner)
+            .poll_next(cx)
+            .map(|item| item.map(|_| ()))
+    }
+}

tokio-signal/src/lib.rs

@@ -1,5 +1,5 @@
 #![doc(html_root_url = "https://docs.rs/tokio-signal/0.2.8")]
-#![deny(missing_docs, rust_2018_idioms)]
+#![deny(missing_debug_implementations, missing_docs, rust_2018_idioms)]
 #![cfg_attr(test, deny(warnings))]
 #![cfg_attr(test, feature(async_await))]
 #![doc(test(no_crate_inject, attr(deny(rust_2018_idioms))))]
@@ -33,12 +33,10 @@
 //! async fn main() -> Result<(), Box<dyn std::error::Error>> {
 //!     // Create an infinite stream of "Ctrl+C" notifications. Each item received
 //!     // on this stream may represent multiple ctrl-c signals.
-//!     let ctrl_c = tokio_signal::ctrl_c().await?;
+//!     let ctrl_c = tokio_signal::CtrlC::new().await?;
 //!
 //!     // Process each ctrl-c as it comes in
-//!     let prog = ctrl_c.for_each(|event| {
-//!         event.expect("failed to get event");
-//!
+//!     let prog = ctrl_c.for_each(|_| {
 //!         println!("ctrl-c received!");
 //!         future::ready(())
 //!     });
@@ -62,12 +60,10 @@
 //! async fn main() -> Result<(), Box<dyn std::error::Error>> {
 //!     // Create an infinite stream of "Ctrl+C" notifications. Each item received
 //!     // on this stream may represent multiple ctrl-c signals.
-//!     let ctrl_c = tokio_signal::ctrl_c().await?;
+//!     let ctrl_c = tokio_signal::CtrlC::new().await?;
 //!
 //!     // Process each ctrl-c as it comes in
-//!     let prog = ctrl_c.for_each(|event| {
-//!         event.expect("failed to get event");
-//!
+//!     let prog = ctrl_c.for_each(|_| {
 //!         println!("ctrl-c received!");
 //!         future::ready(())
 //!     });
@@ -90,13 +86,10 @@ extern crate lazy_static;
 
 use futures_core::future::Future;
 use futures_core::stream::Stream;
-use futures_util::future::FutureExt;
-use futures_util::stream::StreamExt;
-use futures_util::try_future::TryFutureExt;
 use std::io;
 use std::pin::Pin;
-use tokio_reactor::Handle;
 
+mod ctrl_c;
 mod registry;
 
 mod os {
@@ -114,47 +107,4 @@ pub type IoFuture<T> = Pin<Box<dyn Future<Output = io::Result<T>> + Send>>;
 /// A stream whose item is `io::Result<T>`
 pub type IoStream<T> = Pin<Box<dyn Stream<Item = io::Result<T>> + Send>>;
 
-/// Creates a stream which receives "ctrl-c" notifications sent to a process.
-///
-/// In general signals are handled very differently across Unix and Windows, but
-/// this is somewhat cross platform in terms of how it can be handled. A ctrl-c
-/// event to a console process can be represented as a stream for both Windows
-/// and Unix.
-///
-/// This function binds to the default event loop. Note that
-/// there are a number of caveats listening for signals, and you may wish to
-/// read up on the documentation in the `unix` or `windows` module to take a
-/// peek.
-pub fn ctrl_c() -> IoFuture<IoStream<()>> {
-    ctrl_c_handle(&Handle::default())
-}
-
-/// Creates a stream which receives "ctrl-c" notifications sent to a process.
-///
-/// In general signals are handled very differently across Unix and Windows, but
-/// this is somewhat cross platform in terms of how it can be handled. A ctrl-c
-/// event to a console process can be represented as a stream for both Windows
-/// and Unix.
-///
-/// This function receives a `Handle` to an event loop and returns a future
-/// which when resolves yields a stream receiving all signal events. Note that
-/// there are a number of caveats listening for signals, and you may wish to
-/// read up on the documentation in the `unix` or `windows` module to take a
-/// peek.
-pub fn ctrl_c_handle(handle: &Handle) -> IoFuture<IoStream<()>> {
-    return ctrl_c_imp(handle);
-
-    #[cfg(unix)]
-    fn ctrl_c_imp(handle: &Handle) -> IoFuture<IoStream<()>> {
-        unix::Signal::with_handle(unix::libc::SIGINT, &handle)
-            .map_ok(|signal| -> IoStream<()> { signal.map(|_| Ok(())).boxed() })
-            .boxed()
-    }
-
-    #[cfg(windows)]
-    fn ctrl_c_imp(handle: &Handle) -> IoFuture<IoStream<()>> {
-        windows::Event::ctrl_c_handle(&handle)
-            .map_ok(|event| -> IoStream<()> { event.map(|_| Ok(())).boxed() })
-            .boxed()
-    }
-}
+pub use ctrl_c::CtrlC;

tokio-signal/src/registry.rs

@@ -252,7 +252,6 @@ mod tests {
         drop(second_rx);
 
         let (fire, wait) = oneshot::channel();
-        //let mut rt = Runtime::new().unwrap();
 
         tokio::spawn(async {
             wait.await.expect("wait failed");

tokio-signal/src/unix.rs

@@ -159,6 +159,7 @@ fn signal_enable(signal: c_int) -> io::Result<()> {
     }
 }
 
+#[derive(Debug)]
 struct Driver {
     wakeup: PollEvented<UnixStream>,
 }
@@ -249,6 +250,8 @@ impl Driver {
 /// If you've got any questions about this feel free to open an issue on the
 /// repo, though, as I'd love to chat about this! In other words, I'd love to
 /// alleviate some of these limitations if possible!
+#[must_use = "streams do nothing unless polled"]
+#[derive(Debug)]
 pub struct Signal {
     driver: Driver,
     signal: c_int,
@@ -326,6 +329,10 @@ impl Signal {
         })
         .boxed()
     }
+
+    pub(crate) fn ctrl_c(handle: &Handle) -> IoFuture<Signal> {
+        Self::with_handle(libc::SIGINT, handle)
+    }
 }
 
 impl Stream for Signal {
@@ -341,6 +348,11 @@ impl Stream for Signal {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use futures_util::future::FutureExt;
+    use futures_util::StreamExt;
+    use std::time::Duration;
+    use tokio_sync::oneshot;
+    use tokio_timer::Timeout;
 
     #[test]
     fn signal_enable_error_on_invalid_input() {
@@ -351,4 +363,27 @@ mod tests {
     fn signal_enable_error_on_forbidden_input() {
         signal_enable(signal_hook_registry::FORBIDDEN[0]).unwrap_err();
     }
+
+    fn with_timeout<F: Future>(future: F) -> impl Future<Output = F::Output> {
+        Timeout::new(future, Duration::from_secs(1)).map(|result| result.expect("timed out"))
+    }
+
+    #[tokio::test]
+    async fn ctrl_c() {
+        let ctrl_c = with_timeout(crate::CtrlC::new())
+            .await
+            .expect("failed to init ctrl_c");
+
+        let (fire, wait) = oneshot::channel();
+
+        // NB: simulate a signal coming in by exercising our signal handler
+        // to avoid complications with sending SIGINT to the test process
+        tokio::spawn(async {
+            wait.await.expect("wait failed");
+            action(globals(), libc::SIGINT);
+        });
+
+        let _ = fire.send(());
+        let _ = with_timeout(ctrl_c.into_future()).await;
+    }
 }

tokio-signal/src/windows.rs

@@ -89,6 +89,8 @@ static INIT: Once = ONCE_INIT;
 ///   received back-to-back, then the stream may only receive one item about the
 ///   two notifications.
 // FIXME: refactor and combine with unix::Signal
+#[must_use = "streams do nothing unless polled"]
+#[derive(Debug)]
 pub struct Event {
     rx: Receiver<()>,
 }
@@ -103,15 +105,7 @@ impl Event {
     ///
     /// This function will register a handler via `SetConsoleCtrlHandler` and
     /// deliver notifications to the returned stream.
-    pub fn ctrl_c() -> IoFuture<Event> {
-        Event::ctrl_c_handle(&Handle::default())
-    }
-
-    /// Creates a new stream listening for the `CTRL_C_EVENT` events.
-    ///
-    /// This function will register a handler via `SetConsoleCtrlHandler` and
-    /// deliver notifications to the returned stream.
-    pub fn ctrl_c_handle(handle: &Handle) -> IoFuture<Event> {
+    pub(crate) fn ctrl_c(handle: &Handle) -> IoFuture<Event> {
         Event::new(CTRL_C_EVENT, handle)
     }
 
@@ -224,7 +218,7 @@ mod tests {
         // first event loop cannot go away
         let mut rt = current_thread::Runtime::new().unwrap();
         let event_ctrl_c = rt
-            .block_on(with_timeout(Event::ctrl_c()))
+            .block_on(with_timeout(crate::CtrlC::new()))
             .expect("failed to run future");
 
         // Windows doesn't have a good programmatic way of sending events
@@ -234,7 +228,7 @@ mod tests {
             super::handler(CTRL_C_EVENT);
         }
 
-        rt.block_on(with_timeout(event_ctrl_c.into_future()));
+        let _ = rt.block_on(with_timeout(event_ctrl_c.into_future()));
 
         let event_ctrl_break = rt
             .block_on(with_timeout(Event::ctrl_break()))
@@ -244,6 +238,6 @@ mod tests {
             super::handler(CTRL_BREAK_EVENT);
         }
 
-        rt.block_on(with_timeout(event_ctrl_break.into_future()));
+        let _ = rt.block_on(with_timeout(event_ctrl_break.into_future()));
     }
 }

tokio-signal/tests/drop_multi_loop.rs

@@ -29,5 +29,5 @@ fn dropping_loops_does_not_cause_starvation() {
 
     send_signal(TEST_SIGNAL);
 
-    run_with_timeout(&mut rt, signal.into_future());
+    let _ = run_with_timeout(&mut rt, signal.into_future());
 }

tokio-signal/tests/drop_then_get_a_signal.rs

@@ -19,5 +19,5 @@ async fn drop_then_get_a_signal() {
         .await
         .expect("failed to create second signal");
 
-    with_timeout(signal.into_future()).await;
+    let _ = with_timeout(signal.into_future()).await;
 }

tokio-signal/tests/dropping_does_not_deregister_other_instances.rs

@@ -26,5 +26,5 @@ async fn dropping_signal_does_not_deregister_any_other_instances() {
     drop(second_duplicate_signal);
 
     send_signal(TEST_SIGNAL);
-    with_timeout(signal.into_future()).await;
+    let _ = with_timeout(signal.into_future()).await;
 }

tokio-signal/tests/multi_loop.rs

@@ -22,7 +22,7 @@ fn multi_loop() {
                     let mut rt = CurrentThreadRuntime::new().unwrap();
                     let signal = run_with_timeout(&mut rt, Signal::new(libc::SIGHUP)).unwrap();
                     sender.send(()).unwrap();
-                    run_with_timeout(&mut rt, signal.into_future());
+                    let _ = run_with_timeout(&mut rt, signal.into_future());
                 })
             })
             .collect();

tokio-signal/tests/notify_both.rs

@@ -18,5 +18,5 @@ async fn notify_both() {
         .expect("failed to create signal2");
 
     send_signal(libc::SIGUSR2);
-    with_timeout(future::join(signal1.into_future(), signal2.into_future())).await;
+    let _ = with_timeout(future::join(signal1.into_future(), signal2.into_future())).await;
 }

tokio-signal/tests/simple.rs

@@ -15,5 +15,5 @@ async fn simple() {
 
     send_signal(libc::SIGUSR1);
 
-    with_timeout(signal.into_future()).await;
+    let _ = with_timeout(signal.into_future()).await;
 }