nexus_acto_rs/util/
queue.rs

1use std::cmp::Ordering;
2use std::fmt::Debug;
3use std::ops::Add;
4
5use async_trait::async_trait;
6use thiserror::Error;
7
8use crate::util::element::Element;
9
10mod mpsc_bounded_channel_queue;
11mod mpsc_bounded_channel_queue_test;
12mod mpsc_unbounded_channel_queue;
13mod mpsc_unbounded_channel_queue_test;
14mod priority_queue;
15mod priority_queue_test;
16mod ring_queue;
17mod ring_queue_test;
18
19pub(crate) use {self::mpsc_unbounded_channel_queue::*, self::priority_queue::*, self::ring_queue::*};
20
21/// An error that occurs when a queue operation fails.<br/>
22/// キューの操作に失敗した場合に発生するエラー。
23#[derive(Error, Debug, PartialEq)]
24pub enum QueueError<E> {
25  #[error("Failed to offer an element: {0:?}")]
26  OfferError(E),
27  #[error("Failed to pool an element")]
28  PoolError,
29  #[error("Failed to peek an element")]
30  PeekError,
31  #[error("Failed to contains an element")]
32  ContainsError,
33  #[error("Failed to interrupt")]
34  InterruptedError,
35  #[error("Failed to timeout")]
36  TimeoutError,
37}
38
39/// The size of the queue.<br/>
40/// キューのサイズ。
41#[derive(Debug, Clone)]
42pub enum QueueSize {
43  /// The queue has no capacity limit.<br/>
44  /// キューに容量制限がない。
45  Limitless,
46  /// The queue has a capacity limit.<br/>
47  /// キューに容量制限がある。
48  Limited(usize),
49}
50
51impl QueueSize {
52  fn increment(&mut self) {
53    match self {
54      QueueSize::Limited(c) => {
55        *c += 1;
56      }
57      _ => {}
58    }
59  }
60
61  fn decrement(&mut self) {
62    match self {
63      QueueSize::Limited(c) => {
64        *c -= 1;
65      }
66      _ => {}
67    }
68  }
69
70  /// Returns whether the queue has no capacity limit.<br/>
71  /// キューに容量制限がないかどうかを返します。
72  ///
73  /// # Return Value / 戻り値
74  /// - `true` - If the queue has no capacity limit. / キューに容量制限がない場合。
75  /// - `false` - If the queue has a capacity limit. / キューに容量制限がある場合。
76  pub fn is_limitless(&self) -> bool {
77    match self {
78      QueueSize::Limitless => true,
79      _ => false,
80    }
81  }
82
83  /// Converts to an option type.<br/>
84  /// オプション型に変換します。
85  ///
86  /// # Return Value / 戻り値
87  /// - `None` - If the queue has no capacity limit. / キューに容量制限がない場合。
88  /// - `Some(num)` - If the queue has a capacity limit. / キューに容量制限がある場合。
89  pub fn to_option(&self) -> Option<usize> {
90    match self {
91      QueueSize::Limitless => None,
92      QueueSize::Limited(c) => Some(*c),
93    }
94  }
95
96  /// Converts to a usize type.<br/>
97  /// usize型に変換します。
98  ///
99  /// # Return Value / 戻り値
100  /// - `usize::MAX` - If the queue has no capacity limit. / キューに容量制限がない場合。
101  /// - `num` - If the queue has a capacity limit. / キューに容量制限がある場合。
102  pub fn to_usize(&self) -> usize {
103    match self {
104      QueueSize::Limitless => usize::MAX,
105      QueueSize::Limited(c) => *c,
106    }
107  }
108}
109
110impl Add for QueueSize {
111  type Output = QueueSize;
112
113  fn add(self, other: QueueSize) -> QueueSize {
114    match (self, other) {
115      (QueueSize::Limitless, _) | (_, QueueSize::Limitless) => QueueSize::Limitless,
116      (QueueSize::Limited(a), QueueSize::Limited(b)) => QueueSize::Limited(a + b),
117    }
118  }
119}
120
121impl PartialEq<Self> for QueueSize {
122  fn eq(&self, other: &Self) -> bool {
123    match (self, other) {
124      (QueueSize::Limitless, QueueSize::Limitless) => true,
125      (QueueSize::Limited(l), QueueSize::Limited(r)) => l == r,
126      _ => false,
127    }
128  }
129}
130
131impl PartialOrd<QueueSize> for QueueSize {
132  fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
133    match (self, other) {
134      (QueueSize::Limitless, QueueSize::Limitless) => Some(Ordering::Equal),
135      (QueueSize::Limitless, _) => Some(Ordering::Greater),
136      (_, QueueSize::Limitless) => Some(Ordering::Less),
137      (QueueSize::Limited(l), QueueSize::Limited(r)) => l.partial_cmp(r),
138    }
139  }
140}
141
142/// A trait that defines the behavior of a queue.<br/>
143/// キューの振る舞いを定義するトレイト。
144#[async_trait]
145pub trait QueueBase<E: Element>: Debug + Send + Sync {
146  /// Returns whether this queue is empty.<br/>
147  /// このキューが空かどうかを返します。
148  ///
149  /// # Return Value / 戻り値
150  /// - `true` - If the queue is empty. / キューが空の場合。
151  /// - `false` - If the queue is not empty. / キューが空でない場合。
152  async fn is_empty(&self) -> bool {
153    self.len().await == QueueSize::Limited(0)
154  }
155
156  /// Returns whether this queue is non-empty.<br/>
157  /// このキューが空でないかどうかを返します。
158  ///
159  /// # Return Value / 戻り値
160  /// - `true` - If the queue is not empty. / キューが空でない場合。
161  /// - `false` - If the queue is empty. / キューが空の場合。
162  async fn non_empty(&self) -> bool {
163    !self.is_empty().await
164  }
165
166  /// Returns whether the queue size has reached its capacity.<br/>
167  /// このキューのサイズが容量まで到達したかどうかを返します。
168  ///
169  /// # Return Value / 戻り値
170  /// - `true` - If the queue size has reached its capacity. / キューのサイズが容量まで到達した場合。
171  /// - `false` - If the queue size has not reached its capacity. / キューのサイズが容量まで到達してない場合。
172  async fn is_full(&self) -> bool {
173    self.capacity().await == self.len().await
174  }
175
176  /// Returns whether the queue size has not reached its capacity.<br/>
177  /// このキューのサイズが容量まで到達してないかどうかを返します。
178  ///
179  /// # Return Value / 戻り値
180  /// - `true` - If the queue size has not reached its capacity. / キューのサイズが容量まで到達してない場合。
181  /// - `false` - If the queue size has reached its capacity. / キューのサイズが容量まで到達した場合。
182  async fn non_full(&self) -> bool {
183    !self.is_full().await
184  }
185
186  /// Returns the length of this queue.<br/>
187  /// このキューの長さを返します。
188  ///
189  /// # Return Value / 戻り値
190  /// - `QueueSize::Limitless` - If the queue has no capacity limit. / キューに容量制限がない場合。
191  /// - `QueueSize::Limited(num)` - If the queue has a capacity limit. / キューに容量制限がある場合。
192  async fn len(&self) -> QueueSize;
193
194  /// Returns the capacity of this queue.<br/>
195  /// このキューの最大容量を返します。
196  ///
197  /// # Return Value / 戻り値
198  /// - `QueueSize::Limitless` - If the queue has no capacity limit. / キューに容量制限がない場合。
199  /// - `QueueSize::Limited(num)` - If the queue has a capacity limit. / キューに容量制限がある場合。
200  async fn capacity(&self) -> QueueSize;
201}
202
203#[async_trait]
204pub trait QueueWriteFactory<E: Element>: QueueBase<E> {
205  type Writer: QueueWriter<E>;
206  fn writer(&self) -> Self::Writer;
207}
208
209#[async_trait::async_trait]
210pub trait QueueWriter<E: Element>: QueueBase<E> {
211  /// The specified element will be inserted into this queue,
212  /// if the queue can be executed immediately without violating the capacity limit.<br/>
213  /// 容量制限に違反せずにすぐ実行できる場合は、指定された要素をこのキューに挿入します。
214  ///
215  /// # Arguments / 引数
216  /// - `element` - The element to be inserted. / 挿入する要素。
217  ///
218  /// # Return Value / 戻り値
219  /// - `Ok(())` - If the element is inserted successfully. / 要素が正常に挿入された場合。
220  /// - `Err(QueueError::OfferError(element))` - If the element cannot be inserted. / 要素を挿入できなかった場合。
221  async fn offer(&mut self, element: E) -> Result<(), QueueError<E>>;
222
223  /// The specified elements will be inserted into this queue,
224  /// if the queue can be executed immediately without violating the capacity limit.<br/>
225  /// 容量制限に違反せずにすぐ実行できる場合は、指定された複数の要素をこのキューに挿入します。
226  ///
227  /// # Arguments / 引数
228  /// - `elements` - The elements to be inserted. / 挿入する要素。
229  ///
230  /// # Return Value / 戻り値
231  /// - `Ok(())` - If the elements are inserted successfully. / 要素が正常に挿入された場合。
232  /// - `Err(QueueError::OfferError(element))` - If the elements cannot be inserted. / 要素を挿入できなかった場合。
233  async fn offer_all(&mut self, elements: Vec<E>) -> Result<(), QueueError<E>> {
234    for e in elements {
235      self.offer(e).await?;
236    }
237    Ok(())
238  }
239}
240
241#[async_trait::async_trait]
242pub trait QueueReadFactory<E: Element>: QueueBase<E> {
243  type Reader: QueueReader<E>;
244  fn reader(&self) -> Self::Reader;
245}
246
247#[async_trait]
248pub trait QueueReader<E: Element>: QueueBase<E> {
249  /// Retrieves and deletes the head of the queue. Returns None if the queue is empty.<br/>
250  /// キューの先頭を取得および削除します。キューが空の場合は None を返します。
251  ///
252  /// # Return Value / 戻り値
253  /// - `Ok(Some(element))` - If the element is retrieved successfully. / 要素が正常に取得された場合。
254  /// - `Ok(None)` - If the queue is empty. / キューが空の場合。
255  async fn poll(&mut self) -> Result<Option<E>, QueueError<E>>;
256
257  async fn clean_up(&mut self);
258}
259
260/// A trait that defines the behavior of a queue that can be peeked.<br/>
261/// Peekができるキューの振る舞いを定義するトレイト。
262#[async_trait]
263pub trait HasPeekBehavior<E: Element>: QueueReader<E> {
264  /// Gets the head of the queue, but does not delete it. Returns None if the queue is empty.<br/>
265  /// キューの先頭を取得しますが、削除しません。キューが空の場合は None を返します。
266  ///
267  /// # Return Value / 戻り値
268  /// - `Ok(Some(element))` - If the element is retrieved successfully. / 要素が正常に取得された場合。
269  /// - `Ok(None)` - If the queue is empty. / キューが空の場合。
270  async fn peek(&self) -> Result<Option<E>, QueueError<E>>;
271}
272
273/// A trait that defines the behavior of a queue that can be checked for contains.<br/>
274/// Containsができるキューの振る舞いを定義するトレイト。
275#[async_trait]
276pub trait HasContainsBehavior<E: Element>: QueueReader<E> {
277  /// Returns whether the specified element is contained in this queue.<br/>
278  /// 指定された要素がこのキューに含まれているかどうかを返します。
279  ///
280  /// # Arguments / 引数
281  /// - `element` - The element to be checked. / チェックする要素。
282  ///
283  /// # Return Value / 戻り値
284  /// - `true` - If the element is contained in this queue. / 要素がこのキューに含まれている場合。
285  /// - `false` - If the element is not contained in this queue. / 要素がこのキューに含まれていない場合。
286  async fn contains(&self, element: &E) -> bool;
287}
288
289/// A trait that defines the behavior of a blocking queue.<br/>
290/// ブロッキングキューの振る舞いを定義するトレイト。
291#[async_trait]
292pub trait BlockingQueueBase<E: Element>: QueueBase<E> + Send {
293  /// Returns the number of elements that can be inserted into this queue without blocking.<br/>
294  /// ブロックせずにこのキューに挿入できる要素数を返します。
295  ///
296  /// # Return Value / 戻り値
297  /// - `QueueSize::Limitless` - If the queue has no capacity limit. / キューに容量制限がない場合。
298  /// - `QueueSize::Limited(num)` - If the queue has a capacity limit. / キューに容量制限がある場合。
299  async fn remaining_capacity(&self) -> QueueSize;
300
301  /// Returns whether the operation of this queue has been interrupted.<br/>
302  /// このキューの操作が中断されたかどうかを返します。
303  ///
304  /// # Return Value / 戻り値
305  /// - `true` - If the operation is interrupted. / 操作が中断された場合。
306  /// - `false` - If the operation is not interrupted. / 操作が中断されていない場合。
307  async fn is_interrupted(&self) -> bool;
308}
309
310#[async_trait]
311pub trait BlockingQueueWriter<E: Element>: BlockingQueueBase<E> + QueueWriter<E> {
312  /// Inserts the specified element into this queue. If necessary, waits until space is available.<br/>
313  /// 指定された要素をこのキューに挿入します。必要に応じて、空きが生じるまで待機します。
314  ///
315  /// # Arguments / 引数
316  /// - `element` - The element to be inserted. / 挿入する要素。
317  ///
318  /// # Return Value / 戻り値
319  /// - `Ok(())` - If the element is inserted successfully. / 要素が正常に挿入された場合。
320  /// - `Err(QueueError::OfferError(element))` - If the element cannot be inserted. / 要素を挿入できなかった場合。
321  /// - `Err(QueueError::InterruptedError)` - If the operation is interrupted. / 操作が中断された場合。
322  async fn put(&mut self, element: E) -> Result<(), QueueError<E>>;
323
324  /// Interrupts the operation of this queue.<br/>
325  /// このキューの操作を中断します。
326  async fn interrupt(&mut self);
327}
328
329#[async_trait]
330pub trait BlockingQueueReader<E: Element>: BlockingQueueBase<E> {
331  /// Retrieve the head of this queue and delete it. If necessary, wait until an element becomes available.<br/>
332  /// このキューの先頭を取得して削除します。必要に応じて、要素が利用可能になるまで待機します。
333  ///
334  /// # Return Value / 戻り値
335  /// - `Ok(Some(element))` - If the element is retrieved successfully. / 要素が正常に取得された場合。
336  /// - `Ok(None)` - If the queue is empty. / キューが空の場合。
337  /// - `Err(QueueError::InterruptedError)` - If the operation is interrupted. / 操作が中断された場合。
338  async fn take(&mut self) -> Result<Option<E>, QueueError<E>>;
339}
nexus_acto_rs/util/queue.rs

nexus_acto_rs/util/
queue.rs
