osal_rs/freertos/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-based inter-thread communication for FreeRTOS.
22//!
23//! This module provides FIFO queue primitives for safe message passing between threads
24//! and interrupt service routines. Supports both byte-based and typed queues.
25
26use core::ffi::c_void;
27use core::fmt::{Debug, Display};
28use core::marker::PhantomData;
29use core::ops::Deref;
30
31use alloc::vec::Vec;
32
33use super::ffi::{QueueHandle, pdFALSE, vQueueDelete, xQueueGenericCreate, xQueueReceive, xQueueReceiveFromISR};
34use super::types::{BaseType, UBaseType, TickType};
35use super::system::System;
36use crate::traits::{ToTick, QueueFn, SystemFn, QueueStreamedFn, BytesHasLen};
37#[cfg(not(feature = "serde"))]
38use crate::traits::{Serialize, Deserialize};
39
40#[cfg(feature = "serde")]
41use osal_rs_serde::{Serialize, Deserialize, to_bytes};
42
43pub trait StructSerde : Serialize + BytesHasLen + Deserialize {}
44
45impl<T> StructSerde for T where T: Serialize + BytesHasLen + Deserialize {}
46
47use crate::utils::{Result, Error};
48
49
50/// A FIFO queue for byte-based message passing.
51///
52/// Provides a thread-safe queue implementation for sending and receiving
53/// raw byte slices between threads. Supports both blocking and ISR-safe operations.
54///
55/// # Examples
56///
57/// ## Basic queue usage
58///
59/// ```ignore
60/// use osal_rs::os::{Queue, QueueFn};
61/// use core::time::Duration;
62///
63/// // Create a queue with 10 slots, each 32 bytes
64/// let queue = Queue::new(10, 32).unwrap();
65///
66/// // Send data
67/// let data = [1u8, 2, 3, 4];
68/// queue.post_with_to_tick(&data, Duration::from_millis(100)).unwrap();
69///
70/// // Receive data
71/// let mut buffer = [0u8; 4];
72/// queue.fetch_with_to_tick(&mut buffer, Duration::from_millis(100)).unwrap();
73/// assert_eq!(buffer, [1, 2, 3, 4]);
74/// ```
75///
76/// ## Producer-consumer pattern
77///
78/// ```ignore
79/// use osal_rs::os::{Queue, QueueFn, Thread};
80/// use alloc::sync::Arc;
81/// use core::time::Duration;
82///
83/// let queue = Arc::new(Queue::new(5, 4).unwrap());
84/// let queue_clone = queue.clone();
85///
86/// // Consumer thread
87/// let consumer = Thread::new("consumer", 2048, 5, move || {
88/// let mut buffer = [0u8; 4];
89/// loop {
90/// if queue_clone.fetch(&mut buffer, 1000).is_ok() {
91/// println!("Received: {:?}", buffer);
92/// }
93/// }
94/// }).unwrap();
95///
96/// consumer.start().unwrap();
97///
98/// // Producer
99/// let data = [0xAA, 0xBB, 0xCC, 0xDD];
100/// queue.post(&data, 1000).unwrap();
101/// ```
102pub struct Queue (QueueHandle);
103
104unsafe impl Send for Queue {}
105unsafe impl Sync for Queue {}
106
107impl Queue {
108 /// Creates a new queue.
109 ///
110 /// # Parameters
111 ///
112 /// * `size` - Maximum number of messages the queue can hold
113 /// * `message_size` - Size in bytes of each message
114 ///
115 /// # Returns
116 ///
117 /// * `Ok(Self)` - Successfully created queue
118 /// * `Err(Error)` - Creation failed (insufficient memory, etc.)
119 ///
120 /// # Examples
121 ///
122 /// ```ignore
123 /// use osal_rs::os::{Queue, QueueFn};
124 ///
125 /// // Queue for 5 messages of 16 bytes each
126 /// let queue = Queue::new(5, 16).unwrap();
127 /// ```
128 pub fn new (size: UBaseType, message_size: UBaseType) -> Result<Self> {
129 const QUEUE_TYPE_BASE: u8 = 0; // queueQUEUE_TYPE_BASE
130 let handle = unsafe { xQueueGenericCreate(size, message_size, QUEUE_TYPE_BASE) };
131 if handle.is_null() {
132 Err(Error::OutOfMemory)
133 } else {
134 Ok(Self (handle))
135 }
136 }
137
138 /// Receives data from the queue with a convertible timeout.
139 ///
140 /// This is a convenience method that accepts any type implementing `ToTick`
141 /// (like `Duration`) and converts it to ticks before calling `fetch()`.
142 ///
143 /// # Arguments
144 ///
145 /// * `buffer` - Mutable slice to receive data into
146 /// * `time` - Timeout value (e.g., `Duration::from_millis(100)`)
147 ///
148 /// # Returns
149 ///
150 /// * `Ok(())` - Data successfully received
151 /// * `Err(Error::Timeout)` - No data available within timeout
152 ///
153 /// # Examples
154 ///
155 /// ```ignore
156 /// use osal_rs::os::{Queue, QueueFn};
157 /// use core::time::Duration;
158 ///
159 /// let queue = Queue::new(5, 16).unwrap();
160 /// let mut buffer = [0u8; 16];
161 /// queue.fetch_with_to_tick(&mut buffer, Duration::from_millis(100))?;
162 /// ```
163 #[inline]
164 pub fn fetch_with_to_tick(&self, buffer: &mut [u8], time: impl ToTick) -> Result<()> {
165 self.fetch(buffer, time.to_ticks())
166 }
167
168 /// Sends data to the queue with a convertible timeout.
169 ///
170 /// This is a convenience method that accepts any type implementing `ToTick`
171 /// (like `Duration`) and converts it to ticks before calling `post()`.
172 ///
173 /// # Arguments
174 ///
175 /// * `item` - Slice of data to send
176 /// * `time` - Timeout value (e.g., `Duration::from_millis(100)`)
177 ///
178 /// # Returns
179 ///
180 /// * `Ok(())` - Data successfully sent
181 /// * `Err(Error::Timeout)` - Queue full, could not send within timeout
182 ///
183 /// # Examples
184 ///
185 /// ```ignore
186 /// use osal_rs::os::{Queue, QueueFn};
187 /// use core::time::Duration;
188 ///
189 /// let queue = Queue::new(5, 16).unwrap();
190 /// let data = [1u8, 2, 3, 4];
191 /// queue.post_with_to_tick(&data, Duration::from_millis(100))?;
192 /// ```
193 #[inline]
194 pub fn post_with_to_tick(&self, item: &[u8], time: impl ToTick) -> Result<()> {
195 self.post(item, time.to_ticks())
196 }
197}
198
199impl QueueFn for Queue {
200
201 /// Receives data from the queue, blocking until data is available or timeout.
202 ///
203 /// This function blocks the calling thread until data is available or the
204 /// specified timeout expires.
205 ///
206 /// # Arguments
207 ///
208 /// * `buffer` - Mutable byte slice to receive data into
209 /// * `time` - Timeout in system ticks (0 = no wait, MAX = wait forever)
210 ///
211 /// # Returns
212 ///
213 /// * `Ok(())` - Data successfully received into buffer
214 /// * `Err(Error::Timeout)` - No data available within timeout period
215 ///
216 /// # Examples
217 ///
218 /// ```ignore
219 /// use osal_rs::os::{Queue, QueueFn};
220 ///
221 /// let queue = Queue::new(5, 16).unwrap();
222 /// let mut buffer = [0u8; 16];
223 ///
224 /// // Wait up to 1000 ticks for data
225 /// match queue.fetch(&mut buffer, 1000) {
226 /// Ok(()) => println!("Received data: {:?}", buffer),
227 /// Err(_) => println!("Timeout"),
228 /// }
229 /// ```
230 fn fetch(&self, buffer: &mut [u8], time: TickType) -> Result<()> {
231 let ret = unsafe {
232 xQueueReceive(
233 self.0,
234 buffer.as_mut_ptr() as *mut c_void,
235 time,
236 )
237 };
238 if ret == 0 {
239 Err(Error::Timeout)
240 } else {
241 Ok(())
242 }
243 }
244
245 /// Receives data from the queue in an interrupt service routine (ISR).
246 ///
247 /// This is the ISR-safe version of `fetch()`. It does not block and will
248 /// trigger a context switch if a higher priority task is woken.
249 ///
250 /// # Arguments
251 ///
252 /// * `buffer` - Mutable byte slice to receive data into
253 ///
254 /// # Returns
255 ///
256 /// * `Ok(())` - Data successfully received
257 /// * `Err(Error::Timeout)` - Queue is empty
258 ///
259 /// # Safety
260 ///
261 /// Must only be called from ISR context.
262 ///
263 /// # Examples
264 ///
265 /// ```ignore
266 /// // In interrupt handler
267 /// use osal_rs::os::{Queue, QueueFn};
268 ///
269 /// fn irq_handler(queue: &Queue) {
270 /// let mut buffer = [0u8; 16];
271 /// if queue.fetch_from_isr(&mut buffer).is_ok() {
272 /// // Process received data
273 /// }
274 /// }
275 /// ```
276 fn fetch_from_isr(&self, buffer: &mut [u8]) -> Result<()> {
277
278 let mut task_woken_by_receive: BaseType = pdFALSE;
279
280 let ret = unsafe {
281 xQueueReceiveFromISR(
282 self.0,
283 buffer.as_mut_ptr() as *mut c_void,
284 &mut task_woken_by_receive
285 )
286 };
287 if ret == 0 {
288 Err(Error::Timeout)
289 } else {
290
291 System::yield_from_isr(task_woken_by_receive);
292
293 Ok(())
294 }
295 }
296
297 /// Sends data to the back of the queue, blocking until space is available.
298 ///
299 /// This function blocks the calling thread until space becomes available
300 /// or the timeout expires.
301 ///
302 /// # Arguments
303 ///
304 /// * `item` - Byte slice to send
305 /// * `time` - Timeout in system ticks (0 = no wait, MAX = wait forever)
306 ///
307 /// # Returns
308 ///
309 /// * `Ok(())` - Data successfully sent
310 /// * `Err(Error::Timeout)` - Queue full, could not send within timeout
311 ///
312 /// # Examples
313 ///
314 /// ```ignore
315 /// use osal_rs::os::{Queue, QueueFn};
316 ///
317 /// let queue = Queue::new(5, 16).unwrap();
318 /// let data = [0xAA, 0xBB, 0xCC, 0xDD];
319 ///
320 /// // Wait up to 1000 ticks to send
321 /// queue.post(&data, 1000)?;
322 /// ```
323 fn post(&self, item: &[u8], time: TickType) -> Result<()> {
324 let ret = xQueueSendToBack!(
325 self.0,
326 item.as_ptr() as *const c_void,
327 time
328 );
329
330 if ret == 0 {
331 Err(Error::Timeout)
332 } else {
333 Ok(())
334 }
335 }
336
337 /// Sends data to the queue from an interrupt service routine (ISR).
338 ///
339 /// This is the ISR-safe version of `post()`. It does not block and will
340 /// trigger a context switch if a higher priority task is woken.
341 ///
342 /// # Arguments
343 ///
344 /// * `item` - Byte slice to send
345 ///
346 /// # Returns
347 ///
348 /// * `Ok(())` - Data successfully sent
349 /// * `Err(Error::Timeout)` - Queue is full
350 ///
351 /// # Safety
352 ///
353 /// Must only be called from ISR context.
354 ///
355 /// # Examples
356 ///
357 /// ```ignore
358 /// // In interrupt handler
359 /// use osal_rs::os::{Queue, QueueFn};
360 ///
361 /// fn irq_handler(queue: &Queue) {
362 /// let data = [0x01, 0x02, 0x03];
363 /// queue.post_from_isr(&data).ok();
364 /// }
365 /// ```
366 fn post_from_isr(&self, item: &[u8]) -> Result<()> {
367
368 let mut task_woken_by_receive: BaseType = pdFALSE;
369
370 let ret = xQueueSendToBackFromISR!(
371 self.0,
372 item.as_ptr() as *const c_void,
373 &mut task_woken_by_receive
374 );
375
376 if ret == 0 {
377 Err(Error::Timeout)
378 } else {
379 System::yield_from_isr(task_woken_by_receive);
380
381 Ok(())
382 }
383 }
384
385 /// Deletes the queue and frees its resources.
386 ///
387 /// This function destroys the queue and releases any memory allocated for it.
388 /// After calling this, the queue should not be used. The handle is set to null.
389 ///
390 /// # Safety
391 ///
392 /// Ensure no threads are waiting on this queue before deleting it.
393 ///
394 /// # Examples
395 ///
396 /// ```ignore
397 /// use osal_rs::os::{Queue, QueueFn};
398 ///
399 /// let mut queue = Queue::new(5, 16).unwrap();
400 /// // Use the queue...
401 /// queue.delete();
402 /// ```
403 fn delete(&mut self) {
404 unsafe {
405 vQueueDelete(self.0);
406 self.0 = core::ptr::null_mut();
407 }
408 }
409}
410
411/// Automatically deletes the queue when it goes out of scope.
412///
413/// This ensures proper cleanup of FreeRTOS resources.
414impl Drop for Queue {
415 fn drop(&mut self) {
416 if self.0.is_null() {
417 return;
418 }
419 self.delete();
420 }
421}
422
423/// Allows dereferencing to the underlying FreeRTOS queue handle.
424impl Deref for Queue {
425 type Target = QueueHandle;
426
427 fn deref(&self) -> &Self::Target {
428 &self.0
429 }
430}
431
432/// Formats the queue for debugging purposes.
433impl Debug for Queue {
434 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
435 f.debug_struct("Queue")
436 .field("handle", &self.0)
437 .finish()
438 }
439}
440
441/// Formats the queue for display purposes.
442impl Display for Queue {
443 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
444 write!(f, "Queue {{ handle: {:?} }}", self.0)
445 }
446}
447
448/// A type-safe FIFO queue for message passing.
449///
450/// Unlike [`Queue`], which works with raw byte slices, `QueueStreamed` provides
451/// a type-safe interface for sending and receiving structured data. The type must
452/// implement serialization traits.
453///
454/// # Type Parameters
455///
456/// * `T` - The message type. Must implement `ToBytes`, `BytesHasLen`, and `FromBytes`
457///
458/// # Examples
459///
460/// ## Basic typed queue usage
461///
462/// ```ignore
463/// use osal_rs::os::{QueueStreamed, QueueStreamedFn};
464/// use core::time::Duration;
465///
466/// #[derive(Debug, Clone, Copy)]
467/// struct Message {
468/// id: u32,
469/// value: i16,
470/// }
471///
472/// // Assuming Message implements the required traits
473/// let queue: QueueStreamed<Message> = QueueStreamed::new(10, size_of::<Message>()).unwrap();
474///
475/// // Send a message
476/// let msg = Message { id: 1, value: 42 };
477/// queue.post_with_to_tick(&msg, Duration::from_millis(100)).unwrap();
478///
479/// // Receive a message
480/// let mut received = Message { id: 0, value: 0 };
481/// queue.fetch_with_to_tick(&mut received, Duration::from_millis(100)).unwrap();
482/// assert_eq!(received.id, 1);
483/// assert_eq!(received.value, 42);
484/// ```
485///
486/// ## Command queue pattern
487///
488/// ```ignore
489/// use osal_rs::os::{QueueStreamed, Thread};
490/// use alloc::sync::Arc;
491///
492/// enum Command {
493/// Start,
494/// Stop,
495/// SetValue(u32),
496/// }
497///
498/// let cmd_queue = Arc::new(QueueStreamed::<Command>::new(10, 8).unwrap());
499/// let queue_clone = cmd_queue.clone();
500///
501/// let handler = Thread::new("handler", 2048, 5, move || {
502/// loop {
503/// let mut cmd = Command::Stop;
504/// if queue_clone.fetch(&mut cmd, 1000).is_ok() {
505/// match cmd {
506/// Command::Start => { /* start operation */ },
507/// Command::Stop => { /* stop operation */ },
508/// Command::SetValue(val) => { /* set value */ },
509/// }
510/// }
511/// }
512/// }).unwrap();
513/// ```
514pub struct QueueStreamed<T: StructSerde> (Queue, PhantomData<T>);
515
516unsafe impl<T: StructSerde> Send for QueueStreamed<T> {}
517unsafe impl<T: StructSerde> Sync for QueueStreamed<T> {}
518
519impl<T> QueueStreamed<T>
520where
521 T: StructSerde {
522 /// Creates a new type-safe queue.
523 ///
524 /// # Parameters
525 ///
526 /// * `size` - Maximum number of messages
527 /// * `message_size` - Size of each message (typically `size_of::<T>()`)
528 ///
529 /// # Returns
530 ///
531 /// * `Ok(Self)` - Successfully created queue
532 /// * `Err(Error)` - Creation failed
533 #[inline]
534 pub fn new (size: UBaseType, message_size: UBaseType) -> Result<Self> {
535 Ok(Self (Queue::new(size, message_size)?, PhantomData))
536 }
537
538 /// Receives a typed message with a convertible timeout.
539 ///
540 /// This is a convenience method that accepts any type implementing `ToTick`.
541 ///
542 /// # Arguments
543 ///
544 /// * `buffer` - Mutable reference to receive the message into
545 /// * `time` - Timeout value (e.g., `Duration::from_millis(100)`)
546 ///
547 /// # Returns
548 ///
549 /// * `Ok(())` - Message successfully received and deserialized
550 /// * `Err(Error)` - Timeout or deserialization error
551 ///
552 /// # Examples
553 ///
554 /// ```ignore
555 /// use osal_rs::os::QueueStreamed;
556 /// use core::time::Duration;
557 ///
558 /// let queue: QueueStreamed<MyMessage> = QueueStreamed::new(5, size_of::<MyMessage>()).unwrap();
559 /// let mut msg = MyMessage::default();
560 /// queue.fetch_with_to_tick(&mut msg, Duration::from_millis(100))?;
561 /// ```
562 #[allow(dead_code)]
563 #[inline]
564 fn fetch_with_to_tick(&self, buffer: &mut T, time: impl ToTick) -> Result<()> {
565 self.fetch(buffer, time.to_ticks())
566 }
567
568 /// Sends a typed message with a convertible timeout.
569 ///
570 /// This is a convenience method that accepts any type implementing `ToTick`.
571 ///
572 /// # Arguments
573 ///
574 /// * `item` - Reference to the message to send
575 /// * `time` - Timeout value (e.g., `Duration::from_millis(100)`)
576 ///
577 /// # Returns
578 ///
579 /// * `Ok(())` - Message successfully serialized and sent
580 /// * `Err(Error)` - Timeout or serialization error
581 ///
582 /// # Examples
583 ///
584 /// ```ignore
585 /// use osal_rs::os::QueueStreamed;
586 /// use core::time::Duration;
587 ///
588 /// let queue: QueueStreamed<MyMessage> = QueueStreamed::new(5, size_of::<MyMessage>()).unwrap();
589 /// let msg = MyMessage { id: 1, value: 42 };
590 /// queue.post_with_to_tick(&msg, Duration::from_millis(100))?;
591 /// ```
592 #[inline]
593 #[allow(dead_code)]
594 fn post_with_to_tick(&self, item: &T, time: impl ToTick) -> Result<()> {
595 self.post(item, time.to_ticks())
596 }
597}
598
599#[cfg(not(feature = "serde"))]
600impl<T> QueueStreamedFn<T> for QueueStreamed<T>
601where
602 T: StructSerde {
603
604 /// Receives a typed message from the queue (without serde feature).
605 ///
606 /// Deserializes the message from bytes using the custom serialization traits.
607 ///
608 /// # Arguments
609 ///
610 /// * `buffer` - Mutable reference to receive the deserialized message
611 /// * `time` - Timeout in system ticks
612 ///
613 /// # Returns
614 ///
615 /// * `Ok(())` - Message successfully received and deserialized
616 /// * `Err(Error::Timeout)` - Queue empty or timeout
617 /// * `Err(Error)` - Deserialization error
618 fn fetch(&self, buffer: &mut T, time: TickType) -> Result<()> {
619 let mut buf_bytes = Vec::with_capacity(buffer.len());
620
621 if let Ok(()) = self.0.fetch(&mut buf_bytes, time) {
622 *buffer = T::from_bytes(&buf_bytes)?;
623 Ok(())
624 } else {
625 Err(Error::Timeout)
626 }
627 }
628
629 /// Receives a typed message from ISR context (without serde feature).
630 ///
631 /// ISR-safe version that does not block. Deserializes the message from bytes.
632 ///
633 /// # Arguments
634 ///
635 /// * `buffer` - Mutable reference to receive the deserialized message
636 ///
637 /// # Returns
638 ///
639 /// * `Ok(())` - Message successfully received and deserialized
640 /// * `Err(Error::Timeout)` - Queue is empty
641 /// * `Err(Error)` - Deserialization error
642 ///
643 /// # Safety
644 ///
645 /// Must only be called from ISR context.
646 fn fetch_from_isr(&self, buffer: &mut T) -> Result<()> {
647 let mut buf_bytes = Vec::with_capacity(buffer.len());
648
649 if let Ok(()) = self.0.fetch_from_isr(&mut buf_bytes) {
650 *buffer = T::from_bytes(&buf_bytes)?;
651 Ok(())
652 } else {
653 Err(Error::Timeout)
654 }
655 }
656
657 /// Sends a typed message to the queue (without serde feature).
658 ///
659 /// Serializes the message to bytes using the custom serialization traits.
660 ///
661 /// # Arguments
662 ///
663 /// * `item` - Reference to the message to send
664 /// * `time` - Timeout in system ticks
665 ///
666 /// # Returns
667 ///
668 /// * `Ok(())` - Message successfully serialized and sent
669 /// * `Err(Error::Timeout)` - Queue full
670 /// * `Err(Error)` - Serialization error
671 #[inline]
672 fn post(&self, item: &T, time: TickType) -> Result<()> {
673 self.0.post(&item.to_bytes(), time)
674 }
675
676 /// Sends a typed message from ISR context (without serde feature).
677 ///
678 /// ISR-safe version that does not block. Serializes the message to bytes.
679 ///
680 /// # Arguments
681 ///
682 /// * `item` - Reference to the message to send
683 ///
684 /// # Returns
685 ///
686 /// * `Ok(())` - Message successfully serialized and sent
687 /// * `Err(Error::Timeout)` - Queue is full
688 /// * `Err(Error)` - Serialization error
689 ///
690 /// # Safety
691 ///
692 /// Must only be called from ISR context.
693 #[inline]
694 fn post_from_isr(&self, item: &T) -> Result<()> {
695 self.0.post_from_isr(&item.to_bytes())
696 }
697
698 /// Deletes the typed queue.
699 ///
700 /// Delegates to the underlying byte queue's delete method.
701 #[inline]
702 fn delete(&mut self) {
703 self.0.delete()
704 }
705}
706
707#[cfg(feature = "serde")]
708impl<T> QueueStreamedFn<T> for QueueStreamed<T>
709where
710 T: StructSerde {
711
712 /// Receives a typed message from the queue (with serde feature).
713 ///
714 /// Deserializes the message from bytes using the serde framework.
715 ///
716 /// # Arguments
717 ///
718 /// * `buffer` - Mutable reference to receive the deserialized message
719 /// * `time` - Timeout in system ticks
720 ///
721 /// # Returns
722 ///
723 /// * `Ok(())` - Message successfully received and deserialized
724 /// * `Err(Error::Timeout)` - Queue empty or timeout
725 /// * `Err(Error::Unhandled)` - Deserialization error
726 fn fetch(&self, buffer: &mut T, time: TickType) -> Result<()> {
727 let mut buf_bytes = Vec::with_capacity(buffer.len());
728
729 if let Ok(()) = self.0.fetch(&mut buf_bytes, time) {
730
731 to_bytes(buffer, &mut buf_bytes).map_err(|_| Error::Unhandled("Deserializiation error"))?;
732
733 Ok(())
734 } else {
735 Err(Error::Timeout)
736 }
737 }
738
739 /// Receives a typed message from ISR context (with serde feature).
740 ///
741 /// ISR-safe version that does not block. Deserializes using serde.
742 ///
743 /// # Arguments
744 ///
745 /// * `buffer` - Mutable reference to receive the deserialized message
746 ///
747 /// # Returns
748 ///
749 /// * `Ok(())` - Message successfully received and deserialized
750 /// * `Err(Error::Timeout)` - Queue is empty
751 /// * `Err(Error::Unhandled)` - Deserialization error
752 ///
753 /// # Safety
754 ///
755 /// Must only be called from ISR context.
756 fn fetch_from_isr(&self, buffer: &mut T) -> Result<()> {
757 let mut buf_bytes = Vec::with_capacity(buffer.len());
758
759 if let Ok(()) = self.0.fetch_from_isr(&mut buf_bytes) {
760 to_bytes(buffer, &mut buf_bytes).map_err(|_| Error::Unhandled("Deserializiation error"))?;
761 Ok(())
762 } else {
763 Err(Error::Timeout)
764 }
765 }
766
767 /// Sends a typed message to the queue (with serde feature).
768 ///
769 /// Serializes the message to bytes using the serde framework.
770 ///
771 /// # Arguments
772 ///
773 /// * `item` - Reference to the message to send
774 /// * `time` - Timeout in system ticks
775 ///
776 /// # Returns
777 ///
778 /// * `Ok(())` - Message successfully serialized and sent
779 /// * `Err(Error::Timeout)` - Queue full
780 /// * `Err(Error::Unhandled)` - Serialization error
781 fn post(&self, item: &T, time: TickType) -> Result<()> {
782
783
784 let mut buf_bytes = Vec::with_capacity(item.len());
785
786 to_bytes(item, &mut buf_bytes).map_err(|_| Error::Unhandled("Deserializiation error"))?;
787
788 self.0.post(&buf_bytes, time)
789 }
790
791 /// Sends a typed message from ISR context (with serde feature).
792 ///
793 /// ISR-safe version that does not block. Serializes using serde.
794 ///
795 /// # Arguments
796 ///
797 /// * `item` - Reference to the message to send
798 ///
799 /// # Returns
800 ///
801 /// * `Ok(())` - Message successfully serialized and sent
802 /// * `Err(Error::Timeout)` - Queue is full
803 /// * `Err(Error::Unhandled)` - Serialization error
804 ///
805 /// # Safety
806 ///
807 /// Must only be called from ISR context.
808 fn post_from_isr(&self, item: &T) -> Result<()> {
809
810 let mut buf_bytes = Vec::with_capacity(item.len());
811
812 to_bytes(item, &mut buf_bytes).map_err(|_| Error::Unhandled("Deserializiation error"))?;
813
814 self.0.post_from_isr(&buf_bytes)
815 }
816
817 /// Deletes the typed queue (serde version).
818 ///
819 /// Delegates to the underlying byte queue's delete method.
820 #[inline]
821 fn delete(&mut self) {
822 self.0.delete()
823 }
824}
825
826/// Allows dereferencing to the underlying FreeRTOS queue handle.
827impl<T> Deref for QueueStreamed<T>
828where
829 T: StructSerde {
830 type Target = QueueHandle;
831
832 fn deref(&self) -> &Self::Target {
833 &self.0.0
834 }
835}
836
837/// Formats the typed queue for debugging purposes.
838impl<T> Debug for QueueStreamed<T>
839where
840 T: StructSerde {
841 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
842 f.debug_struct("QueueStreamed")
843 .field("handle", &self.0.0)
844 .finish()
845 }
846}
847
848/// Formats the typed queue for display purposes.
849impl<T> Display for QueueStreamed<T>
850where
851 T: StructSerde {
852 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
853 write!(f, "QueueStreamed {{ handle: {:?} }}", self.0.0)
854 }
855}
856