Added remove_if to priority channel

embassy-sync/src/priority_channel.rs

@@ -72,6 +72,17 @@ where
         self.channel.poll_ready_to_send(cx)
     }
 
+    /// Removes the elements from the channel that satisfy the predicate.
+    ///
+    /// See [`PriorityChannel::remove_if()`]
+    pub fn remove_if<F>(&self, predicate: F)
+    where
+        F: Fn(&T) -> bool,
+        T: Clone,
+    {
+        self.channel.remove_if(predicate)
+    }
+
     /// Returns the maximum number of elements the channel can hold.
     ///
     /// See [`PriorityChannel::capacity()`]
@@ -189,6 +200,17 @@ where
         self.channel.poll_receive(cx)
     }
 
+    /// Removes the elements from the channel that satisfy the predicate.
+    ///
+    /// See [`PriorityChannel::remove_if()`]
+    pub fn remove_if<F>(&self, predicate: F)
+    where
+        F: Fn(&T) -> bool,
+        T: Clone,
+    {
+        self.channel.remove_if(predicate)
+    }
+
     /// Returns the maximum number of elements the channel can hold.
     ///
     /// See [`PriorityChannel::capacity()`]
@@ -534,6 +556,26 @@ where
         self.lock(|c| c.try_receive())
     }
 
+    /// Removes elements from the channel based on the given predicate.
+    pub fn remove_if<F>(&self, predicate: F)
+    where
+        F: Fn(&T) -> bool,
+        T: Clone,
+    {
+        self.lock(|c| {
+            let mut new_heap = BinaryHeap::<T, K, N>::new();
+            for item in c.queue.iter() {
+                if !predicate(item) {
+                    match new_heap.push(item.clone()) {
+                        Ok(_) => (),
+                        Err(_) => panic!("Error pushing item to heap"),
+                    }
+                }
+            }
+            c.queue = new_heap;
+        });
+    }
+
     /// Returns the maximum number of elements the channel can hold.
     pub const fn capacity(&self) -> usize {
         N