rtic-sync: require channel-users to deal with non-empty channels

1 files changed, 58 insertions, 0 deletions

diff --git a/rtic-sync/src/channel.rs b/rtic-sync/src/channel.rs
index 3c0a27a..3706e58 100644
--- a/rtic-sync/src/channel.rs
+++ b/rtic-sync/src/channel.rs
@@ -102,10 +102,62 @@ impl<T, const N: usize> Channel<T, N> {
         }
     }
 
+    /// Clear any remaining items from this `Channel`.
+    pub fn clear(&mut self) {
+        for _ in self.queued_items() {}
+    }
+
+    /// Return an iterator over the still-queued items, removing them
+    /// from this channel.
+    pub fn queued_items(&mut self) -> impl Iterator<Item = T> + '_ {
+        struct Iter<'a, T, const N: usize> {
+            inner: &'a mut Channel<T, N>,
+        }
+
+        impl<T, const N: usize> Iterator for Iter<'_, T, N> {
+            type Item = T;
+
+            fn next(&mut self) -> Option<Self::Item> {
+                let slot = self.inner.readyq.as_mut().pop_back()?;
+
+                let value = unsafe {
+                    // SAFETY: `ready` is a valid slot.
+                    let first_element = self.inner.slots.get_unchecked(slot as usize).get_mut();
+                    let ptr = first_element.deref().as_ptr();
+                    // SAFETY: `ptr` points to an initialized `T`.
+                    core::ptr::read(ptr)
+                };
+
+                assert!(!self.inner.freeq.as_mut().is_full());
+                unsafe {
+                    // SAFETY: `freeq` is not ful.
+                    self.inner.freeq.as_mut().push_back_unchecked(slot);
+                }
+
+                Some(value)
+            }
+        }
+
+        Iter { inner: self }
+    }
+
     /// Split the queue into a `Sender`/`Receiver` pair.
+    ///
+    /// # Panics
+    /// This function panics if there are items in this channel while splitting.
+    ///
+    /// Call [`Channel::clear`] to clear all items from it, or [`Channel::queued_items`] to retrieve
+    /// an iterator that yields the values.
     pub fn split(&mut self) -> (Sender<'_, T, N>, Receiver<'_, T, N>) {
+        assert!(
+            self.readyq.as_mut().is_empty(),
+            "Cannot re-split non-empty queue. Call `Channel::clear()`."
+        );
+
         let freeq = self.freeq.as_mut();
 
+        freeq.clear();
+
         // Fill free queue
         for idx in 0..N as u8 {
             // NOTE(assert): `split`-ing does not put `freeq` into a known-empty
@@ -162,6 +214,12 @@ impl<T, const N: usize> Channel<T, N> {
     }
 }
 
+impl<T, const N: usize> Drop for Channel<T, N> {
+    fn drop(&mut self) {
+        self.clear();
+    }
+}
+
 /// Creates a split channel with `'static` lifetime.
 #[macro_export]
 #[cfg(not(loom))]