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

rusteron_media_driver/
lib.rs
