aeron_glide/
lib.rs

1//! Safe, idiomatic Rust wrapper for the [Aeron](https://github.com/real-logic/aeron) C++ API.
2//!
3//! This crate binds directly to the Aeron C++ client using [`cxx`](https://cxx.rs/),
4//! providing zero-cost abstractions over publications, subscriptions, images, and the
5//! embedded media driver. Closures are passed cleanly across the FFI boundary via
6//! trampolines, so the API feels native to Rust.
7//!
8//! # Quick start
9//!
10//! ```no_run
11//! use aeron_glide::AeronClient;
12//!
13//! let mut client = AeronClient::new().unwrap();
14//! client.start();
15//!
16//! let mut pub1 = client.add_publication("aeron:ipc", 1001).unwrap();
17//! let mut sub1 = client.add_subscription("aeron:ipc", 1001).unwrap();
18//!
19//! // Publish
20//! while pub1.offer(b"hello aeron") < 0 {}
21//!
22//! // Subscribe
23//! sub1.poll(10, |data| {
24//!     println!("Received: {}", std::str::from_utf8(data).unwrap());
25//! });
26//! ```
27//!
28//! # Features
29//!
30//! - **IPC and UDP** transports via [`ChannelBuilder`]
31//! - **Publications** ([`Publication`]) and **exclusive publications** ([`ExclusivePublication`])
32//! - **Zero-copy publish** via [`Publication::try_claim`]
33//! - **Fragment reassembly** via [`Subscription::poll_assembled`] with [`ControlledAction`] flow control
34//! - **Image** access for per-session stream inspection
35//! - **Counters** reader for real-time driver statistics
36//! - **Embedded media driver** ([`MediaDriver`]) with full configuration
37//! - **Archive client** (behind the `archive` feature flag): recording, replay, listing, and `ReplayMerge`
38//!
39//! # Prerequisites
40//!
41//! - CMake and a C++14 compiler (Aeron C++ is built from source automatically)
42//! - A running Aeron media driver (use the included `mediadriver` binary or [`MediaDriver`])
43//! - Java 17+ only if building with `--features archive`
44#![cfg_attr(docsrs, feature(doc_cfg))]
45
46#[cfg(feature = "archive")]
47#[cfg_attr(docsrs, doc(cfg(feature = "archive")))]
48pub mod archive;
49
50#[cxx::bridge(namespace = "aeron_rs")]
51pub mod ffi {
52    unsafe extern "C++" {
53        include!("shim.h");
54
55        type ContextWrapper;
56        type AeronWrapper;
57        type PublicationWrapper;
58        type ExclusivePublicationWrapper;
59        type SubscriptionWrapper;
60        type MediaDriverWrapper;
61        type CountersReaderWrapper;
62
63        fn create_context() -> UniquePtr<ContextWrapper>;
64        fn create_aeron(context: UniquePtr<ContextWrapper>) -> Result<UniquePtr<AeronWrapper>>;
65        fn create_media_driver() -> Result<UniquePtr<MediaDriverWrapper>>;
66
67        fn start(self: Pin<&mut AeronWrapper>);
68        fn isClosed(self: &AeronWrapper) -> bool;
69        fn addPublication(
70            self: Pin<&mut AeronWrapper>,
71            channel: &str,
72            stream_id: i32,
73        ) -> Result<UniquePtr<PublicationWrapper>>;
74        fn addExclusivePublication(
75            self: Pin<&mut AeronWrapper>,
76            channel: &str,
77            stream_id: i32,
78        ) -> Result<UniquePtr<ExclusivePublicationWrapper>>;
79        fn addSubscription(
80            self: Pin<&mut AeronWrapper>,
81            channel: &str,
82            stream_id: i32,
83        ) -> Result<UniquePtr<SubscriptionWrapper>>;
84        fn countersReader(self: &AeronWrapper) -> UniquePtr<CountersReaderWrapper>;
85
86        fn start(self: Pin<&mut MediaDriverWrapper>) -> Result<()>;
87
88        fn setDir(self: Pin<&mut MediaDriverWrapper>, dir: &str) -> Result<()>;
89        fn setDirDeleteOnStart(self: Pin<&mut MediaDriverWrapper>, value: bool) -> Result<()>;
90        fn setDirDeleteOnShutdown(self: Pin<&mut MediaDriverWrapper>, value: bool) -> Result<()>;
91        fn setThreadingMode(self: Pin<&mut MediaDriverWrapper>, mode: i32) -> Result<()>;
92        fn setConductorIdleStrategy(self: Pin<&mut MediaDriverWrapper>, name: &str) -> Result<()>;
93        fn setSenderIdleStrategy(self: Pin<&mut MediaDriverWrapper>, name: &str) -> Result<()>;
94        fn setReceiverIdleStrategy(self: Pin<&mut MediaDriverWrapper>, name: &str) -> Result<()>;
95        fn setTermBufferLength(self: Pin<&mut MediaDriverWrapper>, value: usize) -> Result<()>;
96        fn setIpcTermBufferLength(self: Pin<&mut MediaDriverWrapper>, value: usize) -> Result<()>;
97        fn setMtuLength(self: Pin<&mut MediaDriverWrapper>, value: usize) -> Result<()>;
98        fn setIpcMtuLength(self: Pin<&mut MediaDriverWrapper>, value: usize) -> Result<()>;
99        fn setSocketSoRcvbuf(self: Pin<&mut MediaDriverWrapper>, value: usize) -> Result<()>;
100        fn setSocketSoSndbuf(self: Pin<&mut MediaDriverWrapper>, value: usize) -> Result<()>;
101        fn setPrintConfiguration(self: Pin<&mut MediaDriverWrapper>, value: bool) -> Result<()>;
102        fn setConductorCpuAffinity(self: Pin<&mut MediaDriverWrapper>, cpu_id: i32) -> Result<()>;
103        fn setSenderCpuAffinity(self: Pin<&mut MediaDriverWrapper>, cpu_id: i32) -> Result<()>;
104        fn setReceiverCpuAffinity(self: Pin<&mut MediaDriverWrapper>, cpu_id: i32) -> Result<()>;
105
106        fn offer(self: Pin<&mut PublicationWrapper>, buffer: &[u8]) -> i64;
107        fn tryClaim(self: Pin<&mut PublicationWrapper>, length: usize, handler_id: usize) -> i64;
108        fn isConnected(self: &PublicationWrapper) -> bool;
109        fn sessionId(self: &PublicationWrapper) -> i32;
110
111        fn offer(self: Pin<&mut ExclusivePublicationWrapper>, buffer: &[u8]) -> i64;
112        fn tryClaim(
113            self: Pin<&mut ExclusivePublicationWrapper>,
114            length: usize,
115            handler_id: usize,
116        ) -> i64;
117        fn isConnected(self: &ExclusivePublicationWrapper) -> bool;
118
119        fn poll(self: Pin<&mut SubscriptionWrapper>, fragment_limit: i32, handler_id: usize)
120        -> i32;
121        fn pollAssembled(
122            self: Pin<&mut SubscriptionWrapper>,
123            fragment_limit: i32,
124            handler_id: usize,
125        ) -> i32;
126        fn controlledPollAssembled(
127            self: Pin<&mut SubscriptionWrapper>,
128            fragment_limit: i32,
129            handler_id: usize,
130        ) -> i32;
131        fn isConnected(self: &SubscriptionWrapper) -> bool;
132        fn imageCount(self: &SubscriptionWrapper) -> i32;
133        fn imageByIndex(
134            self: Pin<&mut SubscriptionWrapper>,
135            index: usize,
136        ) -> Result<UniquePtr<ImageWrapper>>;
137        fn imageBySessionId(
138            self: Pin<&mut SubscriptionWrapper>,
139            session_id: i32,
140        ) -> Result<UniquePtr<ImageWrapper>>;
141
142        type ImageWrapper;
143        fn sessionId(self: &ImageWrapper) -> i32;
144        fn correlationId(self: &ImageWrapper) -> i64;
145        fn joinPosition(self: &ImageWrapper) -> i64;
146        fn sourceIdentity(self: &ImageWrapper) -> String;
147        fn position(self: &ImageWrapper) -> i64;
148        fn setPosition(self: Pin<&mut ImageWrapper>, new_position: i64);
149        fn isClosed(self: &ImageWrapper) -> bool;
150        fn isEndOfStream(self: &ImageWrapper) -> bool;
151        fn endOfStreamPosition(self: &ImageWrapper) -> i64;
152        fn poll(self: Pin<&mut ImageWrapper>, fragment_limit: i32, handler_id: usize) -> i32;
153        fn controlledPollAssembled(
154            self: Pin<&mut ImageWrapper>,
155            fragment_limit: i32,
156            handler_id: usize,
157        ) -> i32;
158
159        fn maxCounterId(self: &CountersReaderWrapper) -> i32;
160        fn getCounterValue(self: &CountersReaderWrapper, id: i32) -> i64;
161        fn getCounterState(self: &CountersReaderWrapper, id: i32) -> i32;
162        fn getCounterTypeId(self: &CountersReaderWrapper, id: i32) -> i32;
163        fn getCounterLabel(self: &CountersReaderWrapper, id: i32) -> String;
164        fn forEach(self: &CountersReaderWrapper, handler_id: usize);
165    }
166
167    extern "Rust" {
168        fn handle_fragment(handler_id: usize, buffer: &[u8]);
169        fn handle_controlled_fragment(handler_id: usize, buffer: &[u8]) -> i32;
170        fn handle_claim(handler_id: usize, buffer: &mut [u8]) -> bool;
171        fn handle_counters_metadata(
172            handler_id: usize,
173            counter_id: i32,
174            type_id: i32,
175            key: &[u8],
176            label: String,
177        );
178    }
179}
180
181/// Aeron client — the main entry point for creating publications and subscriptions.
182///
183/// Each client maintains its own connection to the media driver. You can create
184/// multiple clients in the same process (e.g., one per thread).
185pub struct AeronClient {
186    inner: cxx::UniquePtr<ffi::AeronWrapper>,
187}
188
189impl AeronClient {
190    /// Create a new Aeron client connected to the media driver.
191    pub fn new() -> Result<Self, Box<dyn std::error::Error>> {
192        let ctx = ffi::create_context();
193        let aeron = ffi::create_aeron(ctx)?;
194
195        Ok(Self { inner: aeron })
196    }
197
198    /// Start the client conductor thread.
199    pub fn start(&mut self) {
200        self.inner.pin_mut().start();
201    }
202
203    /// Returns `true` if the client has been closed.
204    pub fn is_closed(&self) -> bool {
205        self.inner.isClosed()
206    }
207
208    /// Add a concurrent publication on the given channel and stream ID.
209    /// Multiple publishers can share the same channel+stream.
210    pub fn add_publication(
211        &mut self,
212        channel: &str,
213        stream_id: i32,
214    ) -> Result<Publication, Box<dyn std::error::Error>> {
215        let pub_inner = self.inner.pin_mut().addPublication(channel, stream_id)?;
216        Ok(Publication { inner: pub_inner })
217    }
218
219    /// Add an exclusive publication on the given channel and stream ID.
220    /// Only one publisher is allowed per session — lower overhead than concurrent.
221    pub fn add_exclusive_publication(
222        &mut self,
223        channel: &str,
224        stream_id: i32,
225    ) -> Result<ExclusivePublication, Box<dyn std::error::Error>> {
226        let pub_inner = self
227            .inner
228            .pin_mut()
229            .addExclusivePublication(channel, stream_id)?;
230        Ok(ExclusivePublication { inner: pub_inner })
231    }
232
233    /// Add a subscription on the given channel and stream ID.
234    pub fn add_subscription(
235        &mut self,
236        channel: &str,
237        stream_id: i32,
238    ) -> Result<Subscription, Box<dyn std::error::Error>> {
239        let sub_inner = self.inner.pin_mut().addSubscription(channel, stream_id)?;
240        Ok(Subscription { inner: sub_inner })
241    }
242
243    /// Get a reader for the media driver's CNC counters (bytes sent/received, errors, etc.).
244    pub fn counters_reader(&self) -> CountersReader {
245        CountersReader {
246            inner: self.inner.countersReader(),
247        }
248    }
249}
250
251/// A concurrent publication for sending messages on a channel+stream.
252///
253/// Returns negative values from [`offer`](Publication::offer) on back-pressure or when closed.
254pub struct Publication {
255    inner: cxx::UniquePtr<ffi::PublicationWrapper>,
256}
257
258impl Publication {
259    /// Publish a message. Returns the new stream position on success,
260    /// or a negative value on back-pressure / not connected / closed.
261    pub fn offer(&mut self, buffer: &[u8]) -> i64 {
262        self.inner.pin_mut().offer(buffer)
263    }
264
265    /// Zero-copy publish: claims a region of the log buffer, calls `handler` with a mutable
266    /// slice pointing directly into shared memory, then commits or aborts based on the return value.
267    /// Returns the stream position (>0 on success, negative on back-pressure/closed).
268    pub fn try_claim<F>(&mut self, length: usize, mut handler: F) -> i64
269    where
270        F: FnMut(&mut [u8]) -> bool,
271    {
272        let handler_id = &handler as *const _ as usize;
273        let mut_ptr: *mut (dyn FnMut(&mut [u8]) -> bool + 'static) = unsafe {
274            std::mem::transmute::<
275                *mut dyn FnMut(&mut [u8]) -> bool,
276                *mut (dyn FnMut(&mut [u8]) -> bool + 'static),
277            >(&mut handler as *mut dyn FnMut(&mut [u8]) -> bool)
278        };
279
280        CLAIM_HANDLERS.with(|handlers| {
281            handlers.borrow_mut().insert(handler_id, mut_ptr);
282        });
283
284        let result = self.inner.pin_mut().tryClaim(length, handler_id);
285
286        CLAIM_HANDLERS.with(|handlers| {
287            handlers.borrow_mut().remove(&handler_id);
288        });
289
290        result
291    }
292
293    /// Returns `true` if there is at least one subscriber connected to this publication.
294    pub fn is_connected(&self) -> bool {
295        self.inner.isConnected()
296    }
297
298    /// The session ID assigned by the media driver for this publication.
299    pub fn session_id(&self) -> i32 {
300        self.inner.sessionId()
301    }
302}
303
304/// An exclusive publication — single-writer, lower overhead than [`Publication`].
305pub struct ExclusivePublication {
306    inner: cxx::UniquePtr<ffi::ExclusivePublicationWrapper>,
307}
308
309impl ExclusivePublication {
310    /// Publish a message. Returns the new stream position on success,
311    /// or a negative value on back-pressure / not connected / closed.
312    pub fn offer(&mut self, buffer: &[u8]) -> i64 {
313        self.inner.pin_mut().offer(buffer)
314    }
315
316    /// Zero-copy publish: claims a region of the log buffer, calls `handler` with a mutable
317    /// slice pointing directly into shared memory, then commits or aborts based on the return value.
318    /// Returns the stream position (>0 on success, negative on back-pressure/closed).
319    pub fn try_claim<F>(&mut self, length: usize, mut handler: F) -> i64
320    where
321        F: FnMut(&mut [u8]) -> bool,
322    {
323        let handler_id = &handler as *const _ as usize;
324        let mut_ptr: *mut (dyn FnMut(&mut [u8]) -> bool + 'static) = unsafe {
325            std::mem::transmute::<
326                *mut dyn FnMut(&mut [u8]) -> bool,
327                *mut (dyn FnMut(&mut [u8]) -> bool + 'static),
328            >(&mut handler as *mut dyn FnMut(&mut [u8]) -> bool)
329        };
330
331        CLAIM_HANDLERS.with(|handlers| {
332            handlers.borrow_mut().insert(handler_id, mut_ptr);
333        });
334
335        let result = self.inner.pin_mut().tryClaim(length, handler_id);
336
337        CLAIM_HANDLERS.with(|handlers| {
338            handlers.borrow_mut().remove(&handler_id);
339        });
340
341        result
342    }
343
344    /// Returns `true` if there is at least one subscriber connected to this publication.
345    pub fn is_connected(&self) -> bool {
346        self.inner.isConnected()
347    }
348}
349
350use std::cell::RefCell;
351use std::collections::HashMap;
352
353/// Flow-control actions for `poll_assembled` when the handler returns a `ControlledAction`.
354/// Matches Aeron's `ControlledPollAction` enum values.
355#[repr(i32)]
356#[derive(Debug, Clone, Copy, PartialEq, Eq)]
357pub enum ControlledAction {
358    /// Abort polling — rewind position, re-deliver this fragment next poll.
359    Abort = 0,
360    /// Stop polling this image, commit position up to this fragment.
361    Break = 1,
362    /// Checkpoint position for flow control, continue polling.
363    Commit = 2,
364    /// Continue processing (default behavior).
365    Continue = 3,
366}
367
368/// Trait that allows `poll_assembled` to accept handlers returning either `()` or `ControlledAction`.
369/// Closures returning `()` map to `ControlledAction::Continue`.
370pub trait PollAction {
371    fn into_action(self) -> ControlledAction;
372}
373
374impl PollAction for () {
375    #[inline]
376    fn into_action(self) -> ControlledAction {
377        ControlledAction::Continue
378    }
379}
380
381impl PollAction for ControlledAction {
382    #[inline]
383    fn into_action(self) -> ControlledAction {
384        self
385    }
386}
387
388type FragmentHandlerMap = RefCell<HashMap<usize, *mut dyn FnMut(&[u8])>>;
389type ClaimHandlerMap = RefCell<HashMap<usize, *mut dyn FnMut(&mut [u8]) -> bool>>;
390type ControlledHandlerMap = RefCell<HashMap<usize, *mut dyn FnMut(&[u8]) -> ControlledAction>>;
391
392// Thread-local registries for closures passed across the cxx boundary.
393// We use pointer-based handler IDs since cxx doesn't support passing trait objects directly.
394thread_local! {
395    pub(crate) static HANDLERS: FragmentHandlerMap = RefCell::new(HashMap::new());
396    static CLAIM_HANDLERS: ClaimHandlerMap = RefCell::new(HashMap::new());
397    static CONTROLLED_HANDLERS: ControlledHandlerMap = RefCell::new(HashMap::new());
398}
399
400fn handle_fragment(handler_id: usize, buffer: &[u8]) {
401    HANDLERS.with(|handlers| {
402        if let Some(handler_ptr) = handlers.borrow_mut().get_mut(&handler_id) {
403            unsafe {
404                let handler = &mut **handler_ptr;
405                handler(buffer);
406            }
407        }
408    });
409}
410
411fn handle_controlled_fragment(handler_id: usize, buffer: &[u8]) -> i32 {
412    CONTROLLED_HANDLERS.with(|handlers| {
413        if let Some(handler_ptr) = handlers.borrow_mut().get_mut(&handler_id) {
414            unsafe {
415                let handler = &mut **handler_ptr;
416                handler(buffer) as i32
417            }
418        } else {
419            ControlledAction::Abort as i32
420        }
421    })
422}
423
424fn handle_claim(handler_id: usize, buffer: &mut [u8]) -> bool {
425    CLAIM_HANDLERS.with(|handlers| {
426        if let Some(handler_ptr) = handlers.borrow_mut().get_mut(&handler_id) {
427            unsafe {
428                let handler = &mut **handler_ptr;
429                handler(buffer)
430            }
431        } else {
432            false // abort if handler not found
433        }
434    })
435}
436
437/// A subscription for receiving messages on a channel+stream.
438pub struct Subscription {
439    inner: cxx::UniquePtr<ffi::SubscriptionWrapper>,
440}
441
442impl Subscription {
443    /// Poll for new messages, calling `handler` for each fragment received.
444    /// Returns the number of fragments dispatched.
445    pub fn poll<F>(&mut self, limit: i32, mut handler: F) -> i32
446    where
447        F: FnMut(&[u8]),
448    {
449        let handler_id = &handler as *const _ as usize;
450        let mut_ptr: *mut (dyn FnMut(&[u8]) + 'static) = unsafe {
451            std::mem::transmute::<*mut dyn FnMut(&[u8]), *mut (dyn FnMut(&[u8]) + 'static)>(
452                &mut handler as *mut dyn FnMut(&[u8]),
453            )
454        };
455
456        HANDLERS.with(|handlers| {
457            handlers.borrow_mut().insert(handler_id, mut_ptr);
458        });
459
460        let result = self.inner.pin_mut().poll(limit, handler_id);
461
462        HANDLERS.with(|handlers| {
463            handlers.borrow_mut().remove(&handler_id);
464        });
465
466        result
467    }
468
469    /// Poll with automatic fragment reassembly. Messages that span multiple fragments
470    /// are reassembled before being delivered to the handler, which always receives
471    /// complete messages.
472    ///
473    /// The handler can return `()` (maps to Continue) or a `ControlledAction` for
474    /// flow-control (Abort to retry, Break to stop, Commit to checkpoint, Continue to proceed).
475    pub fn poll_assembled<R, F>(&mut self, limit: i32, mut handler: F) -> i32
476    where
477        R: PollAction,
478        F: FnMut(&[u8]) -> R,
479    {
480        // Wrap the user's handler to always produce a ControlledAction
481        let mut controlled = |data: &[u8]| -> ControlledAction { handler(data).into_action() };
482
483        let handler_id = &controlled as *const _ as usize;
484        let mut_ptr: *mut (dyn FnMut(&[u8]) -> ControlledAction + 'static) = unsafe {
485            std::mem::transmute::<
486                *mut dyn FnMut(&[u8]) -> ControlledAction,
487                *mut (dyn FnMut(&[u8]) -> ControlledAction + 'static),
488            >(&mut controlled as *mut dyn FnMut(&[u8]) -> ControlledAction)
489        };
490
491        CONTROLLED_HANDLERS.with(|handlers| {
492            handlers.borrow_mut().insert(handler_id, mut_ptr);
493        });
494
495        let result = self
496            .inner
497            .pin_mut()
498            .controlledPollAssembled(limit, handler_id);
499
500        CONTROLLED_HANDLERS.with(|handlers| {
501            handlers.borrow_mut().remove(&handler_id);
502        });
503
504        result
505    }
506
507    /// Returns `true` if there is at least one publisher connected to this subscription.
508    pub fn is_connected(&self) -> bool {
509        self.inner.isConnected()
510    }
511
512    #[cfg(feature = "archive")]
513    pub(crate) fn inner_pin_mut(&mut self) -> std::pin::Pin<&mut ffi::SubscriptionWrapper> {
514        self.inner.pin_mut()
515    }
516
517    /// The number of active images (one per publisher session) on this subscription.
518    pub fn image_count(&self) -> i32 {
519        self.inner.imageCount()
520    }
521
522    /// Get an image by its index (0-based). Images appear in the order they were connected.
523    pub fn image_by_index(&mut self, index: usize) -> Result<Image, Box<dyn std::error::Error>> {
524        let img = self.inner.pin_mut().imageByIndex(index)?;
525        Ok(Image { inner: img })
526    }
527
528    /// Get an image by the publisher's session ID.
529    pub fn image_by_session_id(
530        &mut self,
531        session_id: i32,
532    ) -> Result<Image, Box<dyn std::error::Error>> {
533        let img = self.inner.pin_mut().imageBySessionId(session_id)?;
534        Ok(Image { inner: img })
535    }
536}
537
538/// A single publisher session as seen by a subscriber.
539///
540/// Each publisher session creates one image on each matching subscription.
541/// Images track their own position and can be polled independently.
542pub struct Image {
543    inner: cxx::UniquePtr<ffi::ImageWrapper>,
544}
545
546impl Image {
547    #[cfg(feature = "archive")]
548    pub(crate) fn from_raw(inner: cxx::UniquePtr<ffi::ImageWrapper>) -> Self {
549        Self { inner }
550    }
551
552    /// The session ID of the publisher that created this image.
553    pub fn session_id(&self) -> i32 {
554        self.inner.sessionId()
555    }
556
557    /// The correlation ID assigned by the media driver when the image was created.
558    pub fn correlation_id(&self) -> i64 {
559        self.inner.correlationId()
560    }
561
562    /// The position at which this image was joined.
563    pub fn join_position(&self) -> i64 {
564        self.inner.joinPosition()
565    }
566
567    /// The source identity string (e.g., `"192.168.1.1:40123"`).
568    pub fn source_identity(&self) -> String {
569        self.inner.sourceIdentity()
570    }
571
572    /// The current consumption position within the stream.
573    pub fn position(&self) -> i64 {
574        self.inner.position()
575    }
576
577    /// Set the subscriber position (e.g., to skip ahead or rewind within the term buffer).
578    pub fn set_position(&mut self, new_position: i64) {
579        self.inner.pin_mut().setPosition(new_position);
580    }
581
582    /// Returns `true` if the image has been closed (publisher disconnected or timed out).
583    pub fn is_closed(&self) -> bool {
584        self.inner.isClosed()
585    }
586
587    /// Returns `true` if the publisher has signalled end-of-stream.
588    pub fn is_end_of_stream(&self) -> bool {
589        self.inner.isEndOfStream()
590    }
591
592    /// The position at which the end-of-stream was signalled.
593    pub fn end_of_stream_position(&self) -> i64 {
594        self.inner.endOfStreamPosition()
595    }
596
597    /// Poll this specific image for fragments. Returns the number of fragments dispatched.
598    pub fn poll<F>(&mut self, limit: i32, mut handler: F) -> i32
599    where
600        F: FnMut(&[u8]),
601    {
602        let handler_id = &handler as *const _ as usize;
603        let mut_ptr: *mut (dyn FnMut(&[u8]) + 'static) = unsafe {
604            std::mem::transmute::<*mut dyn FnMut(&[u8]), *mut (dyn FnMut(&[u8]) + 'static)>(
605                &mut handler as *mut dyn FnMut(&[u8]),
606            )
607        };
608
609        HANDLERS.with(|handlers| {
610            handlers.borrow_mut().insert(handler_id, mut_ptr);
611        });
612
613        let result = self.inner.pin_mut().poll(limit, handler_id);
614
615        HANDLERS.with(|handlers| {
616            handlers.borrow_mut().remove(&handler_id);
617        });
618
619        result
620    }
621
622    pub fn poll_assembled<R, F>(&mut self, limit: i32, mut handler: F) -> i32
623    where
624        R: PollAction,
625        F: FnMut(&[u8]) -> R,
626    {
627        let mut controlled = |data: &[u8]| -> ControlledAction { handler(data).into_action() };
628
629        let handler_id = &controlled as *const _ as usize;
630        let mut_ptr: *mut (dyn FnMut(&[u8]) -> ControlledAction + 'static) = unsafe {
631            std::mem::transmute::<
632                *mut dyn FnMut(&[u8]) -> ControlledAction,
633                *mut (dyn FnMut(&[u8]) -> ControlledAction + 'static),
634            >(&mut controlled as *mut dyn FnMut(&[u8]) -> ControlledAction)
635        };
636
637        CONTROLLED_HANDLERS.with(|handlers| {
638            handlers.borrow_mut().insert(handler_id, mut_ptr);
639        });
640
641        let result = self
642            .inner
643            .pin_mut()
644            .controlledPollAssembled(limit, handler_id);
645
646        CONTROLLED_HANDLERS.with(|handlers| {
647            handlers.borrow_mut().remove(&handler_id);
648        });
649
650        result
651    }
652}
653
654/// Reader for the media driver's CNC (Command and Control) counters.
655///
656/// Provides access to real-time statistics like bytes sent/received, NAKs,
657/// errors, and heartbeats.
658pub struct CountersReader {
659    inner: cxx::UniquePtr<ffi::CountersReaderWrapper>,
660}
661
662type MetadataHandlerMap = RefCell<HashMap<usize, *mut dyn FnMut(i32, i32, &[u8], &str)>>;
663
664thread_local! {
665    static METADATA_HANDLERS: MetadataHandlerMap = RefCell::new(HashMap::new());
666}
667
668fn handle_counters_metadata(
669    handler_id: usize,
670    counter_id: i32,
671    type_id: i32,
672    key: &[u8],
673    label: String,
674) {
675    METADATA_HANDLERS.with(|handlers| {
676        if let Some(handler_ptr) = handlers.borrow_mut().get_mut(&handler_id) {
677            unsafe {
678                let handler = &mut **handler_ptr;
679                handler(counter_id, type_id, key, &label);
680            }
681        }
682    });
683}
684
685impl CountersReader {
686    /// The highest counter ID currently allocated.
687    pub fn max_counter_id(&self) -> i32 {
688        self.inner.maxCounterId()
689    }
690
691    /// Read the current value of a counter by ID.
692    pub fn get_counter_value(&self, id: i32) -> i64 {
693        self.inner.getCounterValue(id)
694    }
695
696    /// Get the state of a counter (e.g., active, inactive).
697    pub fn get_counter_state(&self, id: i32) -> i32 {
698        self.inner.getCounterState(id)
699    }
700
701    /// Get the type ID of a counter.
702    pub fn get_counter_type_id(&self, id: i32) -> i32 {
703        self.inner.getCounterTypeId(id)
704    }
705
706    /// Get the human-readable label of a counter.
707    pub fn get_counter_label(&self, id: i32) -> String {
708        self.inner.getCounterLabel(id)
709    }
710
711    /// Iterate over all counters, calling `handler(counter_id, type_id, key_bytes, label)` for each.
712    pub fn for_each<F>(&self, mut handler: F)
713    where
714        F: FnMut(i32, i32, &[u8], &str),
715    {
716        let handler_id = &handler as *const _ as usize;
717        #[allow(clippy::type_complexity)]
718        let mut_ptr: *mut (dyn FnMut(i32, i32, &[u8], &str) + 'static) = unsafe {
719            std::mem::transmute::<
720                *mut dyn FnMut(i32, i32, &[u8], &str),
721                *mut (dyn FnMut(i32, i32, &[u8], &str) + 'static),
722            >(&mut handler as *mut dyn FnMut(i32, i32, &[u8], &str))
723        };
724
725        METADATA_HANDLERS.with(|handlers| {
726            handlers.borrow_mut().insert(handler_id, mut_ptr);
727        });
728
729        self.inner.forEach(handler_id);
730
731        METADATA_HANDLERS.with(|handlers| {
732            handlers.borrow_mut().remove(&handler_id);
733        });
734    }
735}
736
737impl Default for AeronClient {
738    fn default() -> Self {
739        Self::new().expect("Failed to create AeronClient")
740    }
741}
742
743/// Threading model for the embedded media driver.
744#[repr(i32)]
745#[derive(Debug, Clone, Copy, PartialEq, Eq)]
746pub enum ThreadingMode {
747    /// Separate threads for conductor, sender, and receiver.
748    Dedicated = 0,
749    /// Sender and receiver share a thread; conductor is separate.
750    SharedNetwork = 1,
751    /// All three run on a single shared thread.
752    Shared = 2,
753    /// Caller-driven — the application invokes the driver duty cycle.
754    Invoker = 3,
755}
756
757/// Idle strategy for media driver threads.
758#[derive(Debug, Clone, Copy, PartialEq, Eq)]
759pub enum IdleStrategy {
760    /// Progressive back-off: spin → yield → park.
761    Backoff,
762    /// Busy spin (lowest latency, highest CPU).
763    Spin,
764    /// Thread yield.
765    Yield,
766    /// Thread sleep.
767    Sleeping,
768    /// No-op (do nothing between duty cycles).
769    Noop,
770}
771
772impl IdleStrategy {
773    pub fn as_str(&self) -> &'static str {
774        match self {
775            IdleStrategy::Backoff => "backoff",
776            IdleStrategy::Spin => "spin",
777            IdleStrategy::Yield => "yield",
778            IdleStrategy::Sleeping => "sleeping",
779            IdleStrategy::Noop => "noop",
780        }
781    }
782}
783
784/// An embedded C media driver that manages shared memory buffers and handles
785/// publication/subscription matching.
786pub struct MediaDriver {
787    inner: cxx::UniquePtr<ffi::MediaDriverWrapper>,
788}
789
790impl MediaDriver {
791    /// Create a new media driver with default settings.
792    pub fn new() -> Result<Self, Box<dyn std::error::Error>> {
793        let inner = ffi::create_media_driver()?;
794        Ok(Self { inner })
795    }
796
797    /// Start the media driver. Must be called before any clients can connect.
798    pub fn start(&mut self) -> Result<(), Box<dyn std::error::Error>> {
799        self.inner.pin_mut().start()?;
800        Ok(())
801    }
802
803    /// Set the Aeron directory for shared memory files.
804    pub fn set_dir(&mut self, dir: &str) -> Result<(), Box<dyn std::error::Error>> {
805        self.inner.pin_mut().setDir(dir)?;
806        Ok(())
807    }
808
809    pub fn set_dir_delete_on_start(
810        &mut self,
811        value: bool,
812    ) -> Result<(), Box<dyn std::error::Error>> {
813        self.inner.pin_mut().setDirDeleteOnStart(value)?;
814        Ok(())
815    }
816
817    pub fn set_dir_delete_on_shutdown(
818        &mut self,
819        value: bool,
820    ) -> Result<(), Box<dyn std::error::Error>> {
821        self.inner.pin_mut().setDirDeleteOnShutdown(value)?;
822        Ok(())
823    }
824
825    pub fn set_threading_mode(
826        &mut self,
827        mode: ThreadingMode,
828    ) -> Result<(), Box<dyn std::error::Error>> {
829        self.inner.pin_mut().setThreadingMode(mode as i32)?;
830        Ok(())
831    }
832
833    pub fn set_conductor_idle_strategy(
834        &mut self,
835        strategy: IdleStrategy,
836    ) -> Result<(), Box<dyn std::error::Error>> {
837        self.inner
838            .pin_mut()
839            .setConductorIdleStrategy(strategy.as_str())?;
840        Ok(())
841    }
842
843    pub fn set_sender_idle_strategy(
844        &mut self,
845        strategy: IdleStrategy,
846    ) -> Result<(), Box<dyn std::error::Error>> {
847        self.inner
848            .pin_mut()
849            .setSenderIdleStrategy(strategy.as_str())?;
850        Ok(())
851    }
852
853    pub fn set_receiver_idle_strategy(
854        &mut self,
855        strategy: IdleStrategy,
856    ) -> Result<(), Box<dyn std::error::Error>> {
857        self.inner
858            .pin_mut()
859            .setReceiverIdleStrategy(strategy.as_str())?;
860        Ok(())
861    }
862
863    pub fn set_term_buffer_length(
864        &mut self,
865        value: usize,
866    ) -> Result<(), Box<dyn std::error::Error>> {
867        self.inner.pin_mut().setTermBufferLength(value)?;
868        Ok(())
869    }
870
871    pub fn set_ipc_term_buffer_length(
872        &mut self,
873        value: usize,
874    ) -> Result<(), Box<dyn std::error::Error>> {
875        self.inner.pin_mut().setIpcTermBufferLength(value)?;
876        Ok(())
877    }
878
879    pub fn set_mtu_length(&mut self, value: usize) -> Result<(), Box<dyn std::error::Error>> {
880        self.inner.pin_mut().setMtuLength(value)?;
881        Ok(())
882    }
883
884    pub fn set_ipc_mtu_length(&mut self, value: usize) -> Result<(), Box<dyn std::error::Error>> {
885        self.inner.pin_mut().setIpcMtuLength(value)?;
886        Ok(())
887    }
888
889    pub fn set_socket_so_rcvbuf(&mut self, value: usize) -> Result<(), Box<dyn std::error::Error>> {
890        self.inner.pin_mut().setSocketSoRcvbuf(value)?;
891        Ok(())
892    }
893
894    pub fn set_socket_so_sndbuf(&mut self, value: usize) -> Result<(), Box<dyn std::error::Error>> {
895        self.inner.pin_mut().setSocketSoSndbuf(value)?;
896        Ok(())
897    }
898
899    pub fn set_print_configuration(
900        &mut self,
901        value: bool,
902    ) -> Result<(), Box<dyn std::error::Error>> {
903        self.inner.pin_mut().setPrintConfiguration(value)?;
904        Ok(())
905    }
906
907    pub fn set_conductor_cpu_affinity(
908        &mut self,
909        cpu_id: i32,
910    ) -> Result<(), Box<dyn std::error::Error>> {
911        self.inner.pin_mut().setConductorCpuAffinity(cpu_id)?;
912        Ok(())
913    }
914
915    pub fn set_sender_cpu_affinity(
916        &mut self,
917        cpu_id: i32,
918    ) -> Result<(), Box<dyn std::error::Error>> {
919        self.inner.pin_mut().setSenderCpuAffinity(cpu_id)?;
920        Ok(())
921    }
922
923    pub fn set_receiver_cpu_affinity(
924        &mut self,
925        cpu_id: i32,
926    ) -> Result<(), Box<dyn std::error::Error>> {
927        self.inner.pin_mut().setReceiverCpuAffinity(cpu_id)?;
928        Ok(())
929    }
930}
931
932impl Default for MediaDriver {
933    fn default() -> Self {
934        Self::new().expect("Failed to create MediaDriver")
935    }
936}
937
938/// Builder for Aeron channel URIs (`aeron:ipc` or `aeron:udp?key=value|...`).
939///
940/// # Examples
941///
942/// ```
943/// use aeron_glide::ChannelBuilder;
944///
945/// let ipc = ChannelBuilder::ipc().build();
946/// assert_eq!(ipc, "aeron:ipc");
947///
948/// let udp = ChannelBuilder::udp()
949///     .endpoint("localhost:20121")
950///     .mtu(8192)
951///     .build();
952/// assert_eq!(udp, "aeron:udp?endpoint=localhost:20121|mtu=8192");
953/// ```
954pub struct ChannelBuilder {
955    media: &'static str,
956    params: Vec<(String, String)>,
957}
958
959impl ChannelBuilder {
960    /// Create an IPC (shared memory) channel builder.
961    pub fn ipc() -> Self {
962        Self {
963            media: "ipc",
964            params: Vec::new(),
965        }
966    }
967
968    /// Create a UDP channel builder.
969    pub fn udp() -> Self {
970        Self {
971            media: "udp",
972            params: Vec::new(),
973        }
974    }
975
976    /// Set the endpoint address (e.g., `"localhost:20121"` or `"224.0.1.1:40456"` for multicast).
977    pub fn endpoint(self, value: &str) -> Self {
978        self.param("endpoint", value)
979    }
980    pub fn control(self, value: &str) -> Self {
981        self.param("control", value)
982    }
983    pub fn control_mode(self, value: &str) -> Self {
984        self.param("control-mode", value)
985    }
986    pub fn interface(self, value: &str) -> Self {
987        self.param("interface", value)
988    }
989    pub fn mtu(self, bytes: usize) -> Self {
990        self.param("mtu", &bytes.to_string())
991    }
992    pub fn term_length(self, bytes: usize) -> Self {
993        self.param("term-length", &bytes.to_string())
994    }
995    pub fn session_id(self, id: i32) -> Self {
996        self.param("session-id", &id.to_string())
997    }
998    pub fn ttl(self, hops: u8) -> Self {
999        self.param("ttl", &hops.to_string())
1000    }
1001    pub fn reliable(self, value: bool) -> Self {
1002        self.param("reliable", if value { "true" } else { "false" })
1003    }
1004    pub fn sparse(self, value: bool) -> Self {
1005        self.param("sparse", if value { "true" } else { "false" })
1006    }
1007    pub fn linger(self, ns: u64) -> Self {
1008        self.param("linger", &ns.to_string())
1009    }
1010    pub fn tether(self, value: bool) -> Self {
1011        self.param("tether", if value { "true" } else { "false" })
1012    }
1013    pub fn rejoin(self, value: bool) -> Self {
1014        self.param("rejoin", if value { "true" } else { "false" })
1015    }
1016    pub fn flow_control(self, value: &str) -> Self {
1017        self.param("fc", value)
1018    }
1019    pub fn congestion_control(self, value: &str) -> Self {
1020        self.param("cc", value)
1021    }
1022    pub fn socket_sndbuf(self, bytes: usize) -> Self {
1023        self.param("so-sndbuf", &bytes.to_string())
1024    }
1025    pub fn socket_rcvbuf(self, bytes: usize) -> Self {
1026        self.param("so-rcvbuf", &bytes.to_string())
1027    }
1028    pub fn receiver_window(self, bytes: usize) -> Self {
1029        self.param("rcv-wnd", &bytes.to_string())
1030    }
1031
1032    /// Set an arbitrary channel parameter by key and value.
1033    pub fn param(mut self, key: &str, value: &str) -> Self {
1034        self.params.push((key.to_string(), value.to_string()));
1035        self
1036    }
1037
1038    /// Build the channel URI string.
1039    pub fn build(&self) -> String {
1040        let mut uri = format!("aeron:{}", self.media);
1041        for (i, (key, value)) in self.params.iter().enumerate() {
1042            uri.push(if i == 0 { '?' } else { '|' });
1043            uri.push_str(key);
1044            uri.push('=');
1045            uri.push_str(value);
1046        }
1047        uri
1048    }
1049}
1050
1051#[cfg(test)]
1052mod tests {
1053    use super::*;
1054
1055    #[test]
1056    fn test_aeron_creation_with_driver() {
1057        // 1. Start embedded driver
1058        let mut driver = MediaDriver::new().expect("Failed to create MediaDriver");
1059        driver.start().expect("Failed to start MediaDriver");
1060
1061        // Wait a tiny bit for the driver to spin up its files in /dev/shm
1062        std::thread::sleep(std::time::Duration::from_millis(100));
1063
1064        // 2. Connect client
1065        let mut client = AeronClient::new().expect("Failed to connect to media driver");
1066        client.start();
1067        assert!(!client.is_closed());
1068
1069        // 3. Test Pub/Sub creation
1070        let mut publ = client
1071            .add_publication("aeron:ipc", 10)
1072            .expect("add pub failed");
1073        let mut sub = client
1074            .add_subscription("aeron:ipc", 10)
1075            .expect("add sub failed");
1076
1077        // 4. Wait for connection then test Image API
1078        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
1079        while !sub.is_connected() && std::time::Instant::now() < deadline {
1080            std::thread::sleep(std::time::Duration::from_millis(10));
1081        }
1082        assert!(sub.is_connected(), "subscription should connect");
1083
1084        // Publish a message so the image is active
1085        while publ.offer(b"hello") < 0 {
1086            std::thread::yield_now();
1087        }
1088
1089        assert_eq!(sub.image_count(), 1);
1090        let image = sub.image_by_index(0).expect("image_by_index failed");
1091        assert!(image.session_id() != 0);
1092        assert!(image.position() >= 0);
1093        assert!(!image.is_closed());
1094        assert!(!image.is_end_of_stream());
1095
1096        // Test image_by_session_id
1097        let sid = image.session_id();
1098        let image2 = sub
1099            .image_by_session_id(sid)
1100            .expect("image_by_session_id failed");
1101        assert_eq!(image2.session_id(), sid);
1102    }
1103
1104    #[test]
1105    fn test_channel_builder_ipc() {
1106        assert_eq!(ChannelBuilder::ipc().build(), "aeron:ipc");
1107    }
1108
1109    #[test]
1110    fn test_channel_builder_udp() {
1111        let uri = ChannelBuilder::udp().endpoint("localhost:20121").build();
1112        assert_eq!(uri, "aeron:udp?endpoint=localhost:20121");
1113    }
1114
1115    #[test]
1116    fn test_channel_builder_multiple_params() {
1117        let uri = ChannelBuilder::udp()
1118            .endpoint("localhost:20121")
1119            .mtu(8192)
1120            .term_length(65536)
1121            .reliable(true)
1122            .build();
1123        assert_eq!(
1124            uri,
1125            "aeron:udp?endpoint=localhost:20121|mtu=8192|term-length=65536|reliable=true"
1126        );
1127    }
1128
1129    #[test]
1130    fn test_channel_builder_multicast() {
1131        let uri = ChannelBuilder::udp()
1132            .endpoint("224.0.1.1:40456")
1133            .interface("localhost")
1134            .ttl(4)
1135            .build();
1136        assert_eq!(
1137            uri,
1138            "aeron:udp?endpoint=224.0.1.1:40456|interface=localhost|ttl=4"
1139        );
1140    }
1141
1142    #[test]
1143    fn test_channel_builder_mdc() {
1144        let uri = ChannelBuilder::udp()
1145            .control("localhost:40456")
1146            .control_mode("dynamic")
1147            .build();
1148        assert_eq!(
1149            uri,
1150            "aeron:udp?control=localhost:40456|control-mode=dynamic"
1151        );
1152    }
1153
1154    #[test]
1155    fn test_channel_builder_custom_param() {
1156        let uri = ChannelBuilder::ipc().param("alias", "my-channel").build();
1157        assert_eq!(uri, "aeron:ipc?alias=my-channel");
1158    }
1159}
aeron_glide/lib.rs

aeron_glide/
lib.rs
