signal: add Signal wrappers to tokio-stream (#3510)

tokio-stream/Cargo.toml

@@ -26,15 +26,16 @@ net = ["tokio/net"]
 io-util = ["tokio/io-util"]
 fs = ["tokio/fs"]
 sync = ["tokio/sync", "tokio-util"]
+signal = ["tokio/signal"]
 
 [dependencies]
 futures-core = { version = "0.3.0" }
 pin-project-lite = "0.2.0"
-tokio = { version = "1.0", features = ["sync"] }
+tokio = { version = "1.2.0", features = ["sync"] }
 tokio-util = { version = "0.6.3", optional = true }
 
 [dev-dependencies]
-tokio = { version = "1.0", features = ["full", "test-util"] }
+tokio = { version = "1.2.0", features = ["full", "test-util"] }
 async-stream = "0.3"
 tokio-test = { path = "../tokio-test" }
 futures = { version = "0.3", default-features = false }

tokio-stream/src/macros.rs

@@ -48,6 +48,16 @@ macro_rules! cfg_sync {
     }
 }
 
+macro_rules! cfg_signal {
+    ($($item:item)*) => {
+        $(
+            #[cfg(feature = "signal")]
+            #[cfg_attr(docsrs, doc(cfg(feature = "signal")))]
+            $item
+        )*
+    }
+}
+
 macro_rules! ready {
     ($e:expr $(,)?) => {
         match $e {

tokio-stream/src/wrappers.rs

@@ -1,4 +1,13 @@
 //! Wrappers for Tokio types that implement `Stream`.
+//!
+#![cfg_attr(
+    unix,
+    doc = "You are viewing documentation built under unix. To view windows-specific wrappers, change to the `x86_64-pc-windows-msvc` platform."
+)]
+#![cfg_attr(
+    windows,
+    doc = "You are viewing documentation built under windows. To view unix-specific wrappers, change to the `x86_64-unknown-linux-gnu` platform."
+)]
 
 /// Error types for the wrappers.
 pub mod errors {
@@ -21,6 +30,18 @@ cfg_sync! {
     pub use watch::WatchStream;
 }
 
+cfg_signal! {
+    #[cfg(unix)]
+    mod signal_unix;
+    #[cfg(unix)]
+    pub use signal_unix::SignalStream;
+
+    #[cfg(windows)]
+    mod signal_windows;
+    #[cfg(windows)]
+    pub use signal_windows::{CtrlCStream, CtrlBreakStream};
+}
+
 cfg_time! {
     mod interval;
     pub use interval::IntervalStream;

tokio-stream/src/wrappers/signal_unix.rs

@@ -0,0 +1,46 @@
+use crate::Stream;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::signal::unix::Signal;
+
+/// A wrapper around [`Signal`] that implements [`Stream`].
+///
+/// [`Signal`]: struct@tokio::signal::unix::Signal
+/// [`Stream`]: trait@crate::Stream
+#[derive(Debug)]
+#[cfg_attr(docsrs, doc(cfg(all(unix, feature = "signal"))))]
+pub struct SignalStream {
+    inner: Signal,
+}
+
+impl SignalStream {
+    /// Create a new `SignalStream`.
+    pub fn new(interval: Signal) -> Self {
+        Self { inner: interval }
+    }
+
+    /// Get back the inner `Signal`.
+    pub fn into_inner(self) -> Signal {
+        self.inner
+    }
+}
+
+impl Stream for SignalStream {
+    type Item = ();
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<()>> {
+        self.inner.poll_recv(cx)
+    }
+}
+
+impl AsRef<Signal> for SignalStream {
+    fn as_ref(&self) -> &Signal {
+        &self.inner
+    }
+}
+
+impl AsMut<Signal> for SignalStream {
+    fn as_mut(&mut self) -> &mut Signal {
+        &mut self.inner
+    }
+}

tokio-stream/src/wrappers/signal_windows.rs

@@ -0,0 +1,88 @@
+use crate::Stream;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::signal::windows::{CtrlBreak, CtrlC};
+
+/// A wrapper around [`CtrlC`] that implements [`Stream`].
+///
+/// [`CtrlC`]: struct@tokio::signal::windows::CtrlC
+/// [`Stream`]: trait@crate::Stream
+#[derive(Debug)]
+#[cfg_attr(docsrs, doc(cfg(all(windows, feature = "signal"))))]
+pub struct CtrlCStream {
+    inner: CtrlC,
+}
+
+impl CtrlCStream {
+    /// Create a new `CtrlCStream`.
+    pub fn new(interval: CtrlC) -> Self {
+        Self { inner: interval }
+    }
+
+    /// Get back the inner `CtrlC`.
+    pub fn into_inner(self) -> CtrlC {
+        self.inner
+    }
+}
+
+impl Stream for CtrlCStream {
+    type Item = ();
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<()>> {
+        self.inner.poll_recv(cx)
+    }
+}
+
+impl AsRef<CtrlC> for CtrlCStream {
+    fn as_ref(&self) -> &CtrlC {
+        &self.inner
+    }
+}
+
+impl AsMut<CtrlC> for CtrlCStream {
+    fn as_mut(&mut self) -> &mut CtrlC {
+        &mut self.inner
+    }
+}
+
+/// A wrapper around [`CtrlBreak`] that implements [`Stream`].
+///
+/// [`CtrlBreak`]: struct@tokio::signal::windows::CtrlBreak
+/// [`Stream`]: trait@crate::Stream
+#[derive(Debug)]
+#[cfg_attr(docsrs, doc(cfg(all(windows, feature = "signal"))))]
+pub struct CtrlBreakStream {
+    inner: CtrlBreak,
+}
+
+impl CtrlBreakStream {
+    /// Create a new `CtrlBreakStream`.
+    pub fn new(interval: CtrlBreak) -> Self {
+        Self { inner: interval }
+    }
+
+    /// Get back the inner `CtrlBreak`.
+    pub fn into_inner(self) -> CtrlBreak {
+        self.inner
+    }
+}
+
+impl Stream for CtrlBreakStream {
+    type Item = ();
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<()>> {
+        self.inner.poll_recv(cx)
+    }
+}
+
+impl AsRef<CtrlBreak> for CtrlBreakStream {
+    fn as_ref(&self) -> &CtrlBreak {
+        &self.inner
+    }
+}
+
+impl AsMut<CtrlBreak> for CtrlBreakStream {
+    fn as_mut(&mut self) -> &mut CtrlBreak {
+        &mut self.inner
+    }
+}