laminar_db/ffi/callback/
mod.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;
laminar_db/ffi/callback/mod.rs

laminar_db/ffi/callback/
mod.rs
