osal_rs/traits/queue.rs
1/***************************************************************************
2 *
3 * osal-rs
4 * Copyright (C) 2026 Antonio Salsi <passy.linux@zresa.it>
5 *
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Lesser General Public
8 * License as published by the Free Software Foundation; either
9 * version 2.1 of the License, or (at your option) any later version.
10 *
11 * This library 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 GNU
14 * Lesser General Public License for more details.
15 *
16 * You should have received a copy of the GNU Lesser General Public
17 * License along with this library; if not, see <https://www.gnu.org/licenses/>.
18 *
19 ***************************************************************************/
20
21//! Queue traits for inter-task communication.
22//!
23//! Provides both raw byte-based queues and type-safe streamed queues
24//! for message passing between tasks.
25//!
26//! # Overview
27//!
28//! Queues implement FIFO (First-In-First-Out) message passing between tasks,
29//! enabling the producer-consumer pattern and other inter-task communication
30//! patterns. Messages are copied into and out of the queue.
31//!
32//! # Queue Types
33//!
34//! - **`Queue`**: Raw byte-oriented queue for variable-sized or untyped data
35//! - **`QueueStreamed<T>`**: Type-safe queue for structured messages
36//!
37//! # Communication Patterns
38//!
39//! - **Producer-Consumer**: One or more producers send messages, one consumer processes them
40//! - **Work Queue**: Distribute tasks among multiple worker tasks
41//! - **Event Notification**: Send status updates or notifications between tasks
42//!
43//! # Timeout Behavior
44//!
45//! - `0`: Non-blocking - return immediately if queue is full/empty
46//! - `n`: Wait up to `n` ticks for space/data to become available
47//! - `TickType::MAX`: Block indefinitely until operation succeeds
48//!
49//! # Examples
50//!
51//! ```
52//! use osal_rs::os::{Queue, QueueFn};
53//!
54//! // Create a queue for 10 messages of 4 bytes each
55//! let queue = Queue::new(10, 4).unwrap();
56//!
57//! // Producer task
58//! let data = [1u8, 2, 3, 4];
59//! queue.post(&data, 1000).unwrap();
60//!
61//! // Consumer task
62//! let mut buffer = [0u8; 4];
63//! queue.fetch(&mut buffer, 1000).unwrap();
64//! ```
65#[cfg(not(feature = "serde"))]
66use crate::os::Deserialize;
67
68#[cfg(feature = "serde")]
69use osal_rs_serde::Deserialize;
70
71use crate::os::types::TickType;
72use crate::utils::Result;
73
74/// Raw byte-oriented queue for inter-task message passing.
75///
76/// This trait defines a FIFO queue that works with raw byte arrays,
77/// suitable for variable-sized messages or when type safety is not required.
78///
79/// # Memory Layout
80///
81/// The queue capacity is fixed at creation time. Each message slot can
82/// hold up to the maximum message size specified during creation.
83///
84/// # Thread Safety
85///
86/// All methods are thread-safe. Multiple producers and consumers can
87/// safely access the same queue concurrently.
88///
89/// # Performance
90///
91/// Messages are copied into and out of the queue. For large messages,
92/// consider using a queue of pointers or references instead.
93///
94/// # Examples
95///
96/// ```
97/// use osal_rs::os::*;
98///
99/// // Create queue: 10 slots, 32 bytes per message
100/// let queue = Queue::new(10, 32).unwrap();
101///
102/// // Producer sends data - a whole message-sized slot at a time
103/// let mut data = [0u8; 32];
104/// data[..4].copy_from_slice(&[1, 2, 3, 4]);
105/// queue.post(&data, 100).unwrap();
106///
107/// // Consumer receives data
108/// let mut buffer = [0u8; 32];
109/// queue.fetch(&mut buffer, 100).unwrap();
110/// assert_eq!(&buffer[..4], &[1, 2, 3, 4]);
111/// ```
112pub trait Queue {
113
114 /// Returns `true` if the underlying OS handle is null, i.e. the mutex
115 /// has not been created yet or has already been deleted.
116 fn is_null(&self) -> bool;
117
118 /// Fetches a message from the queue (blocking).
119 ///
120 /// Removes and retrieves the oldest message from the queue (FIFO order).
121 /// Blocks the calling task if the queue is empty.
122 ///
123 /// # Parameters
124 ///
125 /// * `buffer` - Buffer to receive the message data (should match queue message size)
126 /// * `time` - Maximum ticks to wait for a message:
127 /// - `0`: Return immediately if empty
128 /// - `n`: Wait up to `n` ticks
129 /// - `TickType::MAX`: Wait forever
130 ///
131 /// # Returns
132 ///
133 /// * `Ok(())` - Message received successfully
134 /// * `Err(Error::Timeout)` - Queue was empty for entire timeout period
135 /// * `Err(Error)` - Other error occurred
136 ///
137 /// # Examples
138 ///
139 /// ```
140 /// use osal_rs::os::*;
141 ///
142 /// let queue = Queue::new(4, 16).unwrap();
143 /// queue.post(&[0xAAu8; 16], 100).unwrap();
144 ///
145 /// let mut buffer = [0u8; 16];
146 ///
147 /// // Wait up to 1000 ticks
148 /// match queue.fetch(&mut buffer, 1000) {
149 /// Ok(()) => assert_eq!(buffer, [0xAAu8; 16]),
150 /// Err(_) => panic!("timeout - no message available"),
151 /// }
152 ///
153 /// // The queue is empty again, so this one does time out.
154 /// assert!(queue.fetch(&mut buffer, 10).is_err());
155 /// ```
156 fn fetch(&self, buffer: &mut [u8], time: TickType) -> Result<()>;
157
158 /// Fetches a message from ISR context (non-blocking).
159 ///
160 /// ISR-safe version of `fetch()`. Returns immediately without blocking.
161 /// Must only be called from interrupt context.
162 ///
163 /// # Parameters
164 ///
165 /// * `buffer` - Buffer to receive the message data
166 ///
167 /// # Returns
168 ///
169 /// * `Ok(())` - Message received successfully
170 /// * `Err(Error)` - Queue is empty
171 ///
172 /// # Examples
173 ///
174 /// ```
175 /// use osal_rs::os::*;
176 ///
177 /// let queue = Queue::new(4, 16).unwrap();
178 /// queue.post(&[7u8; 16], 100).unwrap();
179 ///
180 /// // In interrupt handler
181 /// let mut buffer = [0u8; 16];
182 /// if queue.fetch_from_isr(&mut buffer).is_ok() {
183 /// // Process message quickly
184 /// assert_eq!(buffer, [7u8; 16]);
185 /// }
186 ///
187 /// // Nothing left: reported immediately instead of blocking the "ISR".
188 /// assert!(queue.fetch_from_isr(&mut buffer).is_err());
189 /// ```
190 fn fetch_from_isr(&self, buffer: &mut [u8]) -> Result<()>;
191
192 /// Posts a message to the queue (blocking).
193 ///
194 /// Adds a new message to the end of the queue (FIFO order).
195 /// Blocks the calling task if the queue is full.
196 ///
197 /// # Parameters
198 ///
199 /// * `item` - The message data to send (must not exceed queue message size)
200 /// * `time` - Maximum ticks to wait if queue is full:
201 /// - `0`: Return immediately if full
202 /// - `n`: Wait up to `n` ticks for space
203 /// - `TickType::MAX`: Wait forever
204 ///
205 /// # Returns
206 ///
207 /// * `Ok(())` - Message sent successfully
208 /// * `Err(Error::Timeout)` - Queue was full for entire timeout period
209 /// * `Err(Error)` - Other error occurred
210 ///
211 /// # Examples
212 ///
213 /// ```
214 /// use osal_rs::os::*;
215 ///
216 /// // Room for a single 4-byte message.
217 /// let queue = Queue::new(1, 4).unwrap();
218 /// let data = [1u8, 2, 3, 4];
219 ///
220 /// // Try to send, wait up to 1000 ticks if full
221 /// match queue.post(&data, 1000) {
222 /// Ok(()) => (), // sent successfully
223 /// Err(_) => panic!("queue full, couldn't send"),
224 /// }
225 ///
226 /// // The only slot is taken and nobody is fetching: this one times out.
227 /// assert!(queue.post(&data, 10).is_err());
228 /// ```
229 fn post(&self, item: &[u8], time: TickType) -> Result<()>;
230
231 /// Posts a message from ISR context (non-blocking).
232 ///
233 /// ISR-safe version of `post()`. Returns immediately without blocking.
234 /// Must only be called from interrupt context.
235 ///
236 /// # Parameters
237 ///
238 /// * `item` - The message data to send
239 ///
240 /// # Returns
241 ///
242 /// * `Ok(())` - Message sent successfully
243 /// * `Err(Error)` - Queue is full
244 ///
245 /// # Examples
246 ///
247 /// ```
248 /// use osal_rs::os::*;
249 ///
250 /// let queue = Queue::new(1, 2).unwrap();
251 ///
252 /// // In interrupt handler
253 /// let data = [0x42u8, 0x13];
254 /// if queue.post_from_isr(&data).is_err() {
255 /// // Queue full, message dropped
256 /// }
257 ///
258 /// // The single slot is now taken, so the next one really is dropped.
259 /// assert!(queue.post_from_isr(&data).is_err());
260 /// ```
261 fn post_from_isr(&self, item: &[u8]) -> Result<()>;
262
263 /// Deletes the queue and frees its resources.
264 ///
265 /// # Safety
266 ///
267 /// Ensure no tasks are blocked on this queue before deletion.
268 /// Calling this while tasks are waiting may cause undefined behavior.
269 ///
270 /// # Examples
271 ///
272 /// ```
273 /// use osal_rs::os::*;
274 ///
275 /// let mut queue = Queue::new(10, 16).unwrap();
276 /// // Use queue...
277 /// queue.delete();
278 /// assert!(queue.is_null());
279 /// ```
280 fn delete(&mut self);
281}
282
283/// Type-safe queue for structured message passing.
284///
285/// This trait provides a queue that works with specific types,
286/// offering compile-time type safety for queue operations.
287///
288/// # Type Safety
289///
290/// Unlike raw `Queue`, `QueueStreamed` ensures that only messages
291/// of type `T` can be sent and received, preventing type confusion
292/// at compile time.
293///
294/// # Serialization
295///
296/// Messages are automatically serialized when sent and deserialized
297/// when received. The type `T` must implement the `Deserialize` trait.
298///
299/// # Type Parameters
300///
301/// * `T` - The message type (must implement `Deserialize`)
302///
303/// # Examples
304///
305#[cfg_attr(not(feature = "serde"), doc = "```")]
306#[cfg_attr(feature = "serde", doc = "```ignore")]
307/// use osal_rs::os::*;
308/// use osal_rs::utils::{Error, Result};
309///
310/// // Wire format: 4 bytes of `id`, 2 of `temperature`, 1 of `humidity`,
311/// // little-endian. Keeping the message in its encoded form is what lets
312/// // `to_bytes` hand out a borrowed slice.
313/// const SENSOR_DATA_LEN: usize = 7;
314///
315/// #[derive(Clone, Copy, Debug, PartialEq)]
316/// struct SensorData([u8; SENSOR_DATA_LEN]);
317///
318/// impl SensorData {
319/// fn new(id: u32, temperature: i16, humidity: u8) -> Self {
320/// let mut raw = [0u8; SENSOR_DATA_LEN];
321/// raw[..4].copy_from_slice(&id.to_le_bytes());
322/// raw[4..6].copy_from_slice(&temperature.to_le_bytes());
323/// raw[6] = humidity;
324/// Self(raw)
325/// }
326///
327/// fn id(&self) -> u32 {
328/// u32::from_le_bytes(self.0[..4].try_into().unwrap())
329/// }
330/// }
331///
332/// impl BytesHasLen for SensorData {
333/// fn len(&self) -> usize { SENSOR_DATA_LEN }
334/// }
335///
336/// impl Serialize for SensorData {
337/// fn to_bytes(&self) -> &[u8] { &self.0 }
338/// }
339///
340/// impl Deserialize for SensorData {
341/// fn from_bytes(bytes: &[u8]) -> Result<Self> {
342/// if bytes.len() < SENSOR_DATA_LEN {
343/// return Err(Error::OutOfIndex);
344/// }
345/// let mut raw = [0u8; SENSOR_DATA_LEN];
346/// raw.copy_from_slice(&bytes[..SENSOR_DATA_LEN]);
347/// Ok(Self(raw))
348/// }
349/// }
350///
351/// let queue = QueueStreamed::<SensorData>::new(10, SENSOR_DATA_LEN as _).unwrap();
352///
353/// // Producer
354/// let data = SensorData::new(1, 235, 65);
355/// queue.post(&data, 100).unwrap();
356///
357/// // Consumer
358/// let mut received = SensorData::new(0, 0, 0);
359/// queue.fetch(&mut received, 100).unwrap();
360/// assert_eq!(received.id(), 1);
361/// assert_eq!(received, data);
362/// ```
363
364pub trait QueueStreamed<T>
365where
366 T: Deserialize + Sized {
367
368 /// Fetches a typed message from the queue (blocking).
369 ///
370 /// Removes and deserializes the oldest message from the queue.
371 /// Blocks the calling task if the queue is empty.
372 ///
373 /// # Parameters
374 ///
375 /// * `buffer` - Mutable reference to receive the deserialized message
376 /// * `time` - Maximum ticks to wait for a message:
377 /// - `0`: Return immediately if empty
378 /// - `n`: Wait up to `n` ticks
379 /// - `TickType::MAX`: Wait forever
380 ///
381 /// # Returns
382 ///
383 /// * `Ok(())` - Message received and deserialized successfully
384 /// * `Err(Error::Timeout)` - Queue was empty for entire timeout period
385 /// * `Err(Error)` - Deserialization error or other error
386 ///
387 /// # Examples
388 ///
389 #[cfg_attr(not(feature = "serde"), doc = "```")]
390 #[cfg_attr(feature = "serde", doc = "```ignore")]
391 /// use osal_rs::os::*;
392 /// # use osal_rs::utils::{Error, Result};
393 /// # #[derive(Clone, Copy, Debug, Default, PartialEq)]
394 /// # struct Message([u8; 4]);
395 /// # impl Message {
396 /// # fn new(id: u32) -> Self { Self(id.to_le_bytes()) }
397 /// # fn id(&self) -> u32 { u32::from_le_bytes(self.0) }
398 /// # }
399 /// # impl BytesHasLen for Message { fn len(&self) -> usize { 4 } }
400 /// # impl Serialize for Message { fn to_bytes(&self) -> &[u8] { &self.0 } }
401 /// # impl Deserialize for Message {
402 /// # fn from_bytes(bytes: &[u8]) -> Result<Self> {
403 /// # if bytes.len() < 4 { return Err(Error::OutOfIndex); }
404 /// # Ok(Self(bytes[..4].try_into().unwrap()))
405 /// # }
406 /// # }
407 /// let queue = QueueStreamed::<Message>::new(4, 4).unwrap();
408 /// queue.post(&Message::new(42), 100).unwrap();
409 ///
410 /// let mut msg = Message::default();
411 ///
412 /// match queue.fetch(&mut msg, 1000) {
413 /// Ok(()) => assert_eq!(msg.id(), 42),
414 /// Err(_) => panic!("no message available"),
415 /// }
416 /// ```
417 fn fetch(&self, buffer: &mut T, time: TickType) -> Result<()>;
418
419 /// Fetches a typed message from ISR context (non-blocking).
420 ///
421 /// ISR-safe version of `fetch()`. Returns immediately without blocking.
422 /// Must only be called from interrupt context.
423 ///
424 /// # Parameters
425 ///
426 /// * `buffer` - Mutable reference to receive the deserialized message
427 ///
428 /// # Returns
429 ///
430 /// * `Ok(())` - Message received and deserialized successfully
431 /// * `Err(Error)` - Queue is empty or deserialization failed
432 ///
433 /// # Examples
434 ///
435 #[cfg_attr(not(feature = "serde"), doc = "```")]
436 #[cfg_attr(feature = "serde", doc = "```ignore")]
437 /// use osal_rs::os::*;
438 /// # use osal_rs::utils::{Error, Result};
439 /// # #[derive(Clone, Copy, Debug, Default, PartialEq)]
440 /// # struct Message([u8; 4]);
441 /// # impl Message {
442 /// # fn new(id: u32) -> Self { Self(id.to_le_bytes()) }
443 /// # fn id(&self) -> u32 { u32::from_le_bytes(self.0) }
444 /// # }
445 /// # impl BytesHasLen for Message { fn len(&self) -> usize { 4 } }
446 /// # impl Serialize for Message { fn to_bytes(&self) -> &[u8] { &self.0 } }
447 /// # impl Deserialize for Message {
448 /// # fn from_bytes(bytes: &[u8]) -> Result<Self> {
449 /// # if bytes.len() < 4 { return Err(Error::OutOfIndex); }
450 /// # Ok(Self(bytes[..4].try_into().unwrap()))
451 /// # }
452 /// # }
453 /// let queue = QueueStreamed::<Message>::new(4, 4).unwrap();
454 /// queue.post(&Message::new(7), 100).unwrap();
455 ///
456 /// // In interrupt handler
457 /// let mut msg = Message::default();
458 /// if queue.fetch_from_isr(&mut msg).is_ok() {
459 /// // Process message
460 /// assert_eq!(msg.id(), 7);
461 /// }
462 ///
463 /// // Queue empty: reported immediately instead of blocking the "ISR".
464 /// assert!(queue.fetch_from_isr(&mut msg).is_err());
465 /// ```
466 fn fetch_from_isr(&self, buffer: &mut T) -> Result<()>;
467
468 /// Posts a typed message to the queue (blocking).
469 ///
470 /// Serializes and adds a new message to the end of the queue.
471 /// Blocks the calling task if the queue is full.
472 ///
473 /// # Parameters
474 ///
475 /// * `item` - Reference to the message to serialize and send
476 /// * `time` - Maximum ticks to wait if queue is full:
477 /// - `0`: Return immediately if full
478 /// - `n`: Wait up to `n` ticks for space
479 /// - `TickType::MAX`: Wait forever
480 ///
481 /// # Returns
482 ///
483 /// * `Ok(())` - Message serialized and sent successfully
484 /// * `Err(Error::Timeout)` - Queue was full for entire timeout period
485 /// * `Err(Error)` - Serialization error or other error
486 ///
487 /// # Examples
488 ///
489 #[cfg_attr(not(feature = "serde"), doc = "```")]
490 #[cfg_attr(feature = "serde", doc = "```ignore")]
491 /// use osal_rs::os::*;
492 /// # use osal_rs::utils::{Error, Result};
493 /// # #[derive(Clone, Copy, Debug, Default, PartialEq)]
494 /// # struct Message([u8; 4]);
495 /// # impl Message {
496 /// # fn new(id: u32) -> Self { Self(id.to_le_bytes()) }
497 /// # fn id(&self) -> u32 { u32::from_le_bytes(self.0) }
498 /// # }
499 /// # impl BytesHasLen for Message { fn len(&self) -> usize { 4 } }
500 /// # impl Serialize for Message { fn to_bytes(&self) -> &[u8] { &self.0 } }
501 /// # impl Deserialize for Message {
502 /// # fn from_bytes(bytes: &[u8]) -> Result<Self> {
503 /// # if bytes.len() < 4 { return Err(Error::OutOfIndex); }
504 /// # Ok(Self(bytes[..4].try_into().unwrap()))
505 /// # }
506 /// # }
507 /// // Room for a single message.
508 /// let queue = QueueStreamed::<Message>::new(1, 4).unwrap();
509 /// let msg = Message::new(42);
510 ///
511 /// match queue.post(&msg, 1000) {
512 /// Ok(()) => (), // sent successfully
513 /// Err(_) => panic!("failed to send"),
514 /// }
515 ///
516 /// // The only slot is taken and nobody is fetching: this one times out.
517 /// assert!(queue.post(&msg, 10).is_err());
518 /// ```
519 fn post(&self, item: &T, time: TickType) -> Result<()>;
520
521 /// Posts a typed message from ISR context (non-blocking).
522 ///
523 /// ISR-safe version of `post()`. Returns immediately without blocking.
524 /// Must only be called from interrupt context.
525 ///
526 /// # Parameters
527 ///
528 /// * `item` - Reference to the message to serialize and send
529 ///
530 /// # Returns
531 ///
532 /// * `Ok(())` - Message serialized and sent successfully
533 /// * `Err(Error)` - Queue is full or serialization failed
534 ///
535 /// # Examples
536 ///
537 #[cfg_attr(not(feature = "serde"), doc = "```")]
538 #[cfg_attr(feature = "serde", doc = "```ignore")]
539 /// use osal_rs::os::*;
540 /// # use osal_rs::utils::{Error, Result};
541 /// # #[derive(Clone, Copy, Debug, Default, PartialEq)]
542 /// # struct Message([u8; 4]);
543 /// # impl Message {
544 /// # fn new(id: u32) -> Self { Self(id.to_le_bytes()) }
545 /// # fn id(&self) -> u32 { u32::from_le_bytes(self.0) }
546 /// # }
547 /// # impl BytesHasLen for Message { fn len(&self) -> usize { 4 } }
548 /// # impl Serialize for Message { fn to_bytes(&self) -> &[u8] { &self.0 } }
549 /// # impl Deserialize for Message {
550 /// # fn from_bytes(bytes: &[u8]) -> Result<Self> {
551 /// # if bytes.len() < 4 { return Err(Error::OutOfIndex); }
552 /// # Ok(Self(bytes[..4].try_into().unwrap()))
553 /// # }
554 /// # }
555 /// let queue = QueueStreamed::<Message>::new(1, 4).unwrap();
556 ///
557 /// // In interrupt handler
558 /// let msg = Message::new(1);
559 /// if queue.post_from_isr(&msg).is_err() {
560 /// // Queue full, message dropped
561 /// }
562 ///
563 /// // The single slot is now taken, so the next one really is dropped.
564 /// assert!(queue.post_from_isr(&msg).is_err());
565 /// ```
566 fn post_from_isr(&self, item: &T) -> Result<()>;
567
568 /// Deletes the queue and frees its resources.
569 ///
570 /// # Safety
571 ///
572 /// Ensure no tasks are blocked on this queue before deletion.
573 /// Calling this while tasks are waiting may cause undefined behavior.
574 ///
575 /// # Examples
576 ///
577 #[cfg_attr(not(feature = "serde"), doc = "```")]
578 #[cfg_attr(feature = "serde", doc = "```ignore")]
579 /// use osal_rs::os::*;
580 /// # use osal_rs::utils::{Error, Result};
581 /// # #[derive(Clone, Copy, Debug, Default, PartialEq)]
582 /// # struct Message([u8; 4]);
583 /// # impl Message {
584 /// # fn new(id: u32) -> Self { Self(id.to_le_bytes()) }
585 /// # fn id(&self) -> u32 { u32::from_le_bytes(self.0) }
586 /// # }
587 /// # impl BytesHasLen for Message { fn len(&self) -> usize { 4 } }
588 /// # impl Serialize for Message { fn to_bytes(&self) -> &[u8] { &self.0 } }
589 /// # impl Deserialize for Message {
590 /// # fn from_bytes(bytes: &[u8]) -> Result<Self> {
591 /// # if bytes.len() < 4 { return Err(Error::OutOfIndex); }
592 /// # Ok(Self(bytes[..4].try_into().unwrap()))
593 /// # }
594 /// # }
595 /// let mut queue = QueueStreamed::<Message>::new(10, core::mem::size_of::<Message>() as _).unwrap();
596 ///
597 /// // Use queue...
598 /// queue.post(&Message::new(1), 100).unwrap();
599 ///
600 /// queue.delete();
601 /// ```
602 fn delete(&mut self);
603}