Repository: tokio-rs/tokio
File: tokio-stream/src/wrappers/watch.rs (Rust)

Code before:
/// let mut rx = WatchStream::new(rx);
///
/// // existing rx output with "hello" is ignored here
///
/// tx.send("goodbye").unwrap();
/// assert_eq!(rx.next().await, Some("goodbye"));
/// # }
/// ```
///
/// Example with [`WatchStream<T>::new_on_changed`]:
///
/// ```
/// # #[tokio::main]
/// # async fn main() {
/// use futures::task::Poll;
/// use futures_test::task::noop_context;
/// use std::pin::Pin;
/// use tokio::sync::watch;
/// use tokio_stream::{Stream, StreamExt, wrappers::WatchStream};
///
/// let (tx, rx) = watch::channel("hello");
/// let mut rx = WatchStream::new_on_changed(rx);
///
/// // no output from rx is available at this point
/// let from_poll = Pin::new(&mut rx).poll_next(&mut noop_context());
/// assert_eq!(from_poll, Poll::Pending);
///
/// tx.send("goodbye").unwrap();
/// assert_eq!(rx.next().await, Some("goodbye"));
/// # }
/// ```
///
/// [`tokio::sync::watch::Receiver`]: struct@tokio::sync::watch::Receiver
/// [`Stream`]: trait@crate::Stream
#[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
pub struct WatchStream<T> {
    inner: ReusableBoxFuture<'static, (Result<(), RecvError>, Receiver<T>)>,
}

async fn make_future<T: Clone + Send + Sync>(
    mut rx: Receiver<T>,
) -> (Result<(), RecvError>, Receiver<T>) {
    let result = rx.changed().await;
    (result, rx)
}

impl<T: 'static + Clone + Send + Sync> WatchStream<T> {
    /// Create a new `WatchStream`.
    pub fn new(rx: Receiver<T>) -> Self {
        Self {
            inner: ReusableBoxFuture::new(async move { (Ok(()), rx) }),

Review comment (refactor):
A little ugly; I think it would be ideal if we could add a utility function to help check whether a stream is pending or ready.

Code after:
/// let mut rx = WatchStream::new(rx);
///
/// // existing rx output with "hello" is ignored here
///
/// tx.send("goodbye").unwrap();
/// assert_eq!(rx.next().await, Some("goodbye"));
/// # }
/// ```
///
/// Example with [`WatchStream<T>::from_changes`]:
///
/// ```
/// # #[tokio::main]
/// # async fn main() {
/// use futures::future::FutureExt;
/// use tokio::sync::watch;
/// use tokio_stream::{StreamExt, wrappers::WatchStream};
///
/// let (tx, rx) = watch::channel("hello");
/// let mut rx = WatchStream::from_changes(rx);
///
/// // no output from rx is available at this point - let's check this:
/// assert!(rx.next().now_or_never().is_none());
///
/// tx.send("goodbye").unwrap();
/// assert_eq!(rx.next().await, Some("goodbye"));
/// # }
/// ```
///
/// [`tokio::sync::watch::Receiver`]: struct@tokio::sync::watch::Receiver
/// [`Stream`]: trait@crate::Stream
#[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
pub struct WatchStream<T> {
    inner: ReusableBoxFuture<'static, (Result<(), RecvError>, Receiver<T>)>,
}

async fn make_future<T: Clone + Send + Sync>(
    mut rx: Receiver<T>,
) -> (Result<(), RecvError>, Receiver<T>) {
    let result = rx.changed().await;
    (result, rx)
}

impl<T: 'static + Clone + Send + Sync> WatchStream<T> {
    /// Create a new `WatchStream`.
    pub fn new(rx: Receiver<T>) -> Self {
        Self {
            inner: ReusableBoxFuture::new(async move { (Ok(()), rx) }),
        }
    }