Skip to content

Commit

Permalink
return cloned value when warpped recv changed
Browse files Browse the repository at this point in the history
  • Loading branch information
liufuyang committed Jan 26, 2021
1 parent ff23c5a commit 3957c05
Showing 1 changed file with 4 additions and 6 deletions.
10 changes: 4 additions & 6 deletions tokio-stream/src/wrappers/watch.rs
Expand Up @@ -11,30 +11,28 @@ use std::task::{Context, Poll};
/// [`tokio::sync::watch::Receiver`]: struct@tokio::sync::watch::Receiver
/// [`Stream`]: trait@crate::Stream
pub struct WatchStream<T> {
inner: Pin<Box<dyn Stream<Item = ()>>>,
_marker: std::marker::PhantomData<T>,
inner: Pin<Box<dyn Stream<Item = T>>>,
}

impl<T: 'static> WatchStream<T> {
impl<T: 'static + Clone + Unpin> WatchStream<T> {
/// Create a new `WatchStream`.
pub fn new(mut rx: Receiver<T>) -> Self {
let stream = stream! {
loop {
match rx.changed().await {
Ok(item) => yield item,
Ok(_) => yield (*rx.borrow()).clone(),
Err(_) => break,
}
}
};
Self {
inner: Box::pin(stream),
_marker: std::marker::PhantomData,
}
}
}

impl<T> Stream for WatchStream<T> {
type Item = ();
type Item = T;

fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
Pin::new(&mut self.inner).poll_next(cx)
Expand Down

0 comments on commit 3957c05

Please sign in to comment.