feat(watch): add a latest-state channel

CHANGELOG.md

@@ -12,6 +12,7 @@ All notable changes to this project will be documented in this file.
 * Add an opt-in `asyncband::blocking::FutureExt` bridge with `block_on` and `wait_timeout` methods for waiting on runtime-agnostic futures from synchronous code.
 * Add opt-in bounded and unbounded runtime-agnostic object pools under `asyncband::pool`.
 * Add opt-in `asyncband::once::LazyCell` for values that own one asynchronous initializer and preserve its in-flight future across caller cancellation.
+* Add an opt-in latest-state channel under `asyncband::watch`.
 
 ### Breaking changes
 

README.md

@@ -78,6 +78,7 @@ Runnable examples live in the [`examples`](examples) workspace crate. They demon
 | Channels              | [`oneshot`](https://docs.rs/asyncband/*/asyncband/oneshot/)                          | `oneshot`      | Send one value between two tasks.                                                                      |
 |                       | [`mpsc`](https://docs.rs/asyncband/*/asyncband/mpsc/)                                | `mpsc`         | Send each value from multiple producers to one receiver.                                               |
 |                       | [`broadcast`](https://docs.rs/asyncband/*/asyncband/broadcast/)                      | `broadcast`    | Broadcast values from one or more producers and retain them until every active receiver consumes them. |
+|                       | [`watch`](https://docs.rs/asyncband/*/asyncband/watch/)                              | `watch`        | Retain the latest state and coalesce intermediate updates.                                             |
 | Resource reuse        | [`pool`](https://docs.rs/asyncband/*/asyncband/pool/)                                | `pool`         | Reuse objects through bounded or unbounded pool variants.                                              |
 | Workload coordination | [`Semaphore`](https://docs.rs/asyncband/*/asyncband/semaphore/struct.Semaphore.html) | `semaphore`    | Control concurrent access with permits.                                                                |
 |                       | [`Group`](https://docs.rs/asyncband/*/asyncband/singleflight/struct.Group.html)      | `singleflight` | Coalesce concurrent calls for the same key.                                                            |

asyncband/Cargo.toml

@@ -62,6 +62,7 @@ semaphore = []
 shutdown = ["latch", "waitgroup"]
 singleflight = ["dep:hashbrown", "once-cell"]
 waitgroup = []
+watch = []
 
 [dependencies]
 hashbrown = { workspace = true, default-features = false, features = [

asyncband/src/channel/mod.rs

@@ -21,3 +21,5 @@ pub mod broadcast;
 pub mod mpsc;
 #[cfg(feature = "oneshot")]
 pub mod oneshot;
+#[cfg(feature = "watch")]
+pub mod watch;

asyncband/src/channel/watch/error.rs

@@ -0,0 +1,70 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::any::type_name;
+use std::fmt;
+
+/// An error returned when sending on a watch channel without any receivers.
+///
+/// The value that could not be sent can be retrieved with [`SendError::into_inner`].
+#[derive(Clone, PartialEq, Eq)]
+pub struct SendError<T>(T);
+
+impl<T> SendError<T> {
+    /// Returns a reference to the value that could not be sent.
+    pub fn as_inner(&self) -> &T {
+        &self.0
+    }
+
+    /// Consumes the error and returns the value that could not be sent.
+    pub fn into_inner(self) -> T {
+        self.0
+    }
+
+    pub(super) fn new(value: T) -> Self {
+        Self(value)
+    }
+}
+
+impl<T> fmt::Display for SendError<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str("sending on a closed channel")
+    }
+}
+
+impl<T> fmt::Debug for SendError<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "SendError<{}>(..)", type_name::<T>())
+    }
+}
+
+impl<T> std::error::Error for SendError<T> {}
+
+/// An error returned when every sender has disconnected and no unseen value remains.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RecvError {
+    /// Every sender has disconnected, so the current value will never change again.
+    Disconnected,
+}
+
+impl fmt::Display for RecvError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str("receiving on a closed channel")
+    }
+}
+
+impl std::error::Error for RecvError {}