gear_common/storage/complex/
queue.rs

1// This file is part of Gear.
2
3// Copyright (C) 2022-2025 Gear Technologies Inc.
4// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
5
6// This program is free software: you can redistribute it and/or modify
7// it under the terms of the GNU General Public License as published by
8// the Free Software Foundation, either version 3 of the License, or
9// (at your option) any later version.
10
11// This program is distributed in the hope that it will be useful,
12// but WITHOUT ANY WARRANTY; without even the implied warranty of
13// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14// GNU General Public License for more details.
15
16// You should have received a copy of the GNU General Public License
17// along with this program. If not, see <https://www.gnu.org/licenses/>.
18
19//! Module for message queue implementation.
20//!
21//! Message queue provides functionality of storing messages,
22//! addressed to programs.
23
24use crate::storage::{Counted, Dequeue, DequeueError, IterableMap, KeyFor};
25use core::marker::PhantomData;
26
27/// Represents message queue managing logic.
28pub trait Queue {
29    /// Stored values type.
30    type Value;
31    /// Inner error type of queue storing algorithm.
32    type Error: DequeueError;
33    /// Output error type of the queue.
34    type OutputError: From<Self::Error>;
35
36    /// Removes and returns message from the beginning of the queue,
37    /// if present,
38    fn dequeue() -> Result<Option<Self::Value>, Self::OutputError>;
39
40    /// Mutates all values in queue with given function.
41    fn mutate_values<F: FnMut(Self::Value) -> Self::Value>(f: F);
42
43    /// Inserts given value at the end of the queue.
44    fn queue(value: Self::Value) -> Result<(), Self::OutputError>;
45
46    /// Removes all values from queue.
47    fn clear();
48
49    /// Inserts given value at the beginning of the queue.
50    ///
51    /// Should be used only for cases, when message was dequeued and
52    /// it's execution should be postponed until the next block.
53    fn requeue(value: Self::Value) -> Result<(), Self::OutputError>;
54}
55
56/// `Mailbox` implementation based on `Dequeue`.
57///
58/// Generic parameter `KeyGen` presents key generation for given values.
59pub struct QueueImpl<T, OutputError, KeyGen>(PhantomData<(T, OutputError, KeyGen)>)
60where
61    T: Dequeue,
62    OutputError: From<T::Error>,
63    KeyGen: KeyFor<Key = T::Key, Value = T::Value>;
64
65// Implementation of `Queue` for `QueueImpl`.
66impl<T, OutputError, KeyGen> Queue for QueueImpl<T, OutputError, KeyGen>
67where
68    T: Dequeue,
69    OutputError: From<T::Error>,
70    T::Error: DequeueError,
71    KeyGen: KeyFor<Key = T::Key, Value = T::Value>,
72{
73    type Value = T::Value;
74    type Error = T::Error;
75    type OutputError = OutputError;
76
77    fn dequeue() -> Result<Option<Self::Value>, Self::OutputError> {
78        T::pop_front().map_err(Into::into)
79    }
80
81    fn mutate_values<F: FnMut(Self::Value) -> Self::Value>(f: F) {
82        T::mutate_values(f)
83    }
84
85    fn queue(value: Self::Value) -> Result<(), Self::OutputError> {
86        let key = KeyGen::key_for(&value);
87        T::push_back(key, value).map_err(Into::into)
88    }
89
90    fn clear() {
91        T::clear()
92    }
93
94    fn requeue(value: Self::Value) -> Result<(), Self::OutputError> {
95        let key = KeyGen::key_for(&value);
96        T::push_front(key, value).map_err(Into::into)
97    }
98}
99
100// Implementation of `Counted` trait for `QueueImpl` in case,
101// when inner `Dequeue` implements `Counted.
102impl<T, OutputError, KeyGen> Counted for QueueImpl<T, OutputError, KeyGen>
103where
104    T: Dequeue + Counted,
105    OutputError: From<T::Error>,
106    KeyGen: KeyFor<Key = T::Key, Value = T::Value>,
107{
108    type Length = T::Length;
109
110    fn len() -> Self::Length {
111        T::len()
112    }
113}
114
115/// Drain iterator over queue's values.
116///
117/// Removes element on each iteration.
118pub struct QueueDrainIter<T, OutputError>(T::DrainIter, PhantomData<OutputError>)
119where
120    T: Dequeue + IterableMap<Result<T::Value, T::Error>>,
121    OutputError: From<T::Error>;
122
123// `Iterator` implementation for `QueueDrainIter`.
124impl<T, OutputError> Iterator for QueueDrainIter<T, OutputError>
125where
126    T: Dequeue + IterableMap<Result<T::Value, T::Error>>,
127    OutputError: From<T::Error>,
128{
129    type Item = Result<T::Value, OutputError>;
130
131    fn next(&mut self) -> Option<Self::Item> {
132        self.0.next().map(|res| res.map_err(Into::into))
133    }
134}
135
136/// Common iterator over queue's values.
137pub struct QueueIter<T, OutputError>(T::Iter, PhantomData<OutputError>)
138where
139    T: Dequeue + IterableMap<Result<T::Value, T::Error>>,
140    OutputError: From<T::Error>;
141
142// `Iterator` implementation for `QueueIter`.
143impl<T, OutputError> Iterator for QueueIter<T, OutputError>
144where
145    T: Dequeue + IterableMap<Result<T::Value, T::Error>>,
146    OutputError: From<T::Error>,
147{
148    type Item = Result<T::Value, OutputError>;
149
150    fn next(&mut self) -> Option<Self::Item> {
151        self.0.next().map(|res| res.map_err(Into::into))
152    }
153}
154
155// `IterableMap` implementation for `QueueImpl`, returning iterators,
156// presented with `QueueIter` and `QueueDrainIter`.
157impl<T, OutputError, KeyGen> IterableMap<Result<T::Value, OutputError>>
158    for QueueImpl<T, OutputError, KeyGen>
159where
160    T: Dequeue + IterableMap<Result<T::Value, T::Error>>,
161    OutputError: From<T::Error>,
162    KeyGen: KeyFor<Key = T::Key, Value = T::Value>,
163{
164    type DrainIter = QueueDrainIter<T, OutputError>;
165    type Iter = QueueIter<T, OutputError>;
166
167    fn drain() -> Self::DrainIter {
168        QueueDrainIter(T::drain(), PhantomData::<OutputError>)
169    }
170
171    fn iter() -> Self::Iter {
172        QueueIter(T::iter(), PhantomData::<OutputError>)
173    }
174}
gear_common/storage/complex/queue.rs

gear_common/storage/complex/
queue.rs
