Add mark_unchanged method to watch::Receiver

tokio/src/sync/watch.rs

@@ -669,6 +669,26 @@ impl<T> Receiver<T> {
         self.version.decrement();
     }
 
+    /// Marks the state as unchanged.
+    ///
+    /// The current value will be considered seen by the receiver.
+    ///
+    /// After calling this method in a single-threaded context, a subsequent call to
+    /// [`has_changed()`](Self::has_changed) will return `false`, and a subsequent call
+    /// to [`changed()`](Self::changed) will yield.
+    ///
+    /// After calling this method in a multi-threaded context, subsequent
+    /// behaviour will vary depending on if a value is sent concurrently.
+    /// For example, a new value could be sent before this method returns
+    /// which would cause a call to [`changed`](Self::changed) to return immediately
+    ///
+    /// This is useful if you are not interested in the current value
+    /// visible in the receiver.
+    pub fn mark_unchanged(&mut self) {
+        let current_version = self.shared.state.load().version();
+        self.version = current_version;
+    }
+
     /// Waits for a change notification, then marks the newest value as seen.
     ///
     /// If the newest value in the channel has not yet been marked seen when

tokio/tests/sync_watch.rs

@@ -102,6 +102,39 @@ fn rx_mark_changed() {
     assert_eq!(*rx.borrow(), "two");
 }
 
+#[test]
+fn rx_mark_unchanged() {
+    let (tx, mut rx) = watch::channel("one");
+
+    let mut rx2 = rx.clone();
+
+    {
+        assert!(!rx.has_changed().unwrap());
+
+        rx.mark_changed();
+        assert!(rx.has_changed().unwrap());
+
+        rx.mark_unchanged();
+        assert!(!rx.has_changed().unwrap());
+
+        let mut t = spawn(rx.changed());
+        assert_pending!(t.poll());
+    }
+
+    {
+        assert!(!rx2.has_changed().unwrap());
+
+        tx.send("two").unwrap();
+        assert!(rx2.has_changed().unwrap());
+
+        rx2.mark_unchanged();
+        assert!(!rx2.has_changed().unwrap());
+        assert_eq!(*rx2.borrow_and_update(), "two");
+    }
+
+    assert_eq!(*rx.borrow(), "two");
+}
+
 #[test]
 fn multi_rx() {
     let (tx, mut rx1) = watch::channel("one");