@@ -7,6 +7,24 @@ use jni::JNIEnv;
77use crate :: error:: SurrealError ;
88use crate :: { get_instance, new_string, take_instance, JniTypes , LiveStreamChannel , TOKIO_RUNTIME } ;
99
10+ /// JNI implementation of `LiveStream.nextNative(long handle)`.
11+ ///
12+ /// Blocks the calling thread until a live-query notification arrives or the
13+ /// stream ends. Returns a `LiveNotification` jobject, or `null` when the
14+ /// stream has been closed (by `releaseNative` or because the server ended the
15+ /// live query).
16+ ///
17+ /// ## Locking protocol
18+ ///
19+ /// 1. Acquire `recv_mutex` — serializes concurrent `nextNative` calls and
20+ /// lets `releaseNative` know when no thread is inside `recv()`.
21+ /// 2. Acquire `rx_mux` — borrows the channel receiver for the blocking call.
22+ /// 3. Call `block_on(rx_ref.recv())` — parks the thread until a message
23+ /// arrives or all senders are dropped (channel closed).
24+ ///
25+ /// Both guards are held for the duration of the `recv()`. This is safe
26+ /// because `releaseNative` only acquires these locks *after* the channel has
27+ /// been closed, guaranteeing `recv()` will have already returned.
1028#[ no_mangle]
1129pub extern "system" fn Java_com_surrealdb_LiveStream_nextNative < ' local > (
1230 mut env : JNIEnv < ' local > ,
@@ -16,23 +34,22 @@ pub extern "system" fn Java_com_surrealdb_LiveStream_nextNative<'local>(
1634 let ( recv_mutex, _join_handle_mux, _shutdown_tx_mux, rx_mux) =
1735 match get_instance :: < LiveStreamChannel > ( handle_ptr, JniTypes :: LiveStream ) {
1836 Ok ( r) => r,
19- Err ( e) => return e. exception ( & mut env, || std:: ptr:: null_mut ( ) ) ,
37+ Err ( e) => return e. exception ( & mut env, std:: ptr:: null_mut) ,
2038 } ;
2139 let _recv_guard = recv_mutex. lock ( ) ;
2240 let rx_opt_guard = rx_mux. lock ( ) ;
2341 let rx_ref = match rx_opt_guard. as_ref ( ) {
2442 Some ( rx) => rx,
25- None => return JObject :: null ( ) . into_raw ( ) , // already released
43+ None => return JObject :: null ( ) . into_raw ( ) ,
2644 } ;
2745 let item = match TOKIO_RUNTIME . block_on ( rx_ref. recv ( ) ) {
2846 Ok ( item) => item,
29- Err ( _) => return JObject :: null ( ) . into_raw ( ) , // channel closed
47+ Err ( _) => return JObject :: null ( ) . into_raw ( ) ,
3048 } ;
3149 let notification = match item {
3250 Ok ( n) => n,
33- Err ( e) => return SurrealError :: from ( e) . exception ( & mut env, || std:: ptr:: null_mut ( ) ) ,
51+ Err ( e) => return SurrealError :: from ( e) . exception ( & mut env, std:: ptr:: null_mut) ,
3452 } ;
35- // Build Java LiveNotification(action, valuePtr, queryId)
3653 let action_raw = new_string ! ( & mut env, notification. action. to_string( ) , || {
3754 std:: ptr:: null_mut( )
3855 } ) ;
@@ -44,7 +61,7 @@ pub extern "system" fn Java_com_surrealdb_LiveStream_nextNative<'local>(
4461 let query_id_str = unsafe { JObject :: from_raw ( query_id_raw) } ;
4562 let class = match env. find_class ( "com/surrealdb/LiveNotification" ) {
4663 Ok ( c) => c,
47- Err ( e) => return SurrealError :: from ( e) . exception ( & mut env, || std:: ptr:: null_mut ( ) ) ,
64+ Err ( e) => return SurrealError :: from ( e) . exception ( & mut env, std:: ptr:: null_mut) ,
4865 } ;
4966 let args = [
5067 JValue :: Object ( & action_str) ,
@@ -53,10 +70,31 @@ pub extern "system" fn Java_com_surrealdb_LiveStream_nextNative<'local>(
5370 ] ;
5471 match env. new_object ( class, "(Ljava/lang/String;JLjava/lang/String;)V" , & args) {
5572 Ok ( obj) => obj. into_raw ( ) ,
56- Err ( e) => SurrealError :: from ( e) . exception ( & mut env, || std:: ptr:: null_mut ( ) ) ,
73+ Err ( e) => SurrealError :: from ( e) . exception ( & mut env, std:: ptr:: null_mut) ,
5774 }
5875}
5976
77+ /// JNI implementation of `LiveStream.releaseNative(long handle)`.
78+ ///
79+ /// Shuts down the live query: stops the background thread, waits for any
80+ /// in-progress `nextNative` call to finish, then frees the native handle.
81+ ///
82+ /// ## Shutdown sequence
83+ ///
84+ /// 1. **Drop `shutdown_tx`** — the background thread's `tokio::select!` loop
85+ /// detects the closed shutdown channel and breaks. This also causes the
86+ /// background thread to drop its `tx_thread` sender, closing the
87+ /// notification channel.
88+ /// 2. **Join the background thread** — ensures `tx_thread` has been dropped
89+ /// and the channel is fully closed before proceeding.
90+ /// 3. **Acquire `recv_mutex`** — at this point the channel is closed, so any
91+ /// `nextNative` call blocked on `recv()` has already returned and released
92+ /// the mutex. Acquiring it here is a final safety barrier.
93+ /// 4. **Take the receiver** — drops it while holding `recv_mutex`, preventing
94+ /// any new `recv()` attempt.
95+ /// 5. **`take_instance`** — reclaims the boxed `LiveStreamChannel`, freeing
96+ /// the allocation. The Java side zeroes its `handle` field after this
97+ /// call returns, preventing further native access.
6098#[ no_mangle]
6199pub extern "system" fn Java_com_surrealdb_LiveStream_releaseNative < ' local > (
62100 _env : JNIEnv < ' local > ,
@@ -66,21 +104,17 @@ pub extern "system" fn Java_com_surrealdb_LiveStream_releaseNative<'local>(
66104 if handle_ptr == 0 {
67105 return ;
68106 }
69- // Do NOT take_instance yet: another thread may be in nextNative (get_instance + recv).
70- // First close the channel via the stored sender, join the background thread, then acquire
71- // the recv mutex (so no thread is in recv()), then take_instance and drop the receiver.
72107 let channel_ref = match get_instance :: < LiveStreamChannel > ( handle_ptr, JniTypes :: LiveStream ) {
73108 Ok ( r) => r,
74109 Err ( _) => return ,
75110 } ;
76111 let ( recv_mutex, join_handle_mux, shutdown_tx_mux, rx_mux) = channel_ref;
77- drop ( shutdown_tx_mux. lock ( ) . take ( ) ) ; // drop sender so background thread exits and channel closes
112+ drop ( shutdown_tx_mux. lock ( ) . take ( ) ) ;
78113 if let Some ( join_handle) = join_handle_mux. lock ( ) . take ( ) {
79114 let _ = join_handle. join ( ) ;
80115 }
81- let _recv_guard = recv_mutex. lock ( ) ; // wait until any thread in nextNative has left recv()
82- let _rx = rx_mux. lock ( ) . take ( ) ; // take and drop receiver while holding recv_guard
116+ let _recv_guard = recv_mutex. lock ( ) ;
117+ let _rx = rx_mux. lock ( ) . take ( ) ;
83118 drop ( _recv_guard) ;
84119 let _ = take_instance :: < LiveStreamChannel > ( handle_ptr, JniTypes :: LiveStream ) ;
85- // free the box
86120}
0 commit comments