laminar_db/ffi/
callback.rs

1//! Async FFI callbacks for push-based notifications.
2
3use std::ffi::{c_char, c_void, CStr, CString};
4use std::sync::atomic::{AtomicBool, Ordering};
5use std::sync::Arc;
6use std::thread::{self, JoinHandle};
7
8use super::connection::LaminarConnection;
9use super::error::{
10    clear_last_error, set_last_error, LAMINAR_ERR_INVALID_UTF8, LAMINAR_ERR_NULL_POINTER,
11    LAMINAR_OK,
12};
13use super::query::LaminarRecordBatch;
14
15/// Event type: insert (+1 weight).
16pub const LAMINAR_EVENT_INSERT: i32 = 0;
17/// Event type: delete (-1 weight).
18pub const LAMINAR_EVENT_DELETE: i32 = 1;
19/// Event type: update (delete + insert pair).
20pub const LAMINAR_EVENT_UPDATE: i32 = 2;
21/// Event type: watermark progress.
22pub const LAMINAR_EVENT_WATERMARK: i32 = 3;
23/// Event type: initial-state snapshot.
24pub const LAMINAR_EVENT_SNAPSHOT: i32 = 4;
25
26/// Callback invoked from the subscription thread with each batch.
27///
28/// # Safety
29///
30/// The callback receives ownership of `batch` and must free it with `laminar_batch_free`.
31pub type LaminarSubscriptionCallback = Option<
32    unsafe extern "C" fn(user_data: *mut c_void, batch: *mut LaminarRecordBatch, event_type: i32),
33>;
34
35/// Callback invoked when a subscription error occurs.
36pub type LaminarErrorCallback = Option<
37    unsafe extern "C" fn(user_data: *mut c_void, error_code: i32, error_message: *const c_char),
38>;
39
40/// Opaque handle for a callback-based subscription.
41///
42/// Created by `laminar_subscribe_callback`, freed by `laminar_subscription_free`.
43#[repr(C)]
44pub struct LaminarSubscriptionHandle {
45    cancelled: Arc<AtomicBool>,
46    thread_handle: Option<JoinHandle<()>>,
47    // Not owned; caller is responsible for lifetime.
48    user_data: *mut c_void,
49}
50
51// SAFETY: user_data thread-safety is the caller's responsibility.
52unsafe impl Send for LaminarSubscriptionHandle {}
53
54impl LaminarSubscriptionHandle {
55    fn new(
56        cancelled: Arc<AtomicBool>,
57        thread_handle: JoinHandle<()>,
58        user_data: *mut c_void,
59    ) -> Self {
60        Self {
61            cancelled,
62            thread_handle: Some(thread_handle),
63            user_data,
64        }
65    }
66
67    fn cancel(&mut self) {
68        self.cancelled.store(true, Ordering::SeqCst);
69        if let Some(handle) = self.thread_handle.take() {
70            let _ = handle.join();
71        }
72    }
73}
74
75struct CallbackContext {
76    user_data: *mut c_void,
77    on_data: LaminarSubscriptionCallback,
78    on_error: LaminarErrorCallback,
79    cancelled: Arc<AtomicBool>,
80}
81
82// SAFETY: only accessed from the subscription thread; user_data is caller's responsibility.
83unsafe impl Send for CallbackContext {}
84
85impl CallbackContext {
86    fn call_on_data(&self, batch: LaminarRecordBatch, event_type: i32) {
87        if let Some(callback) = self.on_data {
88            let batch_ptr = Box::into_raw(Box::new(batch));
89            unsafe { callback(self.user_data, batch_ptr, event_type) };
90        }
91    }
92
93    fn call_on_error(&self, error_code: i32, message: &str) {
94        if let Some(callback) = self.on_error {
95            let c_message = CString::new(message)
96                .unwrap_or_else(|_| CString::new("Error message contained null byte").unwrap());
97            unsafe { callback(self.user_data, error_code, c_message.as_ptr()) };
98        }
99    }
100
101    fn is_cancelled(&self) -> bool {
102        self.cancelled.load(Ordering::SeqCst)
103    }
104}
105
106/// Create a callback-based subscription; `on_data` is called from a background thread.
107///
108/// # Safety
109///
110/// `conn`, `query`, and `out` must be valid non-null pointers; `query` must be a
111/// null-terminated UTF-8 string; callbacks must be thread-safe if `user_data` is shared.
112#[no_mangle]
113pub unsafe extern "C" fn laminar_subscribe_callback(
114    conn: *mut LaminarConnection,
115    query: *const c_char,
116    on_data: LaminarSubscriptionCallback,
117    on_error: LaminarErrorCallback,
118    user_data: *mut c_void,
119    out: *mut *mut LaminarSubscriptionHandle,
120) -> i32 {
121    clear_last_error();
122
123    if conn.is_null() || query.is_null() || out.is_null() {
124        return LAMINAR_ERR_NULL_POINTER;
125    }
126
127    let Ok(query_str) = (unsafe { CStr::from_ptr(query) }).to_str() else {
128        return LAMINAR_ERR_INVALID_UTF8;
129    };
130
131    let conn_ref = unsafe { &(*conn).inner };
132
133    let stream = match conn_ref.query_stream(query_str) {
134        Ok(s) => s,
135        Err(e) => {
136            let code = e.code();
137            set_last_error(e);
138            return code;
139        }
140    };
141
142    let cancelled = Arc::new(AtomicBool::new(false));
143    let cancelled_clone = Arc::clone(&cancelled);
144
145    let ctx = CallbackContext {
146        user_data,
147        on_data,
148        on_error,
149        cancelled: cancelled_clone,
150    };
151
152    let thread_handle = thread::spawn(move || {
153        subscription_thread(stream, ctx);
154    });
155
156    let handle = Box::new(LaminarSubscriptionHandle::new(
157        cancelled,
158        thread_handle,
159        user_data,
160    ));
161
162    unsafe { *out = Box::into_raw(handle) };
163
164    LAMINAR_OK
165}
166
167#[allow(clippy::needless_pass_by_value)]
168fn subscription_thread(mut stream: crate::api::QueryStream, ctx: CallbackContext) {
169    loop {
170        if ctx.is_cancelled() {
171            break;
172        }
173
174        match stream.try_next() {
175            Ok(Some(batch)) => {
176                ctx.call_on_data(LaminarRecordBatch::new(batch), LAMINAR_EVENT_INSERT);
177            }
178            Ok(None) => {
179                if !stream.is_active() {
180                    break;
181                }
182                // Poll interval — avoid busy spin.
183                std::thread::sleep(std::time::Duration::from_millis(1));
184            }
185            Err(e) => {
186                ctx.call_on_error(e.code(), e.message());
187                if !stream.is_active() {
188                    break;
189                }
190            }
191        }
192    }
193}
194
195/// Cancel a callback-based subscription; no callbacks will fire after this returns.
196///
197/// # Safety
198///
199/// `handle` must be a valid subscription handle.
200#[no_mangle]
201pub unsafe extern "C" fn laminar_subscription_cancel(
202    handle: *mut LaminarSubscriptionHandle,
203) -> i32 {
204    clear_last_error();
205
206    if handle.is_null() {
207        return LAMINAR_ERR_NULL_POINTER;
208    }
209
210    let handle_ref = unsafe { &mut *handle };
211    handle_ref.cancel();
212
213    LAMINAR_OK
214}
215
216/// Check if a subscription is still active.
217///
218/// # Safety
219///
220/// `handle` and `out` must be valid non-null pointers.
221#[no_mangle]
222pub unsafe extern "C" fn laminar_subscription_is_active(
223    handle: *mut LaminarSubscriptionHandle,
224    out: *mut bool,
225) -> i32 {
226    clear_last_error();
227
228    if handle.is_null() || out.is_null() {
229        return LAMINAR_ERR_NULL_POINTER;
230    }
231
232    let handle_ref = unsafe { &*handle };
233    let active = !handle_ref.cancelled.load(Ordering::SeqCst) && handle_ref.thread_handle.is_some();
234
235    unsafe { *out = active };
236
237    LAMINAR_OK
238}
239
240/// Return the user data pointer from a subscription handle, or NULL.
241///
242/// # Safety
243///
244/// `handle` must be a valid subscription handle or NULL.
245#[no_mangle]
246pub unsafe extern "C" fn laminar_subscription_user_data(
247    handle: *mut LaminarSubscriptionHandle,
248) -> *mut c_void {
249    if handle.is_null() {
250        return std::ptr::null_mut();
251    }
252
253    unsafe { (*handle).user_data }
254}
255
256/// Free a subscription handle, cancelling it first if still active.
257///
258/// # Safety
259///
260/// `handle` must be a valid handle from `laminar_subscribe_callback`, or NULL.
261#[no_mangle]
262pub unsafe extern "C" fn laminar_subscription_free(handle: *mut LaminarSubscriptionHandle) {
263    if !handle.is_null() {
264        let mut boxed = unsafe { Box::from_raw(handle) };
265        boxed.cancel();
266        drop(boxed);
267    }
268}
269
270#[cfg(test)]
271#[allow(
272    clippy::borrow_as_ptr,
273    clippy::manual_c_str_literals,
274    clippy::items_after_statements
275)]
276mod tests {
277    use super::*;
278    use crate::ffi::connection::{laminar_close, laminar_execute, laminar_open};
279    use std::ptr;
280    use std::sync::atomic::AtomicUsize;
281
282    #[test]
283    fn test_event_type_constants() {
284        assert_eq!(LAMINAR_EVENT_INSERT, 0);
285        assert_eq!(LAMINAR_EVENT_DELETE, 1);
286        assert_eq!(LAMINAR_EVENT_UPDATE, 2);
287        assert_eq!(LAMINAR_EVENT_WATERMARK, 3);
288        assert_eq!(LAMINAR_EVENT_SNAPSHOT, 4);
289    }
290
291    #[test]
292    fn test_subscribe_null_pointer() {
293        let mut out: *mut LaminarSubscriptionHandle = ptr::null_mut();
294
295        // Null connection
296        let rc = unsafe {
297            laminar_subscribe_callback(
298                ptr::null_mut(),
299                b"SELECT 1\0".as_ptr().cast(),
300                None,
301                None,
302                ptr::null_mut(),
303                &mut out,
304            )
305        };
306        assert_eq!(rc, LAMINAR_ERR_NULL_POINTER);
307
308        // Null query
309        let mut conn: *mut LaminarConnection = ptr::null_mut();
310        unsafe { laminar_open(&mut conn) };
311
312        let rc = unsafe {
313            laminar_subscribe_callback(conn, ptr::null(), None, None, ptr::null_mut(), &mut out)
314        };
315        assert_eq!(rc, LAMINAR_ERR_NULL_POINTER);
316
317        // Null out
318        let rc = unsafe {
319            laminar_subscribe_callback(
320                conn,
321                b"SELECT 1\0".as_ptr().cast(),
322                None,
323                None,
324                ptr::null_mut(),
325                ptr::null_mut(),
326            )
327        };
328        assert_eq!(rc, LAMINAR_ERR_NULL_POINTER);
329
330        unsafe { laminar_close(conn) };
331    }
332
333    #[test]
334    fn test_subscription_cancel_null() {
335        let rc = unsafe { laminar_subscription_cancel(ptr::null_mut()) };
336        assert_eq!(rc, LAMINAR_ERR_NULL_POINTER);
337    }
338
339    #[test]
340    fn test_subscription_free_null() {
341        // Should not crash
342        unsafe { laminar_subscription_free(ptr::null_mut()) };
343    }
344
345    #[test]
346    fn test_subscription_user_data_null() {
347        let result = unsafe { laminar_subscription_user_data(ptr::null_mut()) };
348        assert!(result.is_null());
349    }
350
351    #[test]
352    fn test_subscribe_and_cancel() {
353        let mut conn: *mut LaminarConnection = ptr::null_mut();
354        let mut sub: *mut LaminarSubscriptionHandle = ptr::null_mut();
355
356        unsafe {
357            laminar_open(&mut conn);
358
359            // Create a table for querying
360            let sql = b"CREATE TABLE callback_test (id BIGINT PRIMARY KEY)\0";
361            laminar_execute(conn, sql.as_ptr().cast(), ptr::null_mut());
362
363            // Subscribe (no callbacks, just test lifecycle)
364            let query = b"SELECT * FROM callback_test\0";
365            let rc = laminar_subscribe_callback(
366                conn,
367                query.as_ptr().cast(),
368                None,
369                None,
370                ptr::null_mut(),
371                &mut sub,
372            );
373            assert_eq!(rc, LAMINAR_OK);
374            assert!(!sub.is_null());
375
376            // Check active
377            let mut active = false;
378            let rc = laminar_subscription_is_active(sub, &mut active);
379            assert_eq!(rc, LAMINAR_OK);
380            // Note: may be false if stream completed immediately
381
382            // Cancel
383            let rc = laminar_subscription_cancel(sub);
384            assert_eq!(rc, LAMINAR_OK);
385
386            // Should no longer be active
387            let rc = laminar_subscription_is_active(sub, &mut active);
388            assert_eq!(rc, LAMINAR_OK);
389            assert!(!active);
390
391            laminar_subscription_free(sub);
392            laminar_close(conn);
393        }
394    }
395
396    #[test]
397    fn test_subscribe_with_user_data() {
398        let mut conn: *mut LaminarConnection = ptr::null_mut();
399        let mut sub: *mut LaminarSubscriptionHandle = ptr::null_mut();
400
401        // Use a counter as user data
402        static COUNTER: AtomicUsize = AtomicUsize::new(42);
403
404        unsafe {
405            laminar_open(&mut conn);
406
407            let sql = b"CREATE TABLE userdata_test (id BIGINT PRIMARY KEY)\0";
408            laminar_execute(conn, sql.as_ptr().cast(), ptr::null_mut());
409
410            let query = b"SELECT * FROM userdata_test\0";
411            let user_data = std::ptr::addr_of!(COUNTER) as *mut c_void;
412
413            let rc = laminar_subscribe_callback(
414                conn,
415                query.as_ptr().cast(),
416                None,
417                None,
418                user_data,
419                &mut sub,
420            );
421            assert_eq!(rc, LAMINAR_OK);
422
423            // Verify user data is preserved
424            let retrieved = laminar_subscription_user_data(sub);
425            assert_eq!(retrieved, user_data);
426
427            laminar_subscription_cancel(sub);
428            laminar_subscription_free(sub);
429            laminar_close(conn);
430        }
431    }
432
433    // Static counters for callback tests
434    static DATA_CALLBACK_COUNT: AtomicUsize = AtomicUsize::new(0);
435    static ERROR_CALLBACK_COUNT: AtomicUsize = AtomicUsize::new(0);
436
437    unsafe extern "C" fn test_data_callback(
438        _user_data: *mut c_void,
439        batch: *mut LaminarRecordBatch,
440        _event_type: i32,
441    ) {
442        DATA_CALLBACK_COUNT.fetch_add(1, Ordering::SeqCst);
443        // Must free the batch
444        if !batch.is_null() {
445            crate::ffi::query::laminar_batch_free(batch);
446        }
447    }
448
449    unsafe extern "C" fn test_error_callback(
450        _user_data: *mut c_void,
451        _error_code: i32,
452        _error_message: *const c_char,
453    ) {
454        ERROR_CALLBACK_COUNT.fetch_add(1, Ordering::SeqCst);
455    }
456
457    #[test]
458    fn test_subscribe_with_callbacks() {
459        // Reset counters
460        DATA_CALLBACK_COUNT.store(0, Ordering::SeqCst);
461        ERROR_CALLBACK_COUNT.store(0, Ordering::SeqCst);
462
463        let mut conn: *mut LaminarConnection = ptr::null_mut();
464        let mut sub: *mut LaminarSubscriptionHandle = ptr::null_mut();
465
466        unsafe {
467            laminar_open(&mut conn);
468
469            let sql = b"CREATE TABLE callback_data_test (id BIGINT PRIMARY KEY)\0";
470            laminar_execute(conn, sql.as_ptr().cast(), ptr::null_mut());
471
472            let query = b"SELECT * FROM callback_data_test\0";
473            let rc = laminar_subscribe_callback(
474                conn,
475                query.as_ptr().cast(),
476                Some(test_data_callback),
477                Some(test_error_callback),
478                ptr::null_mut(),
479                &mut sub,
480            );
481            assert_eq!(rc, LAMINAR_OK);
482
483            // Give a moment for the subscription thread to run
484            std::thread::sleep(std::time::Duration::from_millis(50));
485
486            laminar_subscription_cancel(sub);
487            laminar_subscription_free(sub);
488            laminar_close(conn);
489        }
490
491        // Note: callbacks may or may not have fired depending on timing
492        // The important thing is no crashes occurred
493    }
494
495    #[test]
496    fn test_subscription_is_active_null_pointer() {
497        let mut active = true;
498        let rc = unsafe { laminar_subscription_is_active(ptr::null_mut(), &mut active) };
499        assert_eq!(rc, LAMINAR_ERR_NULL_POINTER);
500
501        let mut conn: *mut LaminarConnection = ptr::null_mut();
502        let mut sub: *mut LaminarSubscriptionHandle = ptr::null_mut();
503
504        unsafe {
505            laminar_open(&mut conn);
506            let sql = b"CREATE TABLE active_test (id BIGINT PRIMARY KEY)\0";
507            laminar_execute(conn, sql.as_ptr().cast(), ptr::null_mut());
508
509            let query = b"SELECT * FROM active_test\0";
510            laminar_subscribe_callback(
511                conn,
512                query.as_ptr().cast(),
513                None,
514                None,
515                ptr::null_mut(),
516                &mut sub,
517            );
518
519            // Null out pointer
520            let rc = laminar_subscription_is_active(sub, ptr::null_mut());
521            assert_eq!(rc, LAMINAR_ERR_NULL_POINTER);
522
523            laminar_subscription_cancel(sub);
524            laminar_subscription_free(sub);
525            laminar_close(conn);
526        }
527    }
528}
laminar_db/ffi/callback.rs

laminar_db/ffi/
callback.rs
