async_ringbuf/wrap/
cons.rs

Help
1use crate::{consumer::AsyncConsumer, rb::AsyncRbRef, wrap::AsyncCons};
2use core::{
3    pin::Pin,
4    task::{Context, Poll},
5};
6#[cfg(feature = "std")]
7use futures::io::AsyncRead;
8use futures::Stream;
9use ringbuf::{
10    traits::{
11        consumer::{Consumer, DelegateConsumer},
12        Observer,
13    },
14    wrap::Wrap,
15};
16#[cfg(feature = "std")]
17use std::io;
18
19impl<R: AsyncRbRef> DelegateConsumer for AsyncCons<R> {}
20
21impl<R: AsyncRbRef> AsyncConsumer for AsyncCons<R> {
22    fn register_waker(&self, waker: &core::task::Waker) {
23        self.rb().write.register(waker)
24    }
25
26    #[inline]
27    fn close(&mut self) {
28        drop(self.base.take());
29    }
30}
31
32impl<R: AsyncRbRef> Stream for AsyncCons<R> {
33    type Item = <R::Rb as Observer>::Item;
34
35    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
36        let mut waker_registered = false;
37        loop {
38            let closed = self.is_closed();
39            if let Some(item) = self.try_pop() {
40                break Poll::Ready(Some(item));
41            }
42            if closed {
43                break Poll::Ready(None);
44            }
45            if waker_registered {
46                break Poll::Pending;
47            }
48            self.register_waker(cx.waker());
49            waker_registered = true;
50        }
51    }
52}
53
54#[cfg(feature = "std")]
55impl<R: AsyncRbRef> AsyncRead for AsyncCons<R>
56where
57    Self: AsyncConsumer<Item = u8>,
58{
59    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<io::Result<usize>> {
60        let mut waker_registered = false;
61        loop {
62            let closed = self.is_closed();
63            let len = self.pop_slice(buf);
64            if len != 0 || closed {
65                break Poll::Ready(Ok(len));
66            }
67            if waker_registered {
68                break Poll::Pending;
69            }
70            self.register_waker(cx.waker());
71            waker_registered = true;
72        }
73    }
74}
async_ringbuf/wrap/cons.rs

async_ringbuf/wrap/
cons.rs
