gear_common/storage/complex/
queue.rs

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

gear_common/storage/complex/
queue.rs
