async_ach_ring/
lib.rs

1#![no_std]
2
3use ach_ring as ach;
4use ach_util::Error;
5use async_ach_notify::Notify;
6use futures_util::StreamExt;
7
8pub struct Ring<T, const N: usize, const MP: usize, const MC: usize> {
9    buf: ach::Ring<T, N>,
10    consumer: Notify<MP>,
11    producer: Notify<MC>,
12}
13impl<T, const N: usize, const MP: usize, const MC: usize> Ring<T, N, MP, MC> {
14    pub const fn new() -> Self {
15        Self {
16            buf: ach::Ring::new(),
17            consumer: Notify::new(),
18            producer: Notify::new(),
19        }
20    }
21    pub fn len(&self) -> usize {
22        self.buf.len()
23    }
24    pub fn is_empty(&self) -> bool {
25        self.buf.is_empty()
26    }
27}
28impl<T, const N: usize, const MP: usize, const MC: usize> Default for Ring<T, N, MP, MC> {
29    fn default() -> Self {
30        Self::new()
31    }
32}
33impl<T: Unpin, const N: usize, const MP: usize, const MC: usize> Ring<T, N, MP, MC> {
34    /// Appends an element to the back of the Ring.
35    ///
36    /// Returns Err if the Ring is full or in critical section.
37    pub fn try_push(&self, val: T) -> Result<(), Error<T>> {
38        self.buf.push(val).inspect(|_x| {
39            self.producer.notify_one();
40        })
41    }
42    /// Appends an element to the back of the Ring.
43    pub async fn push(&self, mut val: T) {
44        let mut wait_c = self.consumer.listen();
45        while let Err(err) = self.try_push(val) {
46            val = err.input;
47            wait_c.next().await;
48        }
49    }
50
51    /// Removes the first element and returns it.
52    ///
53    /// Returns Err if the Ring is empty or in critical section.
54    pub fn try_pop(&self) -> Result<T, Error<()>> {
55        self.buf.pop().inspect(|_x| {
56            self.consumer.notify_one();
57        })
58    }
59    /// Removes the first element and returns it.
60    pub async fn pop(&self) -> T {
61        let mut wait_p = self.producer.listen();
62        loop {
63            if let Ok(v) = self.try_pop() {
64                break v;
65            } else {
66                wait_p.next().await;
67            }
68        }
69    }
70}
async_ach_ring/lib.rs

async_ach_ring/
lib.rs
