max/embassy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Return a new future each time recv is called

Browse Source
This commit is contained in:
huntc 2021-07-11 10:54:35 +10:00
parent f159beec1c
commit dcd0c38109
1 changed files with 12 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
embassy/src/util/mpsc.rs
View File

@ -141,7 +141,18 @@ where
///
/// [`close`]: Self::close
pub async fn recv(&mut self) -> Option<T> {
self.await
futures::future::poll_fn(|cx| self.recv_poll(cx)).await
}
fn recv_poll(self: &mut Self, cx: &mut Context<'_>) -> Poll<Option<T>> {
match self.try_recv() {
Ok(v) => Poll::Ready(Some(v)),
Err(TryRecvError::Closed) => Poll::Ready(None),
Err(TryRecvError::Empty) => {
self.channel.get().set_receiver_waker(cx.waker().clone());
Poll::Pending
}
}
}
/// Attempts to immediately receive a message on this `Receiver`
@ -167,24 +178,6 @@ where
}
}
impl<'ch, M, T, const N: usize> Future for Receiver<'ch, M, T, N>
where
M: Mutex<Data = ()>,
{
type Output = Option<T>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.try_recv() {
Ok(v) => Poll::Ready(Some(v)),
Err(TryRecvError::Closed) => Poll::Ready(None),
Err(TryRecvError::Empty) => {
self.channel.get().set_receiver_waker(cx.waker().clone());
Poll::Pending
}
}
}
}
impl<'ch, M, T, const N: usize> Drop for Receiver<'ch, M, T, N>
where
M: Mutex<Data = ()>,