From 7fafebf690771a9286a1193ba17a489a2bdd65dd Mon Sep 17 00:00:00 2001 From: tayfuryldzz Date: Mon, 14 Sep 2026 17:09:07 +0300 Subject: [PATCH 1/2] feat(watch): allow non-Clone channel values Relax watch::channel construction so non-Clone values can be published, replaced, subscribed to, and observed for changes while keeping Clone requirements localized to owning reads. Add focused integration coverage and document the public contract. Signed-off-by: tayfuryldzz --- CHANGELOG.md | 1 + asyncband/src/watch/mod.rs | 4 +- .../tests/watch_non_clone_test.rs | 37 +++++++++++++++++++ 3 files changed, 41 insertions(+), 1 deletion(-) create mode 100644 tests-integration/tests/watch_non_clone_test.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 087014c..8487be5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,6 +39,7 @@ All notable changes to this project will be documented in this file. ### Improvements +* Allow `watch::channel` to store non-`Clone` values for publication and change notification; only owning reads through `Receiver::get` and `Receiver::recv` require `Clone`. * Finish releasing buffered bounded MPSC messages even if one message destructor panics. * Improve unbounded MPSC throughput with batched receiving and incremental storage reclamation; empty-buffer retention is bounded independently of previous peak occupancy. * Make completed and abandoned `Completion` waits lock-free while preserving cancellable pending registration. diff --git a/asyncband/src/watch/mod.rs b/asyncband/src/watch/mod.rs index 54b770a..2502034 100644 --- a/asyncband/src/watch/mod.rs +++ b/asyncband/src/watch/mod.rs @@ -81,6 +81,8 @@ use crate::internal::wakerset::WakerToken; /// Creates a watch channel with an initial value. /// /// The receiver returned by this function considers the initial value already observed. +/// The value type does not need to implement `Clone` for construction, publication, or change +/// notification. Only owning reads through [`Receiver::get`] and [`Receiver::recv`] require it. /// /// # Examples /// @@ -90,7 +92,7 @@ use crate::internal::wakerset::WakerToken; /// let (_tx, rx) = watch::channel("ready"); /// assert_eq!(rx.get(), "ready"); /// ``` -pub fn channel(initial: T) -> (Sender, Receiver) { +pub fn channel(initial: T) -> (Sender, Receiver) { let shared = Arc::new(Shared { state: Mutex::new(State { value: initial, diff --git a/tests-integration/tests/watch_non_clone_test.rs b/tests-integration/tests/watch_non_clone_test.rs new file mode 100644 index 0000000..b343047 --- /dev/null +++ b/tests-integration/tests/watch_non_clone_test.rs @@ -0,0 +1,37 @@ +// 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 asyncband::blocking::FutureExt; +use asyncband::watch; + +struct NonClone(usize); + +#[test] +fn non_clone_values_support_publication_and_change_tracking() { + let (tx, mut rx) = watch::channel(NonClone(0)); + + assert_eq!(rx.has_changed(), Ok(false)); + + tx.send(NonClone(1)).unwrap(); + assert_eq!(rx.has_changed(), Ok(true)); + FutureExt::block_on(rx.changed()).unwrap(); + assert_eq!(rx.has_changed(), Ok(false)); + + let previous = tx.send_replace(NonClone(2)); + assert_eq!(previous.0, 1); + assert_eq!(rx.has_changed(), Ok(true)); +} From a303df51ee15ee44247e3689995bce7793a0979d Mon Sep 17 00:00:00 2001 From: tayfuryldzz Date: Mon, 14 Sep 2026 18:29:21 +0300 Subject: [PATCH 2/2] test(watch): consolidate non-Clone coverage Signed-off-by: tayfuryldzz --- asyncband/src/watch/mod.rs | 2 - .../tests/watch_non_clone_test.rs | 37 ------------------- tests-integration/tests/watch_test.rs | 18 +++++++++ 3 files changed, 18 insertions(+), 39 deletions(-) delete mode 100644 tests-integration/tests/watch_non_clone_test.rs diff --git a/asyncband/src/watch/mod.rs b/asyncband/src/watch/mod.rs index 2502034..92aab71 100644 --- a/asyncband/src/watch/mod.rs +++ b/asyncband/src/watch/mod.rs @@ -81,8 +81,6 @@ use crate::internal::wakerset::WakerToken; /// Creates a watch channel with an initial value. /// /// The receiver returned by this function considers the initial value already observed. -/// The value type does not need to implement `Clone` for construction, publication, or change -/// notification. Only owning reads through [`Receiver::get`] and [`Receiver::recv`] require it. /// /// # Examples /// diff --git a/tests-integration/tests/watch_non_clone_test.rs b/tests-integration/tests/watch_non_clone_test.rs deleted file mode 100644 index b343047..0000000 --- a/tests-integration/tests/watch_non_clone_test.rs +++ /dev/null @@ -1,37 +0,0 @@ -// 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 asyncband::blocking::FutureExt; -use asyncband::watch; - -struct NonClone(usize); - -#[test] -fn non_clone_values_support_publication_and_change_tracking() { - let (tx, mut rx) = watch::channel(NonClone(0)); - - assert_eq!(rx.has_changed(), Ok(false)); - - tx.send(NonClone(1)).unwrap(); - assert_eq!(rx.has_changed(), Ok(true)); - FutureExt::block_on(rx.changed()).unwrap(); - assert_eq!(rx.has_changed(), Ok(false)); - - let previous = tx.send_replace(NonClone(2)); - assert_eq!(previous.0, 1); - assert_eq!(rx.has_changed(), Ok(true)); -} diff --git a/tests-integration/tests/watch_test.rs b/tests-integration/tests/watch_test.rs index 994eda0..2636e77 100644 --- a/tests-integration/tests/watch_test.rs +++ b/tests-integration/tests/watch_test.rs @@ -59,6 +59,8 @@ impl Clone for PanicOnceClone { } } +struct NonClone(usize); + #[test] fn initial_value_is_observed_and_updates_coalesce() { let (tx, mut rx) = watch::channel(0); @@ -147,6 +149,22 @@ fn subscriptions_start_at_the_current_version() { assert_eq!(FutureExt::block_on(subscribed.recv()).unwrap(), 2); } +#[test] +fn non_clone_values_support_publication_and_change_tracking() { + let (tx, mut rx) = watch::channel(NonClone(0)); + + assert_eq!(rx.has_changed(), Ok(false)); + + tx.send(NonClone(1)).unwrap(); + assert_eq!(rx.has_changed(), Ok(true)); + FutureExt::block_on(rx.changed()).unwrap(); + assert_eq!(rx.has_changed(), Ok(false)); + + let previous = tx.send_replace(NonClone(2)); + assert_eq!(previous.0, 1); + assert_eq!(rx.has_changed(), Ok(true)); +} + #[test] fn final_unseen_value_is_reported_before_disconnection() { let (tx, mut first) = watch::channel(0);