sync: add watch::Sender::send_modify method

tokio/src/sync/watch.rs

@@ -434,7 +434,7 @@ impl<T> Sender<T> {
             return Err(error::SendError(value));
         }
 
-        self.send_replace(value);
+        self.send_modify(|old| *old = value);
         Ok(())
     }
 
@@ -444,7 +444,15 @@ impl<T> Sender<T> {
     /// having to allocate a new instance. Additionally, this
     /// method permits sending values even when there are no receivers.
     ///
+    /// # Panics
+    ///
+    /// This function panics if calling `func` results in a panic.
+    /// No receivers are notified if panic occurred.
+    /// If the panic is caught, this might leave the watched value
+    /// in an inconsistent state.
+    ///
     /// # Examples
+    ///
     /// ```
     /// use tokio::sync::watch;
     ///

tokio/tests/sync_watch.rs

@@ -201,3 +201,30 @@ fn reopened_after_subscribe() {
     drop(rx);
     assert!(tx.is_closed());
 }
+
+#[test]
+fn send_modify_panic() {
+    let (tx, mut rx) = watch::channel("one");
+
+    tx.send_modify(|old| *old = "two");
+    assert_eq!(*rx.borrow_and_update(), "two");
+
+    let mut rx2 = rx.clone();
+    assert_eq!(*rx2.borrow_and_update(), "two");
+
+    let mut task = spawn(rx2.changed());
+
+    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+        tx.send_modify(|old| {
+            *old = "panicked";
+            panic!();
+        })
+    }));
+    assert!(result.is_err());
+
+    assert_pending!(task.poll());
+
+    tx.send_modify(|old| *old = "three");
+    assert_ready_ok!(task.poll());
+    assert_eq!(*rx.borrow_and_update(), "three");
+}