on_drop as parameter

milyin · milyin · commit 65458324e680 · 2026-04-25T11:43:03.000+02:00
diff --git a/zenoh-flat/src/jni_converter.rs b/zenoh-flat/src/jni_converter.rs
@@ -93,10 +93,6 @@ pub(crate) struct KotlinConfig {
     /// FQN of a singleton referenced inside the generated `init { ... }` block
     /// to force native-library loading — `None` disables the `init`.
     init_load_fqn: Option<String>,
-    /// FQN of the on-close callback type (typically
-    /// `io.zenoh.jni.callbacks.JNIOnCloseCallback`). Used for the synthetic
-    /// `<name>OnClose` parameter injected for every callback arg.
-    on_close_callback_fqn: String,
     /// Per-source-type Kotlin names (FQN or bare) for struct-decoded args.
     struct_kotlin_types: HashMap<String, String>,
     /// Per-element-type Kotlin names for callback args (e.g. `Sample` →
@@ -381,13 +377,6 @@ impl Builder {
         self
     }
 
-    /// FQN of the Kotlin on-close callback type used for the synthetic
-    /// `<name>OnClose` parameter injected for each callback argument.
-    pub fn kotlin_on_close(mut self, fqn: impl Into<String>) -> Self {
-        self.kotlin.get_or_insert_with(KotlinConfig::default).on_close_callback_fqn = fqn.into();
-        self
-    }
-
     pub fn build(self) -> JniConverter {
         JniConverter {
             cfg: self,
@@ -408,7 +397,6 @@ impl Default for KotlinConfig {
             class_name: String::new(),
             throws_class_fqn: None,
             init_load_fqn: None,
-            on_close_callback_fqn: String::new(),
             struct_kotlin_types: HashMap::new(),
             callback_kotlin_types: HashMap::new(),
             return_kotlin_types: HashMap::new(),
@@ -778,11 +766,9 @@ impl JniConverter {
                     }
                 }
                 ArgKind::Callback { decoder, element_type_name } => {
-                    let on_close_ident = format_ident!("{}_on_close", name);
                     jni_params.push(quote! { #name: jni::objects::JObject });
-                    jni_params.push(quote! { #on_close_ident: jni::objects::JObject });
                     prelude.push(quote! {
-                        let #name = #decoder(&mut env, #name, #on_close_ident)?;
+                        let #name = #decoder(&mut env, #name)?;
                     });
                     call_args.push(quote! { #name });
                     if let Some(kt) = kt_cfg {
@@ -792,17 +778,11 @@ impl JniConverter {
                                 element_type_name
                             ));
                         let cb_short = kotlin_register_fqn(&cb_fqn, &mut local_kotlin_fqns);
-                        let oc_short = kotlin_register_fqn(&kt.on_close_callback_fqn, &mut local_kotlin_fqns);
                         kotlin_params.push(format!(
                             "{}: {}",
                             kotlin_param_name(&name.to_string(), false),
                             cb_short
                         ));
-                        kotlin_params.push(format!(
-                            "{}OnClose: {}",
-                            kotlin_param_name(&name.to_string(), false),
-                            oc_short
-                        ));
                     }
                 }
                 ArgKind::OptionString => {
@@ -1209,7 +1189,7 @@ impl JniConverter {
                 }
             }
             syn::Type::ImplTrait(it) => {
-                if let Some(elem) = extract_fn_single_arg_type_name(&it.bounds) {
+                if let Some(elem) = extract_fn_arg_type_name(&it.bounds) {
                     if let Some(decoder) = self.cfg.callback_decoders.get(&elem) {
                         return ArgKind::Callback {
                             decoder: decoder.clone(),
@@ -1308,8 +1288,9 @@ enum ArgKind {
         decoder: syn::Path,
         type_name: String,
     },
-    /// `impl Fn(T) + Send + Sync + 'static` → `(JObject callback, JObject on_close)`
-    /// pair decoded via a callback decoder registered for `T`.
+    /// `impl Fn(T) + Send + Sync + 'static` → single `JObject` decoded via the
+    /// callback decoder registered for `T`. Zero-arg `impl Fn() + Send + Sync`
+    /// callbacks are looked up under the key `"()"`.
     Callback {
         decoder: syn::Path,
         element_type_name: String,
@@ -1335,9 +1316,11 @@ fn type_last_segment(ty: &syn::Type) -> Option<String> {
     tp.path.segments.last().map(|s| s.ident.to_string())
 }
 
-/// Look through the trait bounds of an `impl Fn(T) + ...` for a `Fn`-family
-/// trait and return the last-segment name of its single argument type `T`.
-fn extract_fn_single_arg_type_name(
+/// Look through the trait bounds of an `impl Fn(...) + ...` for a `Fn`-family
+/// trait and return the lookup key for its argument type:
+///   - `impl Fn(T)` → `Some("T")`
+///   - `impl Fn()`  → `Some("()")`  (zero-arg callbacks, e.g. on-close handlers)
+fn extract_fn_arg_type_name(
     bounds: &syn::punctuated::Punctuated<syn::TypeParamBound, syn::Token![+]>,
 ) -> Option<String> {
     for bound in bounds {
@@ -1347,8 +1330,10 @@ fn extract_fn_single_arg_type_name(
             continue;
         }
         let syn::PathArguments::Parenthesized(p) = &seg.arguments else { continue };
-        let first = p.inputs.first()?;
-        return type_last_segment(first);
+        return match p.inputs.first() {
+            Some(first) => type_last_segment(first),
+            None => Some("()".to_string()),
+        };
     }
     None
 }
diff --git a/zenoh-flat/src/session.rs b/zenoh-flat/src/session.rs
@@ -35,6 +35,23 @@ use zenoh_ext::{
     AdvancedSubscriberBuilderExt,
 };
 
+/// Fires `f` exactly once when dropped. Used to bind an `on_close` callback
+/// to the lifetime of a data callback closure: when zenoh drops the data
+/// closure (subscription/queryable/get teardown), the guard's `Drop` runs
+/// `on_close`.
+struct CallOnDrop<F: FnOnce()>(core::mem::MaybeUninit<F>);
+impl<F: FnOnce()> CallOnDrop<F> {
+    fn new(f: F) -> Self {
+        Self(core::mem::MaybeUninit::new(f))
+    }
+}
+impl<F: FnOnce()> Drop for CallOnDrop<F> {
+    fn drop(&mut self) {
+        let f = unsafe { self.0.assume_init_read() };
+        f();
+    }
+}
+
 /// Open a Zenoh session using a borrowed configuration.
 #[prebindgen_proc_macro::prebindgen("jni")]
 pub fn open_session(config: &Config) -> ZResult<Session> {
@@ -134,11 +151,16 @@ pub fn declare_subscriber(
     session: &Session,
     key_expr: KeyExpr<'static>,
     callback: impl Fn(Sample) + Send + Sync + 'static,
+    on_close: impl Fn() + Send + Sync + 'static,
 ) -> ZResult<Subscriber<()>> {
     let key_expr_string = key_expr.to_string();
+    let guard = CallOnDrop::new(on_close);
     session
         .declare_subscriber(key_expr)
-        .callback(callback)
+        .callback(move |sample| {
+            let _ = &guard;
+            callback(sample);
+        })
         .wait()
         .map(|subscriber| {
             trace!("Declared subscriber on '{}'.", key_expr_string);
@@ -190,12 +212,17 @@ pub fn declare_queryable(
     session: &Session,
     key_expr: KeyExpr<'static>,
     callback: impl Fn(Query) + Send + Sync + 'static,
+    on_close: impl Fn() + Send + Sync + 'static,
     complete: bool,
 ) -> ZResult<Queryable<()>> {
     let key_expr_string = key_expr.to_string();
+    let guard = CallOnDrop::new(on_close);
     session
         .declare_queryable(key_expr)
-        .callback(callback)
+        .callback(move |query| {
+            let _ = &guard;
+            callback(query);
+        })
         .complete(complete)
         .wait()
         .map(|queryable| {
@@ -224,6 +251,7 @@ pub fn get(
     key_expr: KeyExpr<'static>,
     selector_params: Option<String>,
     callback: impl Fn(Reply) + Send + Sync + 'static,
+    on_close: impl Fn() + Send + Sync + 'static,
     timeout: Duration,
     query_target: QueryTarget,
     consolidation: ConsolidationMode,
@@ -237,9 +265,13 @@ pub fn get(
 ) -> ZResult<()> {
     let key_expr_string = key_expr.to_string();
     let selector = Selector::owned(&key_expr, selector_params.unwrap_or_default());
+    let guard = CallOnDrop::new(on_close);
     let mut get_builder = session
         .get(selector)
-        .callback(callback)
+        .callback(move |reply| {
+            let _ = &guard;
+            callback(reply);
+        })
         .target(query_target)
         .consolidation(consolidation)
         .congestion_control(congestion_control)
@@ -386,14 +418,19 @@ pub fn declare_advanced_subscriber(
     session: &Session,
     key_expr: KeyExpr<'static>,
     callback: impl Fn(Sample) + Send + Sync + 'static,
+    on_close: impl Fn() + Send + Sync + 'static,
     history: Option<HistoryConfig>,
     recovery: Option<RecoveryConfig>,
     subscriber_detection: bool,
 ) -> ZResult<AdvancedSubscriber<()>> {
     let key_expr_string = key_expr.to_string();
+    let guard = CallOnDrop::new(on_close);
     let mut builder = session
         .declare_subscriber(key_expr)
-        .callback(callback)
+        .callback(move |sample| {
+            let _ = &guard;
+            callback(sample);
+        })
         .advanced();
     if let Some(history) = history {
         builder = builder.history(history.try_into()?);
diff --git a/zenoh-jni/build.rs b/zenoh-jni/build.rs
@@ -18,7 +18,6 @@ fn main() {
         .kotlin_class("JNISessionNative")
         .kotlin_throws("io.zenoh.exceptions.ZError")
         .kotlin_init("io.zenoh.ZenohLoad")
-        .kotlin_on_close("io.zenoh.jni.callbacks.JNIOnCloseCallback")
         .enum_decoder(
             "CongestionControl",
             "crate::utils::decode_congestion_control",
@@ -43,6 +42,11 @@ fn main() {
             "crate::sample_callback::process_kotlin_reply_callback",
             "io.zenoh.jni.callbacks.JNIGetCallback",
         )
+        .callback_decoder(
+            "()",
+            "crate::sample_callback::process_kotlin_on_close_callback",
+            "io.zenoh.jni.callbacks.JNIOnCloseCallback",
+        )
         .struct_decoder(
             "KeyExpr",
             "crate::key_expr::decode_jni_key_expr",
diff --git a/zenoh-jni/src/ext/advanced_subscriber.rs b/zenoh-jni/src/ext/advanced_subscriber.rs
@@ -30,7 +30,7 @@ use zenoh::Wait;
 
 use crate::owned_object::OwnedObject;
 
-use crate::utils::{get_callback_global_ref, get_java_vm, load_on_close};
+use crate::utils::{get_callback_global_ref, get_java_vm, load_on_close, wrap_with_on_close};
 use std::ptr::null;
 
 use crate::throw_exception;
@@ -138,12 +138,12 @@ pub unsafe extern "C" fn Java_io_zenoh_jni_JNIAdvancedSubscriber_declareDetectPu
             advanced_subscriber.key_expr()
         );
 
+        let cb = process_kotlin_sample_callback(&mut env, callback)?;
+        let cb = wrap_with_on_close(&mut env, on_close, cb)?;
         let detect_publishers_subscriber = advanced_subscriber
             .detect_publishers()
             .history(history != 0)
-            .callback(process_kotlin_sample_callback(
-                &mut env, callback, on_close,
-            )?)
+            .callback(cb)
             .wait()
             .map_err(|err| zerror!("Unable to declare detect publishers subscriber: {}", err))?;
 
@@ -196,12 +196,12 @@ pub unsafe extern "C" fn Java_io_zenoh_jni_JNIAdvancedSubscriber_declareBackgrou
             advanced_subscriber.key_expr()
         );
 
+        let cb = process_kotlin_sample_callback(&mut env, callback)?;
+        let cb = wrap_with_on_close(&mut env, on_close, cb)?;
         advanced_subscriber
             .detect_publishers()
             .history(history != 0)
-            .callback(process_kotlin_sample_callback(
-                &mut env, callback, on_close,
-            )?)
+            .callback(cb)
             .background()
             .wait()
             .map_err(|err| {
diff --git a/zenoh-jni/src/liveliness.rs b/zenoh-jni/src/liveliness.rs
@@ -30,7 +30,7 @@ use crate::{
     owned_object::OwnedObject,
     sample_callback::{on_reply_error, on_reply_success, process_kotlin_sample_callback},
     throw_exception,
-    utils::{get_callback_global_ref, get_java_vm, load_on_close},
+    utils::{get_callback_global_ref, get_java_vm, load_on_close, wrap_with_on_close},
 };
 
 #[no_mangle]
@@ -146,13 +146,13 @@ pub unsafe extern "C" fn Java_io_zenoh_jni_JNISession_declareLivelinessSubscribe
         let key_expr = decode_jni_key_expr(&mut env, &key_expr)?;
         tracing::debug!("Declaring liveliness subscriber on '{}'...", key_expr);
 
+        let cb = process_kotlin_sample_callback(&mut env, callback)?;
+        let cb = wrap_with_on_close(&mut env, on_close, cb)?;
         let subscriber = session
             .liveliness()
             .declare_subscriber(key_expr.to_owned())
             .history(history != 0)
-            .callback(process_kotlin_sample_callback(
-                &mut env, callback, on_close,
-            )?)
+            .callback(cb)
             .wait()
             .map_err(|err| zerror!("Unable to declare liveliness subscriber: {}", err))?;
 
diff --git a/zenoh-jni/src/sample_callback.rs b/zenoh-jni/src/sample_callback.rs
@@ -177,15 +177,11 @@ pub(crate) fn on_reply_error(
 pub(crate) unsafe fn process_kotlin_sample_callback(
     env: &mut JNIEnv,
     callback: JObject,
-    on_close: JObject,
 ) -> ZResult<impl Fn(Sample) + Send + Sync + 'static> {
     let java_vm = Arc::new(get_java_vm(env)?);
     let callback_global_ref = get_callback_global_ref(env, callback)?;
-    let on_close_global_ref = get_callback_global_ref(env, on_close)?;
-    let on_close = load_on_close(&java_vm, on_close_global_ref);
 
     Ok(move |sample: Sample| {
-        on_close.noop();
         let _ = || -> ZResult<()> {
             let mut env = java_vm
                 .attach_current_thread_as_daemon()
@@ -250,16 +246,12 @@ pub(crate) unsafe fn process_kotlin_sample_callback(
 pub(crate) unsafe fn process_kotlin_reply_callback(
     env: &mut JNIEnv,
     callback: JObject,
-    on_close: JObject,
 ) -> ZResult<impl Fn(Reply) + Send + Sync + 'static> {
     let java_vm = Arc::new(get_java_vm(env)?);
     let callback_global_ref = get_callback_global_ref(env, callback)?;
-    let on_close_global_ref = get_callback_global_ref(env, on_close)?;
-    let on_close = load_on_close(&java_vm, on_close_global_ref);
 
     Ok(move |reply: Reply| {
         || -> ZResult<()> {
-            on_close.noop();
             tracing::debug!("Receiving reply through JNI: {:?}", reply);
             let mut env = java_vm.attach_current_thread_as_daemon().map_err(|err| {
                 zerror!("Unable to attach thread for GET query callback: {}", err)
@@ -286,15 +278,11 @@ pub(crate) unsafe fn process_kotlin_reply_callback(
 pub(crate) unsafe fn process_kotlin_query_callback(
     env: &mut JNIEnv,
     callback: JObject,
-    on_close: JObject,
 ) -> ZResult<impl Fn(Query) + Send + Sync + 'static> {
     let java_vm = Arc::new(get_java_vm(env)?);
     let callback_global_ref = get_callback_global_ref(env, callback)?;
-    let on_close_global_ref = get_callback_global_ref(env, on_close)?;
-    let on_close = load_on_close(&java_vm, on_close_global_ref);
 
     Ok(move |query: Query| {
-        on_close.noop();
         let env = match java_vm.attach_current_thread_as_daemon() {
             Ok(env) => env,
             Err(err) => {
@@ -311,6 +299,33 @@ pub(crate) unsafe fn process_kotlin_query_callback(
     })
 }
 
+/// Decoder for the zero-arg `impl Fn() + Send + Sync + 'static` callback used
+/// by zenoh-flat for `on_close` parameters. The returned closure attaches the
+/// JVM and invokes `Runnable.run()` on the Kotlin object. zenoh-flat is
+/// responsible for binding it to the data closure's lifetime via its own
+/// `CallOnDrop` so the on-close fires when the subscription/query is dropped.
+pub(crate) unsafe fn process_kotlin_on_close_callback(
+    env: &mut JNIEnv,
+    on_close: JObject,
+) -> ZResult<impl Fn() + Send + Sync + 'static> {
+    let java_vm = Arc::new(get_java_vm(env)?);
+    let on_close_global_ref = get_callback_global_ref(env, on_close)?;
+
+    Ok(move || {
+        let mut env = match java_vm.attach_current_thread_as_daemon() {
+            Ok(env) => env,
+            Err(err) => {
+                tracing::error!("Unable to attach thread for 'onClose' callback: {}", err);
+                return;
+            }
+        };
+        if let Err(err) = env.call_method(&on_close_global_ref, "run", "()V", &[]) {
+            _ = env.exception_describe();
+            tracing::error!("Error while running 'onClose' callback: {}", err);
+        }
+    })
+}
+
 fn on_query(mut env: JNIEnv, query: Query, callback_global_ref: &GlobalRef) -> ZResult<()> {
     let selector_params_jstr = env
         .new_string(query.parameters().to_string())
diff --git a/zenoh-jni/src/utils.rs b/zenoh-jni/src/utils.rs
