[ZEN-688] Add cancellation tokens support (#635)

* add cancellation tokens support

* Merge branch 'main' into fix-concurrent-cancel-ct

* support cancellation tokens for indirect callbacks

* address review comments

* fix is_cancelled evaluation

Author: DenisBiryukov91

Cargo.lock

@@ -0,0 +1,39 @@
+//
+// Copyright (c) 2025 ZettaScale Technology
+//
+// This program and the accompanying materials are made available under the
+// terms of the Eclipse Public License 2.0 which is available at
+// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+// which is available at https://www.apache.org/licenses/LICENSE-2.0.
+//
+// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+//
+// Contributors:
+//   ZettaScale Zenoh Team, <[email protected]>
+//
+use pyo3::prelude::*;
+
+use crate::{macros::wrapper, utils::wait};
+
+wrapper!(zenoh::cancellation::CancellationToken: Clone, Default);
+
+#[pymethods]
+impl CancellationToken {
+    #[new]
+    fn new() -> Self {
+        Self::default()
+    }
+
+    #[getter]
+    fn is_cancelled(&self) -> bool {
+        self.0.is_cancelled()
+    }
+
+    fn cancel(&self, py: Python) -> PyResult<()> {
+        wait(py, self.0.cancel())
+    }
+
+    fn __repr__(&self) -> PyResult<String> {
+        Ok(format!("{:?}", self.0))
+    }
+}

src/cancellation.rs

@@ -571,7 +571,7 @@ impl AdvancedSubscriber {
         py: Python,
         handler: Option<&Bound<PyAny>>,
     ) -> PyResult<SampleMissListener> {
-        let (handler, background) = into_handler(py, handler)?;
+        let (handler, background) = into_handler(py, handler, None)?;
         let builder = self.get_ref()?.sample_miss_listener();
         let mut listener = wait(py, builder.with(handler))?;
         if background {
@@ -587,7 +587,7 @@ impl AdvancedSubscriber {
         handler: Option<&Bound<PyAny>>,
         history: Option<bool>,
     ) -> PyResult<Subscriber> {
-        let (handler, background) = into_handler(py, handler)?;
+        let (handler, background) = into_handler(py, handler, None)?;
         let builder = build!(self.get_ref()?.detect_publishers(), history);
         let mut subscriber = wait(py, builder.with(handler))?;
         if background {
@@ -808,7 +808,7 @@ pub(crate) fn declare_advanced_subscriber(
     recovery: Option<RecoveryConfig>,
     subscriber_detection: Option<bool>,
 ) -> PyResult<AdvancedSubscriber> {
-    let (handler, background) = into_handler(py, handler)?;
+    let (handler, background) = into_handler(py, handler, None)?;
     let mut builder = build!(
         session.0.declare_subscriber(key_expr).advanced(),
         allowed_origin,