Crate rusteron_client

Expand description

§rusteron-client

rusteron-client is a core component of the Rusteron project.
It provides a Rust wrapper around the Aeron C client API, enabling high-performance, low-latency communication in distributed systems built with Rust.

This crate supports publishing, subscribing, and managing Aeron resources, while exposing a flexible and idiomatic interface over unsafe C bindings.
Due to its reliance on raw FFI, developers must take care to manage resource lifetimes and concurrency correctly.

§Sponsored by GSR

Rusteron is proudly sponsored and maintained by GSR, a global leader in algorithmic trading and market making in digital assets.

It powers mission-critical infrastructure in GSR’s real-time trading stack and is now developed under the official GSR GitHub organization as part of our commitment to open-source excellence and community collaboration.

We welcome contributions, feedback, and discussions. If you’re interested in integrating or contributing, please open an issue or reach out directly.

§Features

Client Setup – Create and start an Aeron client using Rust.
Publications – Send messages via offer() or try_claim().
Subscriptions – Poll for incoming messages and handle fragments.
Callbacks & Handlers – React to driver events like availability, errors, and stream lifecycle changes.
Cloneable Wrappers – All client types are cloneable and share ownership of the underlying C resources.
Automatic Resource Management – Objects created with .new() automatically call *_init and *_close, where supported.
Result-Focused API – Methods returning primitive C results return Result<T, AeronCError> for ergonomic error handling.
Efficient String Interop – Inputs use &CStr, outputs return &str, giving developers precise allocation control.

§Installation

Add rusteron-client to your Cargo.toml:

# Dynamic linking (default)
rusteron-client = "0.2"

# Static linking
rusteron-client = { version = "0.2", features = ["static"] }

# Static linking with precompiled C libraries (best for Mac users, no Java/cmake needed)
rusteron-client = { version = "0.2", features = ["static", "precompile"] }

# Static linking with precompiled C libraries using rustls downloader
rusteron-client = { version = "0.2", features = ["static", "precompile-rustls"] }

When using the default dynamic configuration, you must ensure Aeron C libraries are available at runtime. The static option embeds them automatically into the binary.

§General Patterns

new() Initialization: Automatically calls the corresponding *_init method.
One-liner connect: Aeron::connect(Some(dir))? builds the context, client, and starts the conductor; build the AeronContext yourself for tuned setups. (AeronArchive::connect(...) in rusteron-archive.)
Retained images auto-release: subscription.image_at_index(i) / image_by_session_id(id) return Option<AeronImage> that releases back to the subscription on drop; for_each_image(|img| …) borrows without bookkeeping.
Automatic Cleanup (Partial): When possible, Drop will invoke the appropriate *_close or *_destroy methods.
Manual Resource Responsibility: For methods like set_aeron() or where lifetimes aren’t managed internally, users are responsible for safety.
Handlers Are Reference-Counted: Wrap callbacks with Handler::new(...). Methods that register a callback the C client retains keep a clone alive inside the registering resource, so the value is freed automatically — no manual release().

§Handlers and Callbacks

Two kinds of callbacks:

§1. Retained (error handler, image/counter lifecycle, notifications)

The C client stores these and invokes them later from the conductor thread. Setters take the callback by value (a closure or any impl Trait), heap-allocate it into a reference-counted Handler, and keep a clone inside the registering resource — it is freed when that resource drops, no manual release():

ctx.set_error_handler(Some(|code: i32, msg: &str| eprintln!("aeron error {code}: {msg}")))?;

Keep the returned Handler to read the callback’s state later:

let counts = ctx.set_error_handler(Some(ErrorCounter::default()))?.unwrap();
// ... later
println!("{} errors", counts.error_count);

Registration methods that return a resource (e.g. add_subscription with image handlers) take Option<&Handler<T>> and clone it into the resource — same lifetime guarantee.

§2. Synchronous (`poll`, `controlled_poll`, log readers)

Invoked only during the call, so they take a stack closure — zero allocation, may borrow local state:

subscription.poll_fn(|buf: &[u8], header: AeronHeader| {
    println!("received {} bytes", buf.len());
}, 10)?;

For messages larger than the MTU, wrap the delegate in an AeronFragmentAssembler (it reassembles fragments before calling back) — poll_fn delivers raw fragments only. The ergonomic wrapper is AeronFragmentClosureAssembler; its poll borrows a &mut T context for the call (the callback is a fn pointer, not a closure, so pass state through the context):

use rusteron_client::{AeronFragmentClosureAssembler, AeronHeader};

struct Stats { bytes: u64 }
fn on_msg(stats: &mut Stats, buf: &[u8], _hdr: AeronHeader) { stats.bytes += buf.len() as u64; }

let mut assembler = AeronFragmentClosureAssembler::new()?;
let mut stats = Stats { bytes: 0 };
assembler.poll(&subscription, &mut stats, on_msg, 10)?; // 10 = fragment limit
// stats.bytes now holds the reassembled payload sizes

No callback for an optional slot? Handlers::NONE fits any callback parameter.

§Building channel URIs

Prefer the typed AeronUriStringBuilder over hand-written URI strings — parameters are typed setters, so misspelled keys and malformed values are caught before they reach the driver:

let channel = AeronUriStringBuilder::udp("localhost:20121")?.build(256)?;
let mds_sub = AeronUriStringBuilder::udp_control("localhost:9998", ControlMode::Manual)?.build(256)?;
let response = AeronUriStringBuilder::udp_control("localhost:9999", ControlMode::Response)?
    .response_correlation_id(id)?
    .build(256)?;

Constructors: ipc(), udp(endpoint), udp_control(control, ControlMode); plus ~50 typed setters (session_id, mtu_length, term_length, fc, gtag, reliable, …). See examples/multi_destination_subscription.rs and examples/request_response.rs.

§Minimal Pub/Sub

use rusteron_client::*;
use std::time::Duration;

let ctx = AeronContext::new()?;
// Reuse the built-in logger for async client errors (Aeron samples always set one).
ctx.set_error_handler(Some(AeronErrorHandlerLogger))?;
let aeron = Aeron::new(&ctx)?;
aeron.start()?;

let channel = c"aeron:ipc"; // compile-time &CStr, zero runtime cost
let publication = aeron
    .async_add_publication(channel, 123)?
    .poll_blocking(Duration::from_secs(5))?;
let subscription = aeron
    .async_add_subscription(
        channel, 123,
        Handlers::NONE,
        Handlers::NONE,
    )?
    .poll_blocking(Duration::from_secs(5))?;

// offer() returns Ok(position) or a typed AeronOfferError — see "Errors & offer results".
loop {
    match publication.offer(b"hello") {
        Ok(_) => break,
        Err(e) if e.is_retryable() => continue,
        Err(e) => return Err(e.into()),
    }
}
subscription.poll_fn(|buf: &[u8], _hdr: AeronHeader| println!("got {} bytes", buf.len()), 10)?;

Note on poll_blocking / add_*(.., timeout): these block the calling thread in a busy-poll loop and exist for example/test brevity. In production, drive the async poller’s poll() from your own event loop instead. See examples/non_blocking_publisher.rs for the idiomatic pattern.

§AddressSanitizer

The sanitize-address feature compiles the Aeron C sources with -fsanitize=address (implies build-from-source), so use-after-free / double-free across the FFI boundary are caught in your tests:

# macOS — stable rustc; Apple's ASan runtime is linked automatically, it just needs to be
# on the loader path:
export DYLD_LIBRARY_PATH="$(clang -print-resource-dir)/lib/darwin"
ASAN_OPTIONS=detect_leaks=0 cargo test --features rusteron-client/sanitize-address

# Linux — nightly, additionally instruments the Rust side:
RUSTFLAGS="-Zsanitizer=address" cargo +nightly test -Zbuild-std \
  --target x86_64-unknown-linux-gnu --features rusteron-client/sanitize-address

Notes: leak detection is noisy with a media driver in-process — keep detect_leaks=0; other sanitizers (thread, undefined) can be selected with the RUSTERON_SANITIZE env var instead of the feature. In this repo, just test-asan runs the client suite this way.

§Resource Lifecycle & close()

The C client owns every child resource (publications, subscriptions, counters) and frees them all in aeron_close. rusteron therefore reference-counts teardown:

aeron.close() (and drop) is deferred while any child handle or clone is alive — the surviving handles remain fully usable; the C close runs when the last reference releases.
Leaf resources (publication.close() etc.) close immediately; clones of that handle see a nulled pointer and become inert.
unsafe aeron.close_now() forces the C close immediately. Every surviving child handle then dangles — even its Drop is a double free — so mem::forget survivors or never return.

Which structures get which behaviour:

Behaviour	Types
Deferred `close()` + `unsafe close_now()`	`Aeron`, `AeronArchive` (rusteron-archive)
Immediate `close(self)` (+ `close_with_handler` for a close-complete notification); clones become inert	`AeronPublication`, `AeronExclusivePublication`, `AeronSubscription`, `AeronCounter`
No public close — freed when the last reference drops (the client holds an internal clone, so they always outlive it)	`AeronContext`, `AeronArchiveContext`, `AeronDriverContext`
No close at all — the C client frees them when their `poll()` completes (created, errored, or cancelled)	`AeronAsyncAdd*` pollers, `AeronAsyncDestination`

Ordering is structurally safe in any combination: children hold references to the client, the client holds its context, and closing/dropping in any order defers the C teardown until the last handle releases (verified by the complex_object_graph_close_is_safe_in_any_order test).

§Migrating 0.1 → 0.2

Breaking changes, made because the old design allowed double frees and use-after-free (see PR #50):

0.1	0.2
`Handler::leak(h)` + manual `release()`	`Handler::new(h)`; freed automatically (`Arc`-counted, resources keep clones)
`ctx.set_error_handler(Some(&handler))`	`ctx.set_error_handler(Some(h))` — takes the value (closures work), returns the `Handler`
`set_error_handler_fn(closure)` (UB: C retained a stack closure)	removed; `_once` now exists only for synchronous callbacks
`aeron.close()` freed children immediately (UAF on surviving handles)	deferred until the last reference drops; `unsafe close_now()` is the escape hatch
`Handler` was `Sync`	`Send` only (conductor thread may invoke callbacks)
`offer(buf, supplier)` returned a raw `i64` sentinel	`offer(buf)` → `Result<i64, AeronOfferError>` with `is_retryable()`; supplier form: `offer_with_reserved_value`; raw sentinel: `offer_raw`
`offer_result` / `offer_result_simple` / `try_claim_result` (→ `AeronCError`)	`offer_with_reserved_value` / `offer` / `try_claim` (→ typed `AeronOfferError`)
`poll_fn(f, limit)`	`poll_fn(f, limit)` — `for_each_fragment` removed
`Handlers::no_xxx_handler()` per callback	`Handlers::NONE` for any callback parameter

The full old → new table (destinations, fragment assembler, C strings, driver guard) lives in the root README’s migration guide.

§Examples

examples/basic_pub_sub.rs — minimal pub/sub with fragment assembly
examples/streaming_rate.rs — streaming publisher + rate-reporting subscriber (port of streaming_publisher.c + rate_subscriber.c)
examples/multi_destination_subscription.rs — MDS: one manual-control subscription aggregating several endpoints (port of basic_mds_subscriber.c)
examples/driver_stats.rs — CnC tooling: counters, distinct error log, loss report (ports of aeron_stat.c / error_stat.c / loss_stat.c)
examples/embedded_ping_pong.rs — RTT ping/pong with try_claim (port of cping/cpong)
examples/embedded_exclusive_ipc_throughput.rs — exclusive-publication IPC throughput
examples/request_response.rs — response channels (aeron 1.44+): request/response wiring via control-mode=response + response-correlation-id (port of response_server.c/response_client.c)
examples/file_transfer.rs — chunked file transfer with fragment reassembly and verification (port of FileSender/FileReceiver)

§Documentation & Guides

For detailed guides and code snippets on Aeron features in Rust, see:

§Errors & offer results

Client errors: install an error handler on the context (ctx.set_error_handler(Some(handler))) so async errors aren’t silently lost — Aeron’s samples always do.
offer() / try_claim() results: Result<i64, AeronOfferError> — Ok is the new log position; the error is a typed sentinel with is_retryable() (back-pressured, admin action, not connected — retry, ideally with an idle strategy) vs fatal (closed, max position exceeded). This mirrors Aeron’s BasicPublisher.checkResult without magic numbers:
ⓘ
```
match publication.offer(msg) {
    Ok(_) => {}
    Err(e) if e.is_retryable() => idle.idle(), // retry
    Err(e) => return Err(e.into()),            // publication gone
}
```
For branch-free hot paths, offer_raw() / try_claim_raw() return the raw i64 sentinel. The typed path costs the same on the happy path — the error enum only materialises on the error path.
Image handlers: Handlers::NONE for each image slot is a fine default, but real apps usually react to image availability (logging, synchronisation) — Aeron’s Ping sample uses one as a latch.

§Idle strategies

Poll loops should back off when a cycle does no work. Rusteron ports Aeron’s IdleStrategy (idle(work_count) returns immediately when work was done, otherwise backs off):

use rusteron_client::{BackoffIdleStrategy, IdleStrategy};

let mut idle = BackoffIdleStrategy::new(); // spin → yield → sleep, Aeron's default
loop {
    let fragments = subscription.poll(Some(&handler), 10)?;
    if fragments == 0 { /* break when done */ }
    idle.idle(fragments);
}

Available: BusySpinIdleStrategy (lowest latency, pins a core), YieldingIdleStrategy, SleepingIdleStrategy (fixed sleep), BackoffIdleStrategy (adaptive, general-purpose), NoOpIdleStrategy. Latency benchmarks should keep busy-spin.

BackoffIdleStrategy is the default and matches Aeron’s C/Java backoff exactly (10 spins, 20 yields, park 1µs..1ms), so there is no need for a C-backed idle strategy to get parity. The conductor’s own idle strategy is configured on the context — use the typed enum (ctx.set_idle_strategy_kind(AeronIdleStrategyKind::Backoff)?, which also sets coherent init args) — and likewise the media driver’s per-agent strategies (set_conductor_idle_strategy_kind etc. on AeronDriverContext).

For recording, replay, and persistent subscriptions (replay history, then seamlessly join a live stream), see rusteron-archive.

§Contributing & License

See the root README and CONTRIBUTING.md. Dual-licensed under MIT or Apache-2.0.

§Acknowledgments

Special thanks to:

@mimran1980, a core low-latency developer at GSR and the original creator of Rusteron - your work made this possible!
@bspeice for the original libaeron-sys
The Aeron community for open protocol excellence

§Features

static: When enabled, this feature statically links the Aeron C code. By default, the library uses dynamic linking to the Aeron C libraries.
backtrace: When enabled will log a backtrace for each AeronCError
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
log-c-bindings: When enabled will log every C binding call with arguments and return values. Useful for debugging FFI interactions
precompile: When enabled will use precompiled C code instead of requiring cmake and java to be installed

Modules§

bindings

Macros§

cformat: format! for C strings: builds the formatted String and converts it to a CString in one visibly-named step.

Structs§

Aeron
AeronAgentCloseFuncLogger
AeronAgentDoWorkFuncLogger
AeronAgentRunner
AeronAgentStartFuncLogger
AeronAsyncAddCounter
AeronAsyncAddExclusivePublication
AeronAsyncAddPublication
AeronAsyncAddSubscription
AeronAsyncDestination
AeronAsyncDestinationById
AeronAsyncGetNextAvailableSessionId
AeronAvailableCounterLogger
AeronAvailableCounterPair
AeronAvailableImageLogger
AeronBlockHandlerLogger
AeronBufferClaim: Structure used to hold information for a try_claim function call.
AeronCError: Aeron C API error: code + optional message.
AeronClaim: Zero-copy claim on a publication’s term buffer with a RAII commit-or-abort lifecycle.
AeronClientRegisteringResource
AeronCloseClientLogger
AeronCloseClientPair
AeronCnc
AeronCncConstants
AeronCncMetadata
AeronContext
AeronControlledFragmentAssembler
AeronControlledFragmentClosureAssembler
AeronControlledFragmentHandlerLogger
AeronCounter
AeronCounterConstants: Configuration for a counter that does not change during its lifetime.
AeronCounterMetadataDescriptor
AeronCounterValueDescriptor
AeronCountersReader
AeronCountersReaderBuffers
AeronCountersReaderForeachCounterFuncLogger
AeronDataHeader
AeronDataHeaderAsLongs
AeronError
AeronErrorHandlerLogger
AeronErrorLogReaderFuncLogger
AeronErrorLogger: Production error handler that routes Aeron async errors through the Rust log facade at error level with a concise format — the recommended default for AeronContext::set_error_handler.
AeronExclusivePublication
AeronFprintfHandlerLogger
AeronFragmentAssembler
AeronFragmentClosureAssembler
AeronFragmentHandlerLogger
AeronFrameHeader
AeronHeader
AeronHeaderValues
AeronHeaderValuesFrame
AeronIdleStrategy
AeronIdleStrategyFuncLogger
AeronImage
AeronImageConstants: Configuration for an image that does not change during it’s lifetime.
AeronImageControlledFragmentAssembler
AeronImageFragmentAssembler
AeronIovec
AeronIpcChannelParams
AeronLogBuffer
AeronLogbufferMetadata
AeronLossReporter
AeronLossReporterEntry
AeronLossReporterReadEntryFuncLogger
AeronMappedBuffer
AeronMappedFile
AeronMappedRawLog
AeronNakHeader
AeronNewPublicationLogger
AeronNewSubscriptionLogger
AeronNotificationLogger
AeronOptionHeader
AeronPerThreadError
AeronPublication
AeronPublicationConstants: Configuration for a publication that does not change during it’s lifetime.
AeronPublicationErrorFrameHandlerLogger
AeronPublicationErrorValues
AeronReservedValueSupplierLogger
AeronResolutionHeader
AeronResolutionHeaderIpv4
AeronResolutionHeaderIpv6
AeronResponseSetupHeader
AeronRttmHeader
AeronSetupHeader
AeronStatusMessageHeader
AeronStatusMessageOptionalHeader
AeronStatusTracker: Status transition tracker that emits only when the observed AeronStatus differs from the previously recorded state.
AeronStrToPtrHashMap
AeronStrToPtrHashMapForEachFuncLogger
AeronStrToPtrHashMapKey
AeronSubscription
AeronSubscriptionConstants
AeronUdpChannelParams
AeronUnavailableCounterLogger
AeronUnavailableCounterPair
AeronUnavailableImageLogger
AeronUri
AeronUriParam
AeronUriParams
AeronUriParseCallbackLogger
AeronUriStringBuilder
BackoffIdleStrategy: Adaptive backoff: spin a few times, then yield a few times, then sleep with an exponentially growing park up to a max. A good general-purpose strategy. (Aeron BackoffIdleStrategy.)
BusySpinIdleStrategy: Spin with a CPU pause hint. Lowest latency, pins a core. (Aeron BusySpinIdleStrategy.)
ChannelUri: Represents the Aeron URI parser and handler.
FnMutControlledMessageHandler
FnMutMessageHandler
Handler: Handler
Handlers: Utility method for setting empty handlers
ManagedCResource: A custom struct for managing C resources with automatic cleanup.
NoHandler: Type-level “no callback” sentinel.
NoOpIdleStrategy: No-op — never yields the core. (Aeron NoOpIdleStrategy.)
SleepingIdleStrategy: Sleep for a fixed duration when idle. (Aeron SleepingIdleStrategy.)
YieldingIdleStrategy: Yield the OS thread when idle. Lower CPU than busy-spin, slightly higher latency. (Aeron YieldingIdleStrategy.)

Enums§

AeronErrorType
AeronIdleStrategyKind: Aeron’s named idle strategies, as accepted by the context / media-driver set_*_idle_strategy options (the C aeron_idle_strategy_load symbol table).
AeronOfferError: Typed error for offer / try_claim on a publication.
AeronStatus: High-level connection state of a publication or subscription.
AeronSystemCounterType
CResource
ControlMode: Enum for control modes.
Media: Enum for media types.

Constants§

AERON_DIR_PROP_NAME
AERON_IPC_MEDIA
AERON_UDP_MEDIA
DRIVER_TIMEOUT_MS_DEFAULT
MAX_OFFER_PARTS: Max buffer parts accepted by AeronPublication::offer_parts / AeronExclusivePublication::offer_parts — the aeron_iovec_t array is built on the stack, so it has a fixed capacity. Use the raw offerv with your own iovec array for larger gathers.
PUBLICATION_ADMIN_ACTION
PUBLICATION_BACK_PRESSURED
PUBLICATION_CLOSED
PUBLICATION_ERROR
PUBLICATION_MAX_POSITION_EXCEEDED
PUBLICATION_NOT_CONNECTED: Result codes returned by AeronPublication::offer / try_claim (Aeron aeronc.h). A positive value is the resulting log position; the negatives classify the failure.
SPY_PREFIX
TAG_PREFIX

Statics§

AERON_IPC_STREAM

Traits§

AeronAgentCloseFuncCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronAgentDoWorkFuncCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronAgentStartFuncCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronAvailableCounterCallback: Function called by aeron_client_t to deliver notifications that a counter has been added to the driver.
AeronAvailableImageCallback: Function called by aeron_client_t to deliver notifications that an aeron_image_t was added.
AeronBlockHandlerCallback: Callback for handling a block of messages being read from a log.
AeronCloseClientCallback: Function called by aeron_client_t to deliver notifications that the client is closing.
AeronControlledFragmentHandlerCallback: Callback for handling fragments of data being read from a log.
AeronCountersReaderForeachCounterFuncCallback: Function called by aeron_counters_reader_foreach_counter for each counter in the aeron_counters_reader_t.
AeronErrorHandlerCallback: The error handler to be called when an error occurs.
AeronErrorLogReaderFuncCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronFprintfHandlerCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronFragmentHandlerCallback: Callback for handling fragments of data being read from a log.
AeronIdleStrategyFuncCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronLossReporterReadEntryFuncCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronNewPublicationCallback: Function called by aeron_client_t to deliver notification that the media driver has added an aeron_publication_t or aeron_exclusive_publication_t successfully.
AeronNewSubscriptionCallback: Function called by aeron_client_t to deliver notification that the media driver has added an aeron_subscription_t successfully.
AeronNotificationCallback: Generalised notification callback.
AeronPublicationErrorFrameHandlerCallback: The error frame handler to be called when the driver notifies the client about an error frame being received. The data passed to this callback will only be valid for the lifetime of the callback. The user should use aeron_publication_error_values_copy if they require the data to live longer than that.
AeronReservedValueSupplierCallback: Function called when filling in the reserved value field of a message.
AeronStrToPtrHashMapForEachFuncCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronUnavailableCounterCallback: Function called by aeron_client_t to deliver notifications that a counter has been removed from the driver.
AeronUnavailableImageCallback: Function called by aeron_client_t to deliver notifications that an aeron_image_t has been removed from use and should not be used any longer.
AeronUriParseCallbackCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
ControlledFragmentAssemblable: A poll target whose raw fragments can be reassembled by AeronControlledFragmentAssembler.
FragmentAssemblable: A poll target whose raw fragments can be reassembled by AeronFragmentAssembler. Implemented for AeronSubscription here and for archive types in rusteron-archive.
IdleStrategy: Back off a poll loop when the last cycle did no work. Mirrors Aeron’s IdleStrategy: idle(work_count) returns immediately when work_count > 0, otherwise spins / yields / sleeps depending on the implementation.
IntoCString

Functions§

find_unused_udp_port
is_udp_port_available
validate_endpoint_for_aeron_udp: Validate a UDP channel URI endpoint (host:port) for use with Aeron.

Type Aliases§

CleanupBox
RcOrArc: Reference-counting smart pointer: Rc by default, Arc under the multi-threaded feature. Swap is transparent — RcOrArc::new, .clone(), strong_count all work on both.
RefCellOrMutex