rusteron_media_driver/
lib.rs

1#![allow(improper_ctypes_definitions)]
2#![allow(non_upper_case_globals)]
3#![allow(non_camel_case_types)]
4#![allow(non_snake_case)]
5#![allow(clippy::all)]
6#![allow(unused_unsafe)]
7#![allow(unused_variables)]
8#![allow(unsafe_op_in_unsafe_fn)]
9#![doc = include_str!("../README.md")]
10//! # Features
11//!
12//! - **`static`**: When enabled, this feature statically links the Aeron C code.
13//!   By default, the library uses dynamic linking to the Aeron C libraries.
14//! - **`backtrace`**: When enabled will log a backtrace for each AeronCError
15//! - **`extra-logging`**: When enabled will log when resource is created and destroyed. Useful if you're seeing a segfault due to a resource being closed
16//! - **`log-c-bindings`**: When enabled will log every C binding call with arguments and return values. Useful for debugging FFI interactions
17//! - **`precompile`**: When enabled will use precompiled C code instead of requiring cmake and java to be installed
18
19#[allow(improper_ctypes_definitions)]
20#[allow(unpredictable_function_pointer_comparisons)]
21pub mod bindings {
22    include!(concat!(env!("OUT_DIR"), "/bindings.rs"));
23}
24
25use bindings::*;
26use log::info;
27use std::path::Path;
28use std::sync::Arc;
29use std::sync::atomic::{AtomicBool, Ordering};
30use std::thread::{JoinHandle, sleep};
31use std::time::Duration;
32
33include!(concat!(env!("OUT_DIR"), "/aeron.rs"));
34include!(concat!(env!("OUT_DIR"), "/aeron_custom.rs"));
35
36unsafe impl Sync for AeronDriverContext {}
37unsafe impl Send for AeronDriverContext {}
38unsafe impl Sync for AeronDriver {}
39unsafe impl Send for AeronDriver {}
40
41/// RAII guard for an embedded media driver launched via
42/// [`AeronDriver::launch_embedded_guard`]. Signals stop on drop so the driver
43/// thread always joins even on panic / early return.
44pub struct EmbeddedMediaDriver {
45    stop: Option<Arc<AtomicBool>>,
46    handle: Option<JoinHandle<Result<(), AeronCError>>>,
47}
48
49impl EmbeddedMediaDriver {
50    /// Signal the driver thread to stop (idempotent; the thread exits on its next
51    /// idle cycle). Blocks until it joins when consumed by [`Self::join`].
52    pub fn stop(&self) {
53        if let Some(stop) = &self.stop {
54            stop.store(true, Ordering::SeqCst);
55        }
56    }
57
58    /// Whether the driver thread has finished.
59    pub fn is_finished(&self) -> bool {
60        self.handle.as_ref().map_or(true, |h| h.is_finished())
61    }
62
63    /// Signal stop and block until the driver thread joins, returning its result.
64    pub fn join(mut self) -> Result<(), AeronCError> {
65        self.stop();
66        if let Some(h) = self.handle.take() {
67            // Flatten JoinHandle<Result<..>>: a panic becomes an AeronCError,
68            // the inner AeronCError is propagated.
69            return h.join().map_err(|_| AeronCError::from_code(-1)).and_then(|r| r);
70        }
71        Ok(())
72    }
73}
74
75impl Drop for EmbeddedMediaDriver {
76    fn drop(&mut self) {
77        if let Some(stop) = &self.stop {
78            stop.store(true, Ordering::SeqCst);
79        }
80        if let Some(h) = self.handle.take() {
81            let _ = h.join();
82        }
83    }
84}
85
86pub mod testing;
87
88impl AeronDriverContext {
89    /// Typed variant of [`Self::set_conductor_idle_strategy`].
90    pub fn set_conductor_idle_strategy_kind(&self, kind: AeronIdleStrategyKind) -> Result<i32, AeronCError> {
91        self.set_conductor_idle_strategy(kind.name_c())
92    }
93
94    /// Typed variant of [`Self::set_sender_idle_strategy`].
95    pub fn set_sender_idle_strategy_kind(&self, kind: AeronIdleStrategyKind) -> Result<i32, AeronCError> {
96        self.set_sender_idle_strategy(kind.name_c())
97    }
98
99    /// Typed variant of [`Self::set_receiver_idle_strategy`].
100    pub fn set_receiver_idle_strategy_kind(&self, kind: AeronIdleStrategyKind) -> Result<i32, AeronCError> {
101        self.set_receiver_idle_strategy(kind.name_c())
102    }
103
104    /// Typed variant of [`Self::set_sharednetwork_idle_strategy`].
105    pub fn set_sharednetwork_idle_strategy_kind(&self, kind: AeronIdleStrategyKind) -> Result<i32, AeronCError> {
106        self.set_sharednetwork_idle_strategy(kind.name_c())
107    }
108
109    /// Typed variant of [`Self::set_shared_idle_strategy`].
110    pub fn set_shared_idle_strategy_kind(&self, kind: AeronIdleStrategyKind) -> Result<i32, AeronCError> {
111        self.set_shared_idle_strategy(kind.name_c())
112    }
113}
114
115impl AeronDriver {
116    pub fn launch_embedded(
117        aeron_context: AeronDriverContext,
118        register_sigint: bool,
119    ) -> (Arc<AtomicBool>, JoinHandle<Result<(), AeronCError>>) {
120        AeronDriver::wait_for_previous_media_driver_to_timeout(&aeron_context);
121
122        let stop = Arc::new(AtomicBool::new(false));
123        let stop_copy = stop.clone();
124        // Register signal handler for SIGINT (Ctrl+C)
125        if register_sigint {
126            let stop_copy2 = stop.clone();
127            ctrlc::set_handler(move || {
128                stop_copy2.store(true, Ordering::SeqCst);
129            })
130            .expect("Error setting Ctrl-C handler");
131        }
132
133        let started = Arc::new(AtomicBool::new(false));
134        let started2 = started.clone();
135
136        let dir = aeron_context.get_dir().to_string();
137        info!("Starting media driver [dir={}]", dir);
138        let handle = std::thread::spawn(move || {
139            let aeron_context = aeron_context.clone();
140            let aeron_driver = AeronDriver::new(&aeron_context)?;
141            aeron_driver.start(true)?;
142
143            info!("Aeron driver started [dir={}]", aeron_driver.context().get_dir());
144
145            started2.store(true, Ordering::SeqCst);
146
147            // Poll for work until Ctrl+C is pressed
148            while !stop.load(Ordering::Acquire) {
149                aeron_driver.main_idle_strategy(aeron_driver.main_do_work()?);
150            }
151
152            info!("stopping media driver");
153
154            Ok::<_, AeronCError>(())
155        });
156
157        while !started.load(Ordering::SeqCst) && !handle.is_finished() {
158            sleep(Duration::from_millis(100));
159        }
160
161        if handle.is_finished() {
162            panic!("failed to start media driver {:?}", handle.join())
163        }
164        info!("started media driver [dir={}]", dir);
165
166        (stop_copy, handle)
167    }
168
169    /// Launch an embedded media driver, returning a RAII guard that stops the
170    /// driver on drop (so a panic or early return can't leak a driver process).
171    ///
172    /// Prefer this over [`Self::launch_embedded`], which returns a raw
173    /// `(Arc<AtomicBool>, JoinHandle)` tuple the caller must remember to drive.
174    /// `register_sigint` is `true` to mirror the standalone binary.
175    pub fn launch_embedded_guard(aeron_context: AeronDriverContext, register_sigint: bool) -> EmbeddedMediaDriver {
176        let (stop, handle) = Self::launch_embedded(aeron_context, register_sigint);
177        EmbeddedMediaDriver {
178            stop: Some(stop),
179            handle: Some(handle),
180        }
181    }
182
183    /// if you have existing shm files and its before the driver timeout it will try to reuse it and fail
184    /// this makes sure that if that is the case it will wait else it proceeds
185    pub fn wait_for_previous_media_driver_to_timeout(aeron_context: &AeronDriverContext) {
186        if !aeron_context.get_dir_delete_on_start() {
187            let cnc_file = Path::new(aeron_context.get_dir()).join("cnc.dat");
188
189            if cnc_file.exists() {
190                let timeout = Duration::from_millis(aeron_context.get_driver_timeout_ms() * 2).as_nanos() as i64;
191
192                let mut duration = timeout;
193
194                if let Ok(md) = cnc_file.metadata() {
195                    if let Ok(modified_time) = md.modified() {
196                        if let Ok(took) = modified_time.elapsed() {
197                            duration = took.as_nanos() as i64;
198                        }
199                    }
200                }
201
202                let delay = timeout - duration;
203
204                if delay > 0 {
205                    let sleep_duration = Duration::from_nanos((delay + 1_000_000) as u64);
206                    info!(
207                        "cnc file already exists, will need to wait {sleep_duration:?} for timeout [file={cnc_file:?}]"
208                    );
209                    sleep(sleep_duration);
210                }
211            }
212        }
213    }
214}
215
216#[cfg(test)]
217mod tests {
218    use super::*;
219    use log::error;
220    use std::os::raw::c_int;
221    use std::sync::atomic::Ordering;
222    use std::time::Duration;
223
224    #[test]
225    fn driver_idle_strategy_kinds_round_trip() {
226        let ctx = AeronDriverContext::new().unwrap();
227        for (kind, name) in [
228            (AeronIdleStrategyKind::Sleeping, "sleeping"),
229            (AeronIdleStrategyKind::Yielding, "yield"),
230            (AeronIdleStrategyKind::BusySpin, "spin"),
231            (AeronIdleStrategyKind::NoOp, "noop"),
232            (AeronIdleStrategyKind::Backoff, "backoff"),
233        ] {
234            ctx.set_conductor_idle_strategy_kind(kind).unwrap();
235            assert_eq!(name, ctx.get_conductor_idle_strategy(), "conductor {kind:?}");
236            ctx.set_sender_idle_strategy_kind(kind).unwrap();
237            ctx.set_receiver_idle_strategy_kind(kind).unwrap();
238            ctx.set_sharednetwork_idle_strategy_kind(kind).unwrap();
239            ctx.set_shared_idle_strategy_kind(kind).unwrap();
240        }
241    }
242
243    #[test]
244    fn version_check() {
245        let major = unsafe { crate::aeron_version_major() };
246        let minor = unsafe { crate::aeron_version_minor() };
247        let patch = unsafe { crate::aeron_version_patch() };
248
249        let aeron_version = format!("{}.{}.{}", major, minor, patch);
250        let cargo_version = "1.52.2";
251        assert_eq!(aeron_version, cargo_version);
252    }
253
254    #[test]
255    fn send_message() -> Result<(), AeronCError> {
256        rusteron_code_gen::test_logger::init(log::LevelFilter::Info);
257        let topic = AERON_IPC_STREAM;
258        let stream_id = 32;
259
260        let aeron_context = AeronDriverContext::new()?;
261        aeron_context.set_dir_delete_on_shutdown(true)?;
262        aeron_context.set_dir_delete_on_start(true)?;
263
264        let (stop, _driver_handle) = AeronDriver::launch_embedded(aeron_context.clone(), false);
265
266        // aeron_driver
267        //     .conductor()
268        //     .context()
269        //     .print_configuration();
270        // aeron_driver.main_do_work()?;
271        info!("aeron dir: {:?}", aeron_context.get_dir());
272
273        let dir = aeron_context.get_dir().to_string();
274        let ctx = AeronContext::new()?;
275        ctx.set_dir(&dir.into_c_string())?;
276
277        let client = Aeron::new(&ctx)?;
278
279        #[derive(Default, Debug)]
280        struct ErrorCount {
281            error_count: usize,
282        }
283
284        impl AeronErrorHandlerCallback for ErrorCount {
285            fn handle_aeron_error_handler(&mut self, error_code: c_int, msg: &str) {
286                error!("Aeron error {}: {}", error_code, msg);
287                self.error_count += 1;
288            }
289        }
290
291        let error_handler = Handler::new(ErrorCount::default());
292        ctx.set_error_handler(Some(error_handler.clone()))?;
293
294        struct Test {}
295        impl AeronAvailableCounterCallback for Test {
296            fn handle_aeron_on_available_counter(
297                &mut self,
298                counters_reader: AeronCountersReader,
299                registration_id: i64,
300                counter_id: i32,
301            ) -> () {
302                info!(
303                    "new counter counters_reader={counters_reader:?} registration_id={registration_id} counter_id={counter_id}"
304                );
305            }
306        }
307
308        impl AeronNewPublicationCallback for Test {
309            fn handle_aeron_on_new_publication(
310                &mut self,
311                channel: &str,
312                stream_id: i32,
313                session_id: i32,
314                correlation_id: i64,
315            ) -> () {
316                info!("on new publication {channel} {stream_id} {session_id} {correlation_id}")
317            }
318        }
319        let handler = Handler::new(Test {});
320        ctx.set_on_available_counter(Some(handler.clone()))?;
321        ctx.set_on_new_publication(Some(handler.clone()))?;
322
323        client.start()?;
324        info!("aeron driver started");
325        assert!(Aeron::epoch_clock() > 0);
326        assert!(Aeron::nano_clock() > 0);
327
328        let counter_async = AeronAsyncAddCounter::new(&client, 2543543, "12312312".as_bytes(), "abcd")?;
329
330        let counter = counter_async.poll_blocking(Duration::from_secs(15))?;
331        unsafe {
332            *counter.addr() += 1;
333        }
334
335        let result = AeronAsyncAddPublication::new(&client, topic, stream_id)?;
336
337        let publication = result.poll_blocking(std::time::Duration::from_secs(15))?;
338
339        info!("publication channel: {:?}", publication.channel());
340        info!("publication stream_id: {:?}", publication.stream_id());
341        info!("publication status: {:?}", publication.channel_status());
342
343        drop(publication);
344        drop(counter);
345        drop(client);
346        stop.store(true, Ordering::SeqCst);
347
348        Ok(())
349    }
350
351    #[test]
352    pub fn test_debug() -> Result<(), Box<dyn std::error::Error>> {
353        let ctx = AeronDriverContext::new()?;
354
355        println!("{:#?}", ctx);
356
357        // Capture the raw context pointer, NOT a context clone: the handler is
358        // owned by `ctx` (set_agent_on_start_function stores it as a dependency),
359        // so a clone would form a strong-reference cycle (ctx -> handler -> ctx)
360        // that close_resource_deferred_if_shared can never drain, leaking the C
361        // context. The raw pointer is sound because the handler cannot outlive
362        // the context that owns it.
363        struct AgentStartHandler {
364            ctx_ptr: *mut aeron_driver_context_t,
365        }
366        // SAFETY: see comment above — handler is owned by the context and only
367        // dereferenced on the driver agent thread while the context is alive.
368        unsafe impl Send for AgentStartHandler {}
369
370        impl AeronAgentStartFuncCallback for AgentStartHandler {
371            fn handle_aeron_agent_on_start_func(&mut self, role: &str) -> () {
372                unsafe {
373                    aeron_set_thread_affinity_on_start(
374                        self.ctx_ptr as *mut _,
375                        std::ffi::CString::new(role).unwrap().into_raw(),
376                    );
377                }
378            }
379        }
380
381        let agent_handler = Handler::new(AgentStartHandler {
382            ctx_ptr: ctx.get_inner(),
383        });
384        ctx.set_agent_on_start_function(Some(agent_handler.clone()))?;
385
386        println!("{:#?}", ctx);
387
388        Ok(())
389    }
390}
rusteron_media_driver/lib.rs

rusteron_media_driver/
lib.rs
