From f92f46f489827ad9518d42c6bfa6d55fc6145bcf Mon Sep 17 00:00:00 2001
From: Dion Dokter <diondokter@gmail.com>
Date: Thu, 16 Jun 2022 14:19:16 +0200
Subject: [PATCH] Added convenience methods that ignore lag

---
 embassy/src/channel/pubsub.rs | 23 +++++++++++++++++++++++
 1 file changed, 23 insertions(+)

diff --git a/embassy/src/channel/pubsub.rs b/embassy/src/channel/pubsub.rs
index 4c360fea..02122580 100644
--- a/embassy/src/channel/pubsub.rs
+++ b/embassy/src/channel/pubsub.rs
@@ -274,6 +274,16 @@ impl<'a, T: Clone> Subscriber<'a, T> {
         SubscriberWaitFuture { subscriber: self }
     }
 
+    /// Wait for a published message (ignoring lag results)
+    pub async fn next_message_pure(&mut self) -> T {
+        loop {
+            match self.next_message().await {
+                WaitResult::Lagged(_) => continue,
+                WaitResult::Message(message) => break message,
+            }
+        }
+    }
+
     /// Try to see if there's a published message we haven't received yet.
     ///
     /// This function does not peek. The message is received if there is one.
@@ -289,6 +299,19 @@ impl<'a, T: Clone> Subscriber<'a, T> {
             }
         }
     }
+
+    /// Try to see if there's a published message we haven't received yet (ignoring lag results).
+    ///
+    /// This function does not peek. The message is received if there is one.
+    pub fn try_next_message_pure(&mut self) -> Option<T> {
+        loop {
+            match self.try_next_message() {
+                Some(WaitResult::Lagged(_)) => continue,
+                Some(WaitResult::Message(message)) => break Some(message),
+                None => break None,
+            }
+        }
+    }
 }
 
 impl<'a, T: Clone> Drop for Subscriber<'a, T> {