pub const CUSTOM_AERON_CODE: &str = "// code here is included in all modules and extends generated classes\npub static AERON_IPC_STREAM: &std::ffi::CStr = c\"aeron:ipc\";\n\n/// Max buffer parts accepted by [`AeronPublication::offer_parts`] /\n/// [`AeronExclusivePublication::offer_parts`] \u{2014} the `aeron_iovec_t` array is\n/// built on the stack, so it has a fixed capacity. Use the raw `offerv` with\n/// your own iovec array for larger gathers.\npub const MAX_OFFER_PARTS: usize = 8;\n\n// SAFETY: these handles wrap `Rc` (via `CResource::OwnedOnHeap`), so they are\n// `!Send + !Sync` in principle. `Rc` (non-atomic refcount) is kept for latency.\n// The supported usage pattern is to MOVE a handle to a single owning thread and\n// use it exclusively there; `Send` is retained to allow that move.\n//\n// `Sync` is intentionally NOT implemented by default: sharing `&Handle` across\n// threads would let two threads `Rc::clone` concurrently and race the refcount.\n//\n// `Send` over `Rc` is technically unsound (a non-atomic refcount touched from\n// more than one thread races). Callers must not clone these handles across\n// threads and must not use one handle from multiple threads concurrently. The\n// `multi-threaded` feature (below) switches to `Arc` and removes this caveat.\nunsafe impl Send for AeronCountersReader {}\nunsafe impl Send for AeronSubscription {}\nunsafe impl Send for AeronPublication {}\nunsafe impl Send for AeronCounter {}\n\n// SAFETY: under `multi-threaded` the refcount is atomic (`Arc`), so `Send` is sound,\n// and `Sync` is implemented so `&Handle` can be shared across threads. The `UnsafeCell`\n// fields inside `ManagedCResource` are mutated only during construction and close,\n// never during the shared-read window.\n//\n// This only lifts the Rust-side barrier; the caller must still confirm the underlying\n// Aeron object is thread-safe (e.g. `AeronPublication` is, `AeronExclusivePublication`\n// is not). See the README \"Multi-threaded handles\" section.\n// Enable with `features = [\"multi-threaded\"]` in Cargo.toml.\n#[cfg(feature = \"multi-threaded\")]\nunsafe impl Sync for AeronCountersReader {}\n#[cfg(feature = \"multi-threaded\")]\nunsafe impl Sync for AeronSubscription {}\n#[cfg(feature = \"multi-threaded\")]\nunsafe impl Sync for AeronPublication {}\n#[cfg(feature = \"multi-threaded\")]\nunsafe impl Sync for AeronCounter {}\n\n/// High-level connection state of a publication or subscription.\n///\n/// [`AeronPublication::status`] / [`AeronSubscription::status`] derive\n/// `Disconnected` / `Connected` / `Closed` from the handle\'s `is_closed` /\n/// `is_connected` flags. `BackPressured` is only observable at `offer` /\n/// `try_claim` time (via the returned error) and is surfaced by\n/// [`AeronStatus::from_error`].\n#[derive(Debug, Clone, Copy, PartialEq, Eq)]\npub enum AeronStatus {\n /// No publication/subscription image is currently connected.\n Disconnected,\n /// At least one image/publication is connected and the channel is usable.\n Connected,\n /// The publication is applying back-pressure; retry the offer shortly.\n BackPressured,\n /// The publication/subscription has been closed and is unusable.\n Closed,\n}\n\nimpl AeronStatus {\n /// Derive a status from an `offer` / `try_claim` error when it corresponds\n /// to a known transition (`BackPressured` / `Closed`). Returns `None` for\n /// errors that are not status-like (e.g. `MaxPositionExceeded`).\n pub fn from_error(error: &AeronOfferError) -> Option<Self> {\n match error {\n AeronOfferError::BackPressured => Some(Self::BackPressured),\n AeronOfferError::Closed => Some(Self::Closed),\n AeronOfferError::NotConnected => Some(Self::Disconnected),\n _ => None,\n }\n }\n}\n\n/// Aeron\'s named idle strategies, as accepted by the context / media-driver\n/// `set_*_idle_strategy` options (the C `aeron_idle_strategy_load` symbol table).\n///\n/// Strategy parameters are configured separately via `set_idle_strategy_init_args`:\n/// - `Sleeping`: sleep period in nanoseconds (e.g. `\"1000000\"` = 1ms)\n/// - `Backoff`: `maxSpins-maxYields-minParkNs-maxParkNs` (e.g. `\"10-20-1000-1000000\"`)\n/// - Others: init args ignored\n#[derive(Debug, Clone, Copy, PartialEq, Eq)]\npub enum AeronIdleStrategyKind {\n /// Sleep when idle (`\"sleeping\"`); period from the init args.\n Sleeping,\n /// Yield the OS thread when idle (`\"yield\"`).\n Yielding,\n /// Busy-spin (`\"spin\"`) \u{2014} lowest latency, pins a core.\n BusySpin,\n /// Never back off (`\"noop\"`).\n NoOp,\n /// Spin, then yield, then park with exponential backoff (`\"backoff\"`) \u{2014} the default.\n Backoff,\n}\n\nimpl AeronIdleStrategyKind {\n /// The name the C loader expects.\n pub const fn name(&self) -> &\'static str {\n match self {\n AeronIdleStrategyKind::Sleeping => \"sleeping\",\n AeronIdleStrategyKind::Yielding => \"yield\",\n AeronIdleStrategyKind::BusySpin => \"spin\",\n AeronIdleStrategyKind::NoOp => \"noop\",\n AeronIdleStrategyKind::Backoff => \"backoff\",\n }\n }\n\n /// [`Self::name`] as a compile-time C string \u{2014} lets setters pass it straight to\n /// the FFI with no runtime allocation.\n pub const fn name_c(&self) -> &\'static std::ffi::CStr {\n match self {\n AeronIdleStrategyKind::Sleeping => c\"sleeping\",\n AeronIdleStrategyKind::Yielding => c\"yield\",\n AeronIdleStrategyKind::BusySpin => c\"spin\",\n AeronIdleStrategyKind::NoOp => c\"noop\",\n AeronIdleStrategyKind::Backoff => c\"backoff\",\n }\n }\n}\n\nimpl AeronIdleStrategyKind {\n /// Default init args accepted by this strategy\'s loader (the C client validates the\n /// currently-set init args when the strategy is set, so they must be coherent):\n /// sleeping takes a period in ns; backoff takes `maxSpins-maxYields-minParkNs-maxParkNs`\n /// (hyphen-separated); the rest ignore init args.\n pub const fn default_init_args(&self) -> &\'static str {\n match self {\n AeronIdleStrategyKind::Sleeping => \"1000000\", // 1ms\n AeronIdleStrategyKind::Backoff => \"10-20-1000-1000000\", // aeron defaults\n _ => \"\",\n }\n }\n\n /// [`Self::default_init_args`] as a compile-time C string (no runtime allocation).\n pub const fn default_init_args_c(&self) -> &\'static std::ffi::CStr {\n match self {\n AeronIdleStrategyKind::Sleeping => c\"1000000\", // 1ms\n AeronIdleStrategyKind::Backoff => c\"10-20-1000-1000000\", // aeron defaults\n _ => c\"\",\n }\n }\n}\n\nimpl Aeron {\n /// Connect to a media driver in one call: context, client, and conductor start.\n ///\n /// `dir` is the media driver directory (`None` uses the aeron default). For tuned\n /// setups \u{2014} error handlers, driver timeout, idle strategy \u{2014} build the\n /// [`AeronContext`] yourself and use [`Aeron::new`].\n pub fn connect(dir: Option<&str>) -> Result<Aeron, AeronCError> {\n let ctx = AeronContext::new()?;\n if let Some(dir) = dir {\n ctx.set_dir(&dir.into_c_string())?;\n }\n let aeron = Aeron::new(&ctx)?;\n aeron.start()?;\n Ok(aeron)\n }\n\n /// Connect to a media driver in the default aeron directory. Equivalent to\n /// [`Self::connect`]`(None)`, but reads more naturally at the call site.\n #[inline]\n pub fn connect_default() -> Result<Aeron, AeronCError> {\n Self::connect(None)\n }\n\n /// Connect to a media driver in `dir`. Equivalent to\n /// [`Self::connect`]`(Some(dir))`.\n #[inline]\n pub fn connect_dir(dir: &str) -> Result<Aeron, AeronCError> {\n Self::connect(Some(dir))\n }\n}\n\nimpl AeronContext {\n /// Typed variant of [`Self::set_idle_strategy`]: configures the conductor\'s idle\n /// strategy without stringly-typed names, setting coherent default init args\n /// (override afterwards with [`Self::set_idle_strategy_init_args`] if needed).\n pub fn set_idle_strategy_kind(&self, kind: AeronIdleStrategyKind) -> Result<i32, AeronCError> {\n self.set_idle_strategy_init_args(kind.default_init_args_c())?;\n self.set_idle_strategy(kind.name_c())\n }\n}\n\n/// **Counters and Control (CnC) file** \u{2014} the shared-memory interface between the\n/// media driver and its clients.\n///\n/// The CnC file contains:\n/// - **Counters**: stream positions, recording positions, liveness indicators,\n/// and application-defined counters.\n/// - **Error log**: a ring-buffer of recent driver/client errors.\n/// - **Loss reporter**: observed packet-loss entries.\n/// - **To-driver / to-clients** command buffers.\n///\n/// `AeronCnc` is a read-only view of the CnC file \u{2014} use it to inspect driver\n/// state, dump counters, or read the error log (the upstream `aeron_stat`,\n/// `error_stat`, and `loss_stat` tools all do this).\n///\n/// # Two access patterns\n///\n/// **Scoped read** (zero allocation, preferred for one-shot queries):\n/// ```ignore\n/// AeronCnc::read(driver_ctx.get_dir(), |cnc| {\n/// cnc.foreach_counter_fn(|value, id, type_id, key, label| {\n/// println!(\"{id}: {label} = {value}\");\n/// });\n/// })?;\n/// ```\n/// Opens the CnC file, runs the closure, closes \u{2014} the resource wrapper lives on\n/// the stack and is freed immediately after.\n///\n/// **Owned handle** (for repeated polling, e.g. a stats dashboard):\n/// ```ignore\n/// let cnc = AeronCnc::open(driver_ctx.get_dir())?;\n/// loop {\n/// let heartbeat = cnc.to_driver_heartbeat();\n/// // ... poll counters, error log, etc. ...\n/// sleep(Duration::from_secs(1));\n/// }\n/// // closed on drop\n/// ```\nimpl AeronCnc {\n /// Open the CnC file, run `handler`, close \u{2014} **zero heap allocation** for the\n /// resource wrapper (the C struct is stack-borrowed). Preferred for one-shot\n /// reads (dump counters, print the error log). Accepts `&CStr` so `c\"\"`\n /// literals work directly.\n #[inline]\n pub fn read(aeron_dir: &std::ffi::CStr, mut handler: impl FnMut(&mut AeronCnc)) -> Result<(), AeronCError> {\n let cnc = ManagedCResource::initialise(move |cnc| unsafe { aeron_cnc_init(cnc, aeron_dir.as_ptr(), 0) })?;\n let mut cnc = Self {\n inner: CResource::Borrowed(cnc),\n };\n handler(&mut cnc);\n unsafe { aeron_cnc_close(cnc.get_inner()) };\n Ok(())\n }\n\n /// Open the CnC file and return an **owned handle** you can store and poll\n /// repeatedly (e.g. a monitoring loop). Allocates the resource wrapper on\n /// the heap; closed on drop. Accepts `&CStr` so `c\"\"` literals work.\n #[inline]\n pub fn open(aeron_dir: &std::ffi::CStr) -> Result<AeronCnc, AeronCError> {\n let resource = ManagedCResource::new(\n move |cnc| unsafe { aeron_cnc_init(cnc, aeron_dir.as_ptr(), 0) },\n Some(Box::new(move |cnc| unsafe {\n aeron_cnc_close(*cnc);\n 0\n })),\n false,\n )?;\n\n let result = Self {\n inner: CResource::OwnedOnHeap(RcOrArc::new(resource)),\n };\n Ok(result)\n }\n\n #[doc = \" Gets the timestamp of the last heartbeat sent to the media driver from any client.\\n\\n @param aeron_cnc to query\\n @return last heartbeat timestamp in ms.\"]\n #[inline]\n pub fn get_to_driver_heartbeat_ms(&self) -> Result<i64, AeronCError> {\n unsafe {\n let timestamp = aeron_cnc_to_driver_heartbeat(self.get_inner());\n if timestamp >= 0 {\n return Ok(timestamp);\n } else {\n return Err(AeronCError::from_code(timestamp as i32));\n }\n }\n }\n}\n\nimpl AeronHeader {\n /// returns AeronImage, **must** be called in poll method\n pub fn image(&self) -> Option<AeronImage> {\n let ptr = self.context();\n if ptr.is_null() {\n None\n } else {\n Some(AeronImage::from(ptr as *mut aeron_image_t))\n }\n }\n\n /// Session id of this fragment, or `None` if the underlying values lookup\n /// failed. Collapses the `get_values().frame().session_id()` hop and never\n /// panics on the fast path.\n #[inline]\n pub fn session_id(&self) -> Option<i32> {\n self.get_values().ok().map(|v| v.frame().session_id())\n }\n\n /// Stream id of this fragment, or `None` if the underlying values lookup\n /// failed. Collapses the `get_values().frame().stream_id()` hop and never\n /// panics on the fast path.\n #[inline]\n pub fn stream_id(&self) -> Option<i32> {\n self.get_values().ok().map(|v| v.frame().stream_id())\n }\n\n /// Reserved value of this fragment, or `None` if the underlying values\n /// lookup failed. A sender can stamp a timestamp here (see\n /// [`AeronPublication::offer_timestamped`]) so the receiver can measure\n /// end-to-end latency.\n #[inline]\n pub fn reserved_value(&self) -> Option<i64> {\n self.get_values().ok().map(|v| v.frame().reserved_value())\n }\n\n /// Term id of this fragment, or `None` if the underlying values lookup\n /// failed.\n #[inline]\n pub fn term_id(&self) -> Option<i32> {\n self.get_values().ok().map(|v| v.frame().term_id())\n }\n\n /// Term offset of this fragment, or `None` if the underlying values lookup\n /// failed.\n #[inline]\n pub fn term_offset(&self) -> Option<i32> {\n self.get_values().ok().map(|v| v.frame().term_offset())\n }\n}\n\n#[repr(u32)]\n#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq)]\npub enum AeronSystemCounterType {\n /// Running total of bytes sent for data over UDP, excluding IP headers.\n BytesSent = 0,\n /// Running total of bytes received for data over UDP, excluding IP headers.\n BytesReceived = 1,\n /// Failed offers to the receiver proxy suggesting back-pressure.\n ReceiverProxyFails = 2,\n /// Failed offers to the sender proxy suggesting back-pressure.\n SenderProxyFails = 3,\n /// Failed offers to the driver conductor proxy suggesting back-pressure.\n ConductorProxyFails = 4,\n /// Count of NAKs sent back to senders requesting re-transmits.\n NakMessagesSent = 5,\n /// Count of NAKs received from receivers requesting re-transmits.\n NakMessagesReceived = 6,\n /// Count of status messages sent back to senders for flow control.\n StatusMessagesSent = 7,\n /// Count of status messages received from receivers for flow control.\n StatusMessagesReceived = 8,\n /// Count of heartbeat data frames sent to indicate liveness in the absence of data to send.\n HeartbeatsSent = 9,\n /// Count of heartbeat data frames received to indicate liveness in the absence of data to send.\n HeartbeatsReceived = 10,\n /// Count of data packets re-transmitted as a result of NAKs.\n RetransmitsSent = 11,\n /// Count of packets received which under-run the current flow control window for images.\n FlowControlUnderRuns = 12,\n /// Count of packets received which over-run the current flow control window for images.\n FlowControlOverRuns = 13,\n /// Count of invalid packets received.\n InvalidPackets = 14,\n /// Count of errors observed by the driver and an indication to read the distinct error log.\n Errors = 15,\n /// Count of socket send operations which resulted in less than the packet length being sent.\n ShortSends = 16,\n /// Count of attempts to free log buffers no longer required by the driver that are still held by clients.\n FreeFails = 17,\n /// Count of the times a sender has entered the state of being back-pressured when it could have sent faster.\n SenderFlowControlLimits = 18,\n /// Count of the times a publication has been unblocked after a client failed to complete an offer within a timeout.\n UnblockedPublications = 19,\n /// Count of the times a command has been unblocked after a client failed to complete an offer within a timeout.\n UnblockedCommands = 20,\n /// Count of the times the channel endpoint detected a possible TTL asymmetry between its config and a new connection.\n PossibleTtlAsymmetry = 21,\n /// Current status of the ControllableIdleStrategy if configured.\n ControllableIdleStrategy = 22,\n /// Count of the times a loss gap has been filled when NAKs have been disabled.\n LossGapFills = 23,\n /// Count of the Aeron clients that have timed out without a graceful close.\n ClientTimeouts = 24,\n /// Count of the times a connection endpoint has been re-resolved resulting in a change.\n ResolutionChanges = 25,\n /// The maximum time spent by the conductor between work cycles.\n ConductorMaxCycleTime = 26,\n /// Count of the number of times the cycle time threshold has been exceeded by the conductor in its work cycle.\n ConductorCycleTimeThresholdExceeded = 27,\n /// The maximum time spent by the sender between work cycles.\n SenderMaxCycleTime = 28,\n /// Count of the number of times the cycle time threshold has been exceeded by the sender in its work cycle.\n SenderCycleTimeThresholdExceeded = 29,\n /// The maximum time spent by the receiver between work cycles.\n ReceiverMaxCycleTime = 30,\n /// Count of the number of times the cycle time threshold has been exceeded by the receiver in its work cycle.\n ReceiverCycleTimeThresholdExceeded = 31,\n /// The maximum time spent by the NameResolver in one of its operations.\n NameResolverMaxTime = 32,\n /// Count of the number of times the time threshold has been exceeded by the NameResolver.\n NameResolverTimeThresholdExceeded = 33,\n /// The version of the media driver.\n AeronVersion = 34,\n /// The total number of bytes currently mapped in log buffers, the CnC file, and the loss report.\n BytesCurrentlyMapped = 35,\n /// A minimum bound on the number of bytes re-transmitted as a result of NAKs.\\n///\\n/// MDC retransmits are only counted once; therefore, this is a minimum bound rather than the actual number\\n/// of retransmitted bytes. Note that retransmitted bytes are not included in the `BytesSent` counter value.\n RetransmittedBytes = 36,\n /// A count of the number of times that the retransmit pool has been overflowed.\n RetransmitOverflow = 37,\n /// A count of the number of error frames received by this driver.\n ErrorFramesReceived = 38,\n /// A count of the number of error frames sent by this driver.\n ErrorFramesSent = 39,\n DummyLast = 40,\n}\n\nimpl std::convert::TryFrom<i32> for AeronSystemCounterType {\n type Error = AeronCError;\n\n fn try_from(value: i32) -> Result<Self, Self::Error> {\n if value < 0 {\n return Err(AeronCError::from_code(value));\n }\n match value as u32 {\n 0 => Ok(AeronSystemCounterType::BytesSent),\n 1 => Ok(AeronSystemCounterType::BytesReceived),\n 2 => Ok(AeronSystemCounterType::ReceiverProxyFails),\n 3 => Ok(AeronSystemCounterType::SenderProxyFails),\n 4 => Ok(AeronSystemCounterType::ConductorProxyFails),\n 5 => Ok(AeronSystemCounterType::NakMessagesSent),\n 6 => Ok(AeronSystemCounterType::NakMessagesReceived),\n 7 => Ok(AeronSystemCounterType::StatusMessagesSent),\n 8 => Ok(AeronSystemCounterType::StatusMessagesReceived),\n 9 => Ok(AeronSystemCounterType::HeartbeatsSent),\n 10 => Ok(AeronSystemCounterType::HeartbeatsReceived),\n 11 => Ok(AeronSystemCounterType::RetransmitsSent),\n 12 => Ok(AeronSystemCounterType::FlowControlUnderRuns),\n 13 => Ok(AeronSystemCounterType::FlowControlOverRuns),\n 14 => Ok(AeronSystemCounterType::InvalidPackets),\n 15 => Ok(AeronSystemCounterType::Errors),\n 16 => Ok(AeronSystemCounterType::ShortSends),\n 17 => Ok(AeronSystemCounterType::FreeFails),\n 18 => Ok(AeronSystemCounterType::SenderFlowControlLimits),\n 19 => Ok(AeronSystemCounterType::UnblockedPublications),\n 20 => Ok(AeronSystemCounterType::UnblockedCommands),\n 21 => Ok(AeronSystemCounterType::PossibleTtlAsymmetry),\n 22 => Ok(AeronSystemCounterType::ControllableIdleStrategy),\n 23 => Ok(AeronSystemCounterType::LossGapFills),\n 24 => Ok(AeronSystemCounterType::ClientTimeouts),\n 25 => Ok(AeronSystemCounterType::ResolutionChanges),\n 26 => Ok(AeronSystemCounterType::ConductorMaxCycleTime),\n 27 => Ok(AeronSystemCounterType::ConductorCycleTimeThresholdExceeded),\n 28 => Ok(AeronSystemCounterType::SenderMaxCycleTime),\n 29 => Ok(AeronSystemCounterType::SenderCycleTimeThresholdExceeded),\n 30 => Ok(AeronSystemCounterType::ReceiverMaxCycleTime),\n 31 => Ok(AeronSystemCounterType::ReceiverCycleTimeThresholdExceeded),\n 32 => Ok(AeronSystemCounterType::NameResolverMaxTime),\n 33 => Ok(AeronSystemCounterType::NameResolverTimeThresholdExceeded),\n 34 => Ok(AeronSystemCounterType::AeronVersion),\n 35 => Ok(AeronSystemCounterType::BytesCurrentlyMapped),\n 36 => Ok(AeronSystemCounterType::RetransmittedBytes),\n 37 => Ok(AeronSystemCounterType::RetransmitOverflow),\n 38 => Ok(AeronSystemCounterType::ErrorFramesReceived),\n 39 => Ok(AeronSystemCounterType::ErrorFramesSent),\n 40 => Ok(AeronSystemCounterType::DummyLast),\n _ => Err(AeronCError::from_code(-1)),\n }\n }\n}\n\n// SAFETY: `aeron_mapped_file_t` is a plain `{ addr: *mut c_void, length: u64 }`.\n// It is touched only inside the init closure (runs once at construction) and the\n// cleanup closure (runs once at drop) of `AeronCncMetadata::load_from_file` \u{2014} never\n// concurrently. This matches the `unsafe impl Send` policy for `ManagedCResource<T>`\n// and the handle types at the top of this file, and unblocks the cleanup closure\n// under the `multi-threaded` feature where `CleanupBox<T>: Send`.\nunsafe impl Send for aeron_mapped_file_t {}\n\nimpl AeronCncMetadata {\n #[inline]\n /// allocates on heap\n pub fn load_from_file(aeron_dir: &str) -> Result<Self, AeronCError> {\n let aeron_dir = std::ffi::CString::new(aeron_dir).map_err(|_| AeronCError::from_code(-1))?;\n // Shared between init and cleanup so the mapping populated by\n // aeron_cnc_map_file_and_load_metadata stays live for aeron_unmap on drop.\n let mapped_file = RcOrArc::new(RefCellOrMutex::new(aeron_mapped_file_t {\n addr: std::ptr::null_mut(),\n length: 0,\n }));\n let mapped_file_for_cleanup = RcOrArc::clone(&mapped_file);\n let resource = ManagedCResource::new(\n move |ctx| {\n #[cfg(not(feature = \"multi-threaded\"))]\n let mut g = mapped_file.borrow_mut();\n #[cfg(feature = \"multi-threaded\")]\n let mut g = mapped_file.lock().unwrap();\n let result = unsafe {\n aeron_cnc_map_file_and_load_metadata(\n aeron_dir.as_ptr(),\n &mut *g as *mut aeron_mapped_file_t,\n ctx,\n )\n };\n if result == aeron_cnc_load_result_t::AERON_CNC_LOAD_SUCCESS {\n 1\n } else {\n -1\n }\n },\n Some(Box::new(move |_ctx| {\n #[cfg(not(feature = \"multi-threaded\"))]\n let mut g = mapped_file_for_cleanup.borrow_mut();\n #[cfg(feature = \"multi-threaded\")]\n let mut g = mapped_file_for_cleanup.lock().unwrap();\n unsafe { aeron_unmap(&mut *g as *mut aeron_mapped_file_t) };\n 0\n })),\n false,\n )?;\n\n let result = Self {\n inner: CResource::OwnedOnHeap(RcOrArc::new(resource)),\n };\n Ok(result)\n }\n\n #[inline]\n /// allocates on stack\n pub fn read_from_file(aeron_dir: &std::ffi::CString, mut handler: impl FnMut(Self)) -> Result<(), AeronCError> {\n let mut mapped_file = aeron_mapped_file_t {\n addr: std::ptr::null_mut(),\n length: 0,\n };\n let ctx = ManagedCResource::initialise(move |ctx| {\n let result = unsafe {\n aeron_cnc_map_file_and_load_metadata(\n aeron_dir.as_ptr(),\n &mut mapped_file as *mut aeron_mapped_file_t,\n ctx,\n )\n };\n if result == aeron_cnc_load_result_t::AERON_CNC_LOAD_SUCCESS {\n 1\n } else {\n -1\n }\n })?;\n\n let result = Self {\n inner: CResource::Borrowed(ctx),\n };\n\n handler(result);\n unsafe { aeron_unmap(&mut mapped_file as *mut aeron_mapped_file_t) };\n Ok(())\n }\n}\n\nunsafe extern \"C\" fn rusteron_image_visitor<F: FnMut(&AeronImage)>(\n image: *mut aeron_image_t,\n clientd: *mut ::std::os::raw::c_void,\n) {\n let f = &mut *(clientd as *mut F);\n let image = AeronImage {\n inner: CResource::Borrowed(image),\n };\n f(&image);\n}\n\nimpl AeronSubscription {\n /// A retained image handle for `index`, or `None` when no such image exists.\n ///\n /// The C client retains the image on this call; the returned [`AeronImage`] releases it\n /// automatically when the last clone drops (no manual `aeron_image_release`). If the\n /// subscription is closed first, the release is skipped \u{2014} the C client has already\n /// reclaimed the image.\n pub fn image_at_index(&self, index: usize) -> Option<AeronImage> {\n let image = unsafe { aeron_subscription_image_at_index(self.get_inner(), index) };\n self.wrap_retained_image(image)\n }\n\n /// A retained image handle for `session_id`, or `None` when no such image exists.\n /// Same automatic-release semantics as [`Self::image_at_index`].\n pub fn image_by_session_id(&self, session_id: i32) -> Option<AeronImage> {\n let image = unsafe { aeron_subscription_image_by_session_id(self.get_inner(), session_id) };\n self.wrap_retained_image(image)\n }\n\n fn wrap_retained_image(&self, image: *mut aeron_image_t) -> Option<AeronImage> {\n if image.is_null() {\n return None;\n }\n let subscription = self.clone();\n let resource = ManagedCResource::new(\n move |ctx| {\n unsafe { *ctx = image };\n 0\n },\n Some(Box::new(move |ctx| unsafe {\n // skip the release if the subscription was closed first \u{2014} the C client\n // reclaimed the image during the subscription close\n if !subscription.get_inner().is_null() {\n aeron_subscription_image_release(subscription.get_inner(), *ctx)\n } else {\n 0\n }\n })),\n false,\n )\n .ok()?;\n Some(AeronImage {\n inner: CResource::OwnedOnHeap(RcOrArc::new(resource)),\n })\n }\n\n /// Borrow-scoped iteration over the current images \u{2014} zero retain/release bookkeeping.\n /// The borrowed [`AeronImage`] is only valid inside the closure; call\n /// [`Self::image_at_index`] / [`Self::image_by_session_id`] for a handle that outlives it.\n pub fn for_each_image<F: FnMut(&AeronImage)>(&self, mut f: F) {\n unsafe {\n aeron_subscription_for_each_image(\n self.get_inner(),\n Some(rusteron_image_visitor::<F>),\n &mut f as *mut _ as *mut ::std::os::raw::c_void,\n )\n }\n }\n\n pub fn async_add_destination(\n &self,\n client: &Aeron,\n destination: &std::ffi::CStr,\n ) -> Result<AeronAsyncDestination, AeronCError> {\n AeronAsyncDestination::aeron_subscription_async_add_destination(client, self, destination)\n }\n\n /// Add `destination`, polling until the driver acknowledges or `timeout` elapses.\n /// The owning [`Aeron`] client is retrieved automatically from the subscription\'s\n /// dependency graph \u{2014} pass it explicitly via [`Self::async_add_destination`] only\n /// when you hold it already.\n pub fn add_destination(\n &self,\n destination: &std::ffi::CStr,\n timeout: std::time::Duration,\n ) -> Result<(), AeronCError> {\n let client = self\n .inner\n .get_dependency::<Aeron>()\n .ok_or_else(|| AeronCError::with_message(-1, \"subscription has no owning Aeron client\"))?;\n let result = self.async_add_destination(&client, destination)?;\n if result.aeron_subscription_async_destination_poll().unwrap_or_default() > 0 {\n return Ok(());\n }\n let time = std::time::Instant::now();\n while time.elapsed() < timeout {\n if result.aeron_subscription_async_destination_poll().unwrap_or_default() > 0 {\n return Ok(());\n }\n #[cfg(debug_assertions)]\n std::thread::sleep(std::time::Duration::from_millis(10));\n }\n log::error!(\"failed async poll for {:?} {:?}\", destination, self);\n Err(AeronErrorType::TimedOut.into())\n }\n\n}\n\nimpl AeronExclusivePublication {\n pub fn async_add_destination(\n &self,\n client: &Aeron,\n destination: &std::ffi::CStr,\n ) -> Result<AeronAsyncDestination, AeronCError> {\n AeronAsyncDestination::aeron_exclusive_publication_async_add_destination(client, self, destination)\n }\n\n /// Add `destination` (see [`AeronSubscription::add_destination`]); the owning\n /// [`Aeron`] client is retrieved from the publication\'s dependency graph.\n pub fn add_destination(\n &self,\n destination: &std::ffi::CStr,\n timeout: std::time::Duration,\n ) -> Result<(), AeronCError> {\n let client = self\n .inner\n .get_dependency::<Aeron>()\n .ok_or_else(|| AeronCError::with_message(-1, \"publication has no owning Aeron client\"))?;\n let result = self.async_add_destination(&client, destination)?;\n if result.aeron_subscription_async_destination_poll().unwrap_or_default() > 0 {\n return Ok(());\n }\n let time = std::time::Instant::now();\n while time.elapsed() < timeout {\n if result.aeron_subscription_async_destination_poll().unwrap_or_default() > 0 {\n return Ok(());\n }\n #[cfg(debug_assertions)]\n std::thread::sleep(std::time::Duration::from_millis(10));\n }\n log::error!(\"failed async poll for {:?} {:?}\", destination, self);\n Err(AeronErrorType::TimedOut.into())\n }\n}\n\nimpl AeronPublication {\n pub fn async_add_destination(\n &self,\n client: &Aeron,\n destination: &std::ffi::CStr,\n ) -> Result<AeronAsyncDestination, AeronCError> {\n AeronAsyncDestination::aeron_publication_async_add_destination(client, self, destination)\n }\n\n /// Add `destination` (see [`AeronSubscription::add_destination`]); the owning\n /// [`Aeron`] client is retrieved from the publication\'s dependency graph.\n pub fn add_destination(\n &self,\n destination: &std::ffi::CStr,\n timeout: std::time::Duration,\n ) -> Result<(), AeronCError> {\n let client = self\n .inner\n .get_dependency::<Aeron>()\n .ok_or_else(|| AeronCError::with_message(-1, \"publication has no owning Aeron client\"))?;\n let result = self.async_add_destination(&client, destination)?;\n if result.aeron_subscription_async_destination_poll().unwrap_or_default() > 0 {\n return Ok(());\n }\n let time = std::time::Instant::now();\n while time.elapsed() < timeout {\n if result.aeron_subscription_async_destination_poll().unwrap_or_default() > 0 {\n return Ok(());\n }\n #[cfg(debug_assertions)]\n std::thread::sleep(std::time::Duration::from_millis(10));\n }\n log::error!(\"failed async poll for {:?} {:?}\", destination, self);\n Err(AeronErrorType::TimedOut.into())\n }\n}\n\nimpl std::str::FromStr for AeronUriStringBuilder {\n type Err = AeronCError;\n\n fn from_str(s: &str) -> Result<Self, Self::Err> {\n let builder = AeronUriStringBuilder::default();\n let s = std::ffi::CString::new(s).map_err(|_| AeronCError::from_code(-1))?;\n builder.init_on_string(&s)?;\n Ok(builder)\n }\n}\n\n// AeronUriStringBuilder does not follow convention so manually adding Default method which calls close\nimpl Default for AeronUriStringBuilder {\n fn default() -> Self {\n let r_constructor = ManagedCResource::new(\n move |ctx_field| {\n let inst: aeron_uri_string_builder_t = unsafe { std::mem::zeroed() };\n let inner_ptr: *mut aeron_uri_string_builder_t = Box::into_raw(Box::new(inst));\n unsafe { *ctx_field = inner_ptr };\n 0\n },\n Some(Box::new(move |ctx_field| unsafe {\n aeron_uri_string_builder_close(*ctx_field)\n })),\n true,\n )\n .expect(\"should not happen\");\n Self {\n inner: CResource::OwnedOnHeap(RcOrArc::new(r_constructor)),\n }\n }\n}\n\nimpl AeronCError {\n pub fn get_last_err_message(&self) -> &str {\n self.message().unwrap_or_else(|| Aeron::errmsg())\n }\n\n /// Attach the current `aeron_errmsg()` text to this error (one allocation).\n /// Call at the error site when the error will be stored, logged later, or sent\n /// across threads \u{2014} otherwise `Display` reads the live buffer, which a later\n /// error can overwrite.\n pub fn capture_errmsg(mut self) -> Self {\n if self.msg.is_none() {\n self.msg = Some(Aeron::errmsg().into());\n }\n self\n }\n}\n\nimpl std::fmt::Debug for AeronCError {\n fn fmt(&self, f: &mut std::fmt::Formatter<\'_>) -> std::fmt::Result {\n f.debug_struct(\"AeronCError\")\n .field(\"code\", &self.code)\n .field(\"kind\", &self.kind())\n .field(\"lastError\", &self.get_last_err_message())\n .finish()\n }\n}\n\nimpl std::fmt::Display for AeronCError {\n fn fmt(&self, f: &mut std::fmt::Formatter<\'_>) -> std::fmt::Result {\n write!(\n f,\n \"Aeron error {}: {:?} [lastError={}]\",\n self.code,\n self.kind(),\n self.get_last_err_message()\n )\n }\n}\n\nimpl std::error::Error for AeronCError {}\n\nconst PARSE_CSTR_ERROR_CODE: i32 = -132131;\n\nimpl AeronUriStringBuilder {\n /// Fresh builder for an `aeron:ipc` channel.\n pub fn ipc() -> Result<Self, AeronCError> {\n let builder = AeronUriStringBuilder::new_zeroed_on_heap();\n builder.init_new()?;\n builder.media(Media::Ipc)?;\n Ok(builder)\n }\n\n /// Fresh builder for an `aeron:udp` channel with the given `endpoint` (`host:port`).\n pub fn udp(endpoint: &str) -> Result<Self, AeronCError> {\n let builder = AeronUriStringBuilder::new_zeroed_on_heap();\n builder.init_new()?;\n builder.media(Media::Udp)?.endpoint(endpoint)?;\n Ok(builder)\n }\n\n /// Fresh builder for an `aeron:udp` multi-destination channel with the given\n /// `control` endpoint and [`ControlMode`] (`Dynamic` for MDC publications,\n /// `Manual` for MDS subscriptions/publications, `Response` for response channels).\n pub fn udp_control(control: &str, mode: ControlMode) -> Result<Self, AeronCError> {\n let builder = AeronUriStringBuilder::new_zeroed_on_heap();\n builder.init_new()?;\n builder.media(Media::Udp)?.control(control)?.control_mode(mode)?;\n Ok(builder)\n }\n\n /// Close previous builder state and run a re-init function.\n ///\n /// Closes the previous C builder state directly (reserving the cleanup\n /// closure for the final Drop), runs `f`, and re-arms the cleanup gate on\n /// success so ManagedCResource::Drop calls `aeron_uri_string_builder_close`\n /// on the new state.\n #[inline]\n fn reinit_run<F>(&self, log_msg: &str, f: F) -> Result<i32, AeronCError>\n where\n F: FnOnce(*mut aeron_uri_string_builder_t) -> i32,\n {\n if let Some(inner) = self.inner.as_owned() {\n #[cfg(feature = \"multi-threaded\")]\n if !inner.close_already_called.load(std::sync::atomic::Ordering::SeqCst) {\n unsafe {\n aeron_uri_string_builder_close(inner.get());\n }\n inner.close_already_called.store(true, std::sync::atomic::Ordering::SeqCst);\n }\n #[cfg(not(feature = \"multi-threaded\"))]\n if !inner.close_already_called.get() {\n unsafe {\n aeron_uri_string_builder_close(inner.get());\n }\n inner.close_already_called.set(true);\n }\n }\n #[cfg(feature = \"log-c-bindings\")]\n log::info!(\"{}\", log_msg);\n let result = unsafe { f(self.get_inner()) };\n if result < 0 {\n Err(AeronCError::from_code(result))\n } else {\n if let Some(inner) = self.inner.as_owned() {\n #[cfg(feature = \"multi-threaded\")]\n inner.close_already_called.store(false, std::sync::atomic::Ordering::SeqCst);\n #[cfg(not(feature = \"multi-threaded\"))]\n inner.close_already_called.set(false);\n }\n Ok(result)\n }\n }\n\n #[inline]\n #[doc = \"Initialize a new AeronUriStringBuilder. If already initialized, it will close the previous builder to prevent memory leaks.\"]\n pub fn init_new(&self) -> Result<i32, AeronCError> {\n self.reinit_run(\"aeron_uri_string_builder_init_new(self.get_inner())\", |ptr| unsafe {\n aeron_uri_string_builder_init_new(ptr)\n })\n }\n\n #[inline]\n #[doc = \"Initialize AeronUriStringBuilder with an existing URI string. If already initialized, it will close the previous builder.\"]\n pub fn init_on_string(&self, uri: &std::ffi::CStr) -> Result<i32, AeronCError> {\n self.reinit_run(\n \"aeron_uri_string_builder_init_on_string(self.get_inner(), uri)\",\n |ptr| unsafe { aeron_uri_string_builder_init_on_string(ptr, uri.as_ptr()) },\n )\n }\n\n #[inline]\n pub fn build(&self, max_str_length: usize) -> Result<String, AeronCError> {\n let mut result = String::with_capacity(max_str_length);\n self.build_into(&mut result)?;\n Ok(result)\n }\n\n pub fn put_str(&self, key: &std::ffi::CStr, value: &str) -> Result<&Self, AeronCError> {\n let value = std::ffi::CString::new(value).map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put(&key, &value)?;\n Ok(self)\n }\n\n pub fn media(&self, value: Media) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_STRING_BUILDER_MEDIA_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value.as_str())?;\n Ok(self)\n }\n\n pub fn control_mode(&self, value: ControlMode) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_UDP_CHANNEL_CONTROL_MODE_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value.as_str())?;\n Ok(self)\n }\n\n pub fn prefix(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_STRING_BUILDER_PREFIX_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n\n pub fn initial_term_id(&self, value: i32) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_INITIAL_TERM_ID_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int32(key, value)?;\n Ok(self)\n }\n pub fn term_id(&self, value: i32) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_TERM_ID_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int32(key, value)?;\n Ok(self)\n }\n pub fn term_offset(&self, value: i32) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_TERM_OFFSET_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int32(key, value)?;\n Ok(self)\n }\n pub fn alias(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_ALIAS_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n pub fn term_length(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_TERM_LENGTH_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n pub fn linger_timeout(&self, value: i64) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_LINGER_TIMEOUT_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int64(key, value)?;\n Ok(self)\n }\n pub fn mtu_length(&self, value: i32) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_MTU_LENGTH_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int32(key, value)?;\n Ok(self)\n }\n pub fn ttl(&self, value: i32) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_UDP_CHANNEL_TTL_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int32(key, value)?;\n Ok(self)\n }\n pub fn sparse_term(&self, value: bool) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_SPARSE_TERM_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, if value { \"true\" } else { \"false\" })?;\n Ok(self)\n }\n pub fn reliable(&self, value: bool) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_UDP_CHANNEL_RELIABLE_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, if value { \"true\" } else { \"false\" })?;\n Ok(self)\n }\n pub fn eos(&self, value: bool) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_EOS_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, if value { \"true\" } else { \"false\" })?;\n Ok(self)\n }\n pub fn tether(&self, value: bool) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_TETHER_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, if value { \"true\" } else { \"false\" })?;\n Ok(self)\n }\n pub fn tags(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_TAGS_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n pub fn endpoint(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_UDP_CHANNEL_ENDPOINT_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n pub fn interface(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_UDP_CHANNEL_INTERFACE_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n pub fn control(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_UDP_CHANNEL_CONTROL_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n pub fn session_id(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_SESSION_ID_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n pub fn group(&self, value: bool) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_GROUP_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, if value { \"true\" } else { \"false\" })?;\n Ok(self)\n }\n pub fn rejoin(&self, value: bool) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_REJOIN_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, if value { \"true\" } else { \"false\" })?;\n Ok(self)\n }\n pub fn fc(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_FC_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n pub fn gtag(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_GTAG_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n pub fn cc(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_CC_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n pub fn spies_simulate_connection(&self, value: bool) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_SPIES_SIMULATE_CONNECTION_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, if value { \"true\" } else { \"false\" })?;\n Ok(self)\n }\n pub fn ats(&self, value: bool) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_ATS_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, if value { \"true\" } else { \"false\" })?;\n Ok(self)\n }\n pub fn socket_sndbuf(&self, value: i32) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_SOCKET_SNDBUF_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int32(key, value)?;\n Ok(self)\n }\n pub fn socket_rcvbuf(&self, value: i32) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_SOCKET_RCVBUF_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int32(key, value)?;\n Ok(self)\n }\n pub fn receiver_window(&self, value: i32) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_RECEIVER_WINDOW_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int32(key, value)?;\n Ok(self)\n }\n pub fn media_rcv_timestamp_offset(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_MEDIA_RCV_TIMESTAMP_OFFSET_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n pub fn channel_rcv_timestamp_offset(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_CHANNEL_RCV_TIMESTAMP_OFFSET_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n pub fn channel_snd_timestamp_offset(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_CHANNEL_SND_TIMESTAMP_OFFSET_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n pub fn timestamp_offset_reserved(&self, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_TIMESTAMP_OFFSET_RESERVED)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_str(key, value)?;\n Ok(self)\n }\n pub fn response_correlation_id(&self, value: i64) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_RESPONSE_CORRELATION_ID_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int64(key, value)?;\n Ok(self)\n }\n pub fn nak_delay(&self, value: i64) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_NAK_DELAY_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int64(key, value)?;\n Ok(self)\n }\n pub fn untethered_window_limit_timeout(&self, value: i64) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_UNTETHERED_WINDOW_LIMIT_TIMEOUT_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int64(key, value)?;\n Ok(self)\n }\n pub fn untethered_resting_timeout(&self, value: i64) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_UNTETHERED_RESTING_TIMEOUT_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int64(key, value)?;\n Ok(self)\n }\n pub fn max_resend(&self, value: i32) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_MAX_RESEND_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int32(key, value)?;\n Ok(self)\n }\n pub fn stream_id(&self, value: i32) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_STREAM_ID_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int32(key, value)?;\n Ok(self)\n }\n pub fn publication_window(&self, value: i32) -> Result<&Self, AeronCError> {\n let key = std::ffi::CStr::from_bytes_until_nul(AERON_URI_PUBLICATION_WINDOW_KEY)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put_int32(key, value)?;\n Ok(self)\n }\n\n #[inline]\n pub fn build_into(&self, dst: &mut String) -> Result<(), AeronCError> {\n self.sprint_into(dst)?;\n Ok(())\n }\n\n /// Remove a key from the URI builder by setting it to null.\n ///\n /// # Example\n /// ```ignore\n /// let builder = AeronUriStringBuilder::default();\n /// builder.remove(std::ffi::CStr::from_bytes_until_nul(b\"tags\\0\").unwrap())?;\n /// ```\n #[inline]\n pub fn remove(&self, key: &std::ffi::CStr) -> Result<i32, AeronCError> {\n unsafe {\n let result = aeron_uri_string_builder_put(self.get_inner(), key.as_ptr(), std::ptr::null());\n if result < 0 {\n Err(AeronCError::from_code(result))\n } else {\n Ok(result)\n }\n }\n }\n\n /// Remove a key from the URI builder by string slice.\n ///\n /// # Example\n /// ```ignore\n /// let builder = AeronUriStringBuilder::default();\n /// builder.remove_str(\"tags\")?;\n /// ```\n #[inline]\n pub fn remove_str(&self, key: &str) -> Result<i32, AeronCError> {\n let key = std::ffi::CString::new(key).map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.remove(&key)\n }\n}\n\nimpl AeronCountersReader {\n /// Find the first counter of `type_id` whose key matches `predicate`.\n ///\n /// The type ids are the bindgen\'d `AERON_COUNTER_*_TYPE_ID` constants; the key layout\n /// is type-specific (see `aeron_counters.h`). Mirrors Java\'s `CountersReader` lookups.\n pub fn find_by_type_id(&self, type_id: i32, mut predicate: impl FnMut(&[u8]) -> bool) -> Option<i32> {\n let mut found = None;\n self.foreach_counter_fn(|_value, id, counter_type_id, key, _label| {\n if found.is_none() && counter_type_id == type_id && predicate(key) {\n found = Some(id);\n }\n });\n found\n }\n\n /// Find the first counter of `type_id` keyed by `registration_id` \u{2014} the layout used by\n /// aeron\'s per-stream counters (publisher limit/position, sender/receiver positions,\n /// subscriber position: the key starts with the registration id as a little-endian\n /// `i64`). Mirrors Java\'s `findCounterIdByRegistration`-style helpers, e.g.:\n ///\n /// ```no_compile\n /// let registration_id = publication.get_constants()?.registration_id;\n /// let limit_counter = counters_reader\n /// .find_by_type_and_registration_id(AERON_COUNTER_PUBLISHER_LIMIT_TYPE_ID as i32, registration_id);\n /// ```\n pub fn find_by_type_and_registration_id(&self, type_id: i32, registration_id: i64) -> Option<i32> {\n self.find_by_type_id(type_id, |key| {\n key.len() >= 8 && i64::from_le_bytes(key[..8].try_into().unwrap()) == registration_id\n })\n }\n\n #[inline]\n #[doc = \"Get the label for a counter.\"]\n #[doc = \"\"]\n #[doc = \" \\n**param** counters_reader that contains the counter\"]\n #[doc = \" \\n**param** counter_id to find\"]\n #[doc = \" \\n**param** buffer to store the counter in.\"]\n #[doc = \" \\n**param** buffer_length length of the output buffer\"]\n #[doc = \" \\n**return** -1 on failure, number of characters copied to buffer on success.\"]\n pub fn get_counter_label(&self, counter_id: i32, max_length: usize) -> Result<String, AeronCError> {\n let mut result = String::with_capacity(max_length);\n self.get_counter_label_into(counter_id, &mut result)?;\n Ok(result)\n }\n\n #[inline]\n #[doc = \"Get the label for a counter.\"]\n pub fn get_counter_label_into(&self, counter_id: i32, dst: &mut String) -> Result<(), AeronCError> {\n unsafe {\n let capacity = dst.capacity();\n let vec = dst.as_mut_vec();\n vec.set_len(capacity);\n let written = self.counter_label(counter_id, &mut vec[..])? as usize;\n vec.set_len(std::cmp::min(written, capacity));\n }\n Ok(())\n }\n\n #[inline]\n #[doc = \"Get the key for a counter.\"]\n pub fn get_counter_key(&self, counter_id: i32) -> Result<Vec<u8>, AeronCError> {\n let mut dst = Vec::new();\n self.get_counter_key_into(counter_id, &mut dst)?;\n Ok(dst)\n }\n\n #[inline]\n #[doc = \"Get the key for a counter.\"]\n pub fn get_counter_key_into(&self, counter_id: i32, dst: &mut Vec<u8>) -> Result<(), AeronCError> {\n let mut key_ptr: *mut u8 = std::ptr::null_mut();\n unsafe {\n let result = bindings::aeron_counters_reader_metadata_key(self.get_inner(), counter_id, &mut key_ptr);\n if result < 0 || key_ptr.is_null() {\n return Err(AeronCError::from_code(result));\n }\n\n loop {\n let val = *key_ptr.add(dst.len());\n if val == 0 {\n break;\n } else {\n dst.push(val);\n }\n }\n Ok(())\n }\n }\n\n #[inline]\n pub fn get_counter_value(&self, counter_id: i32) -> i64 {\n unsafe { *self.addr(counter_id) }\n }\n}\n\nimpl Aeron {\n pub fn new_blocking(context: &AeronContext, timeout: std::time::Duration) -> Result<Self, AeronCError> {\n if let Ok(aeron) = Aeron::new(&context) {\n return Ok(aeron);\n }\n let time = std::time::Instant::now();\n while time.elapsed() < timeout {\n if let Ok(aeron) = Aeron::new(&context) {\n return Ok(aeron);\n }\n #[cfg(debug_assertions)]\n std::thread::sleep(std::time::Duration::from_millis(10));\n }\n log::error!(\"failed to create aeron client for {:?}\", context);\n Err(AeronErrorType::TimedOut.into())\n }\n}\n\nimpl AeronFragmentHandlerCallback for AeronFragmentAssembler {\n fn handle_aeron_fragment_handler(&mut self, buffer: &[u8], header: AeronHeader) -> () {\n unsafe {\n aeron_fragment_assembler_handler(\n self.get_inner() as *mut _,\n buffer.as_ptr(),\n buffer.len(),\n header.get_inner(),\n )\n }\n }\n}\n\nimpl AeronControlledFragmentHandlerCallback for AeronControlledFragmentAssembler {\n fn handle_aeron_controlled_fragment_handler(\n &mut self,\n buffer: &[u8],\n header: AeronHeader,\n ) -> aeron_controlled_fragment_handler_action_t {\n unsafe {\n aeron_controlled_fragment_assembler_handler(\n self.get_inner() as *mut _,\n buffer.as_ptr(),\n buffer.len(),\n header.get_inner(),\n )\n }\n }\n}\n\nimpl<T: AeronFragmentHandlerCallback + \'static> Handler<T> {\n /// Wrap `handler` in a fragment assembler; both are reference-counted and freed\n /// automatically when the last clones drop (the assembler keeps the delegate alive).\n pub fn with_fragment_assembler(handler: T) -> Result<(Handler<AeronFragmentAssembler>, Handler<T>), AeronCError> {\n let handler = Handler::new(handler);\n Ok((Handler::new(AeronFragmentAssembler::new(Some(&handler))?), handler))\n }\n\n #[deprecated(note = \"use Handler::with_fragment_assembler\")]\n pub fn leak_with_fragment_assembler(\n handler: T,\n ) -> Result<(Handler<AeronFragmentAssembler>, Handler<T>), AeronCError> {\n Self::with_fragment_assembler(handler)\n }\n}\n\nimpl<T: AeronControlledFragmentHandlerCallback + \'static> Handler<T> {\n /// Wrap `handler` in a controlled fragment assembler; both are reference-counted and\n /// freed automatically when the last clones drop.\n pub fn with_controlled_fragment_assembler(\n handler: T,\n ) -> Result<(Handler<AeronControlledFragmentAssembler>, Handler<T>), AeronCError> {\n let handler = Handler::new(handler);\n Ok((\n Handler::new(AeronControlledFragmentAssembler::new(Some(&handler))?),\n handler,\n ))\n }\n\n #[deprecated(note = \"use Handler::with_controlled_fragment_assembler\")]\n pub fn leak_with_controlled_fragment_assembler(\n handler: T,\n ) -> Result<(Handler<AeronControlledFragmentAssembler>, Handler<T>), AeronCError> {\n Self::with_controlled_fragment_assembler(handler)\n }\n}\n\nimpl AeronBufferClaim {\n /// Writable view over the claimed term-buffer region.\n ///\n /// # Safety contract ( upheld by construction )\n /// Only sound on a **genuinely claimed** buffer (produced by a successful\n /// `try_claim`). `length` and `data` come straight from the Aeron C claim\n /// and are trusted; a default-constructed/null claim would be unsound to\n /// dereference \u{2014} the `debug_assert` guards that in debug builds only (no\n /// release-build cost on this hot path).\n #[inline]\n pub fn data_mut(&self) -> &mut [u8] {\n debug_assert!(!self.data.is_null());\n unsafe { std::slice::from_raw_parts_mut(self.data, self.length) }\n }\n\n #[inline]\n pub fn frame_header_mut(&self) -> &mut aeron_header_values_frame_t {\n debug_assert!(!self.frame_header.is_null());\n unsafe { &mut *self.frame_header.cast::<aeron_header_values_frame_t>() }\n }\n}\n\n/// Zero-copy claim on a publication\'s term buffer with a RAII commit-or-abort\n/// lifecycle.\n///\n/// A `AeronClaim` that is dropped without being explicitly committed or aborted\n/// is **aborted** in [`Drop`], releasing the term-buffer slot immediately\n/// instead of waiting for `AERON_PUBLICATION_UNBLOCK_TIMEOUT_NS` (default 15s).\n/// [`AeronClaim::commit`] / [`AeronClaim::abort`] consume `self`, encoding\n/// Aeron\'s one-shot claim contract in the type system.\n///\n/// Construct via [`AeronPublication::try_claim_owned`] or\n/// [`AeronExclusivePublication::try_claim_owned`]; the claim is returned only on\n/// a successful `try_claim`, so its inner buffer is always a genuinely claimed\n/// slot (never a null zero-default).\npub struct AeronClaim {\n claim: AeronBufferClaim,\n position: i64,\n finalised: bool,\n}\n\nimpl AeronClaim {\n /// The writable claimed slice (zero-copy into the publication term buffer).\n #[inline]\n pub fn data(&mut self) -> &mut [u8] {\n self.claim.data()\n }\n\n /// Length of the claimed slice in bytes.\n #[inline]\n pub fn len(&self) -> usize {\n self.claim.length()\n }\n\n /// Whether the claimed slice is empty.\n #[inline]\n pub fn is_empty(&self) -> bool {\n self.claim.length() == 0\n }\n\n /// Stream position Aeron assigned to this claim.\n #[inline]\n pub fn position(&self) -> i64 {\n self.position\n }\n\n /// Commit the claimed bytes, publishing them to subscribers. Consumes `self`.\n pub fn commit(mut self) -> Result<i64, AeronCError> {\n self.finalised = true;\n self.claim.commit()?;\n Ok(self.position)\n }\n\n /// Abort the claim, discarding the slot as padding for subscribers.\n /// Consumes `self`.\n pub fn abort(mut self) -> Result<(), AeronCError> {\n self.finalised = true;\n self.claim.abort()?;\n Ok(())\n }\n}\n\nimpl Drop for AeronClaim {\n fn drop(&mut self) {\n if !self.finalised {\n // Defensive backstop: abort so the term slot is released now rather\n // than after the publication unblock timeout. The claim is always a\n // genuine claimed slot (constructed only on a successful try_claim),\n // so calling abort here is safe.\n let _ = self.claim.abort();\n }\n }\n}\n\nmacro_rules! impl_publication_methods {\n ($ty:ty, $offer:ident, $offerv:ident, $try_claim:ident) => {\n impl $ty {\n /// Raw, branch-free variant of [`Self::offer`]: returns the new stream position, or a\n /// negative Aeron sentinel (see [`AeronOfferError::from_position`]).\n ///\n /// Distinct name from the generated `offer` method (different signatures)\n /// to avoid macro-hygiene name conflicts.\n #[inline]\n pub fn offer_raw<H: AeronReservedValueSupplierCallback>(\n &self,\n buffer: &[u8],\n reserved_value_supplier: Option<&Handler<H>>,\n ) -> i64 {\n unsafe {\n $offer(\n self.get_inner(),\n buffer.as_ptr() as *mut _,\n buffer.len(),\n {\n let callback: aeron_reserved_value_supplier_t = if reserved_value_supplier.is_none() {\n None\n } else {\n Some(aeron_reserved_value_supplier_t_callback::<H>)\n };\n callback\n },\n reserved_value_supplier\n .map(|m| m.as_raw())\n .unwrap_or_else(|| std::ptr::null_mut()),\n )\n }\n }\n\n /// Raw, branch-free variant of [`Self::try_claim`]: position or negative sentinel.\n #[inline]\n pub fn try_claim_raw(&self, length: usize, buffer_claim: &AeronBufferClaim) -> i64 {\n unsafe { $try_claim(self.get_inner(), length, buffer_claim.get_inner()) }\n }\n\n /// [`Self::offer`] with a reserved-value supplier.\n #[inline]\n pub fn offer_with_reserved_value<H: AeronReservedValueSupplierCallback>(\n &self,\n buffer: &[u8],\n reserved_value_supplier: Option<&Handler<H>>,\n ) -> Result<i64, AeronOfferError> {\n AeronOfferError::from_position(self.offer_raw(buffer, reserved_value_supplier))\n }\n\n /// Zero-copy claim returning a RAII [`AeronClaim`] that is auto-aborted on\n /// drop unless explicitly committed or aborted.\n pub fn try_claim_owned(&self, length: usize) -> Result<AeronClaim, AeronOfferError> {\n let claim = AeronBufferClaim::new_zeroed_on_stack();\n let position = self.try_claim(length, &claim)?;\n Ok(AeronClaim {\n claim,\n position,\n finalised: false,\n })\n }\n\n /// Gathering (vectored) publish: offer up to [`MAX_OFFER_PARTS`] buffers as ONE\n /// message without concatenating them \u{2014} **zero allocation, zero copy** on the\n /// caller side (the driver gathers the parts directly).\n ///\n /// This is the header+payload send: instead of building a `Vec` per message\n /// (`vec.extend(header); vec.extend(payload); offer(&vec)`), pass the parts:\n ///\n /// ```ignore\n /// publication.offer_parts(&[&header_bytes, payload])?;\n /// ```\n ///\n /// The `aeron_iovec_t` array is built on the stack. Same typed-error semantics\n /// as [`Self::offer`]. Returns [`AeronOfferError::TooManyParts`] if more than\n /// [`MAX_OFFER_PARTS`] parts are passed (use [`Self::offerv`] with your own\n /// iovec array for larger gathers).\n #[inline]\n pub fn offer_parts(&self, parts: &[&[u8]]) -> Result<i64, AeronOfferError> {\n if parts.len() > MAX_OFFER_PARTS {\n return Err(AeronOfferError::TooManyParts);\n }\n let mut iov = [aeron_iovec_t {\n iov_base: std::ptr::null_mut(),\n iov_len: 0,\n }; MAX_OFFER_PARTS];\n for (slot, part) in iov.iter_mut().zip(parts) {\n slot.iov_base = part.as_ptr() as *mut u8;\n slot.iov_len = part.len();\n }\n let position = unsafe {\n $offerv(\n self.get_inner(),\n iov.as_mut_ptr(),\n parts.len(),\n None,\n std::ptr::null_mut(),\n )\n };\n AeronOfferError::from_position(position)\n }\n\n /// High-level connection state derived from the publication handle.\n #[inline]\n pub fn status(&self) -> AeronStatus {\n if self.is_closed() {\n AeronStatus::Closed\n } else if self.is_connected() {\n AeronStatus::Connected\n } else {\n AeronStatus::Disconnected\n }\n }\n\n /// Convenience accessor for the registration id (no direct C accessor exists;\n /// backed by [`Self::get_constants`]). `session_id`, `stream_id` and `channel`\n /// have cheap direct getters \u{2014} prefer those; for `registration_id` /\n /// `max_payload_length` in a hot loop, call [`Self::get_constants`] once and\n /// reuse the returned [`AeronPublicationConstants`].\n #[inline]\n pub fn registration_id(&self) -> Result<i64, AeronCError> {\n self.get_constants().map(|c| c.registration_id())\n }\n\n /// Convenience accessor for the max payload length (see [`Self::registration_id`]).\n #[inline]\n pub fn max_payload_length(&self) -> Result<usize, AeronCError> {\n self.get_constants().map(|c| c.max_payload_length())\n }\n }\n };\n}\n\nimpl_publication_methods!(AeronPublication, aeron_publication_offer, aeron_publication_offerv, aeron_publication_try_claim);\nimpl_publication_methods!(AeronExclusivePublication, aeron_exclusive_publication_offer, aeron_exclusive_publication_offerv, aeron_exclusive_publication_try_claim);\n\nimpl AeronPublication {\n /// Non-blocking publish of `buffer`, returning the new stream position.\n ///\n /// Typed-error convenience wrapper over [`Self::offer_raw`]; see\n /// [`AeronOfferError::is_retryable`] for retry-loop guidance.\n #[inline]\n pub fn offer(&self, buffer: &[u8]) -> Result<i64, AeronOfferError> {\n AeronOfferError::from_position(self.offer_raw::<AeronReservedValueSupplierLogger>(buffer, None))\n }\n\n /// Zero-copy claim with typed errors \u{2014} see [`Self::try_claim_owned`] for RAII.\n #[inline]\n pub fn try_claim(&self, length: usize, buffer_claim: &AeronBufferClaim) -> Result<i64, AeronOfferError> {\n AeronOfferError::from_position(self.try_claim_raw(length, buffer_claim))\n }\n}\n\nimpl AeronExclusivePublication {\n #[inline]\n pub fn offer(&self, buffer: &[u8]) -> Result<i64, AeronOfferError> {\n AeronOfferError::from_position(self.offer_raw::<AeronReservedValueSupplierLogger>(buffer, None))\n }\n\n #[inline]\n pub fn try_claim(&self, length: usize, buffer_claim: &AeronBufferClaim) -> Result<i64, AeronOfferError> {\n AeronOfferError::from_position(self.try_claim_raw(length, buffer_claim))\n }\n}\n\nimpl AeronSubscription {\n /// High-level connection state derived from the subscription handle.\n #[inline]\n pub fn status(&self) -> AeronStatus {\n if self.is_closed() {\n AeronStatus::Closed\n } else if self.is_connected() {\n AeronStatus::Connected\n } else {\n AeronStatus::Disconnected\n }\n }\n\n /// Convenience accessor for the stream id (no direct C accessor exists for\n /// subscriptions; backed by [`Self::get_constants`]). For a hot loop, call\n /// [`Self::get_constants`] once and reuse the returned\n /// [`AeronSubscriptionConstants`].\n #[inline]\n pub fn stream_id(&self) -> Result<i32, AeronCError> {\n self.get_constants().map(|c| c.stream_id())\n }\n\n /// Convenience accessor for the channel (see [`Self::stream_id`]).\n #[inline]\n pub fn channel(&self) -> Result<String, AeronCError> {\n self.get_constants().map(|c| c.channel().to_string())\n }\n\n /// Convenience accessor for the registration id (see [`Self::stream_id`]).\n #[inline]\n pub fn registration_id(&self) -> Result<i64, AeronCError> {\n self.get_constants().map(|c| c.registration_id())\n }\n}\n\nimpl AeronImage {\n /// Instrumented wrapper around [`AeronImage::poll`] that adds tracing spans\n /// when the `instrument-ops` feature is enabled.\n #[inline]\n pub fn poll_instrumented<AeronFragmentHandlerHandlerImpl: AeronFragmentHandlerCallback>(\n &self,\n handler: Option<&Handler<AeronFragmentHandlerHandlerImpl>>,\n fragment_limit: usize,\n ) -> Result<i32, AeronCError> {\n self.poll(handler, fragment_limit)\n }\n}\n\n/// Production error handler that routes Aeron async errors through the Rust `log`\n/// facade at **error** level with a concise format \u{2014} the recommended default for\n/// `AeronContext::set_error_handler`.\n///\n/// Not to be confused with the generated [`AeronErrorHandlerLogger`], which is the\n/// codegen\'s generic callback tracer (info level, verbose per-call format, useful\n/// only behind `--features log-c-bindings`-style debugging).\npub struct AeronErrorLogger;\nimpl AeronErrorHandlerCallback for AeronErrorLogger {\n fn handle_aeron_error_handler(&mut self, error_code: std::ffi::c_int, msg: &str) -> () {\n log::error!(\"aeron error {}: {}\", error_code, msg);\n }\n}\nunsafe impl Send for AeronErrorLogger {}\nunsafe impl Sync for AeronErrorLogger {}\n\npub struct FnMutMessageHandler {\n func: fn(*mut (), &[u8], AeronHeader),\n ctx: *mut (),\n}\n\nimpl AeronFragmentHandlerCallback for FnMutMessageHandler {\n fn handle_aeron_fragment_handler(&mut self, buffer: &[u8], header: AeronHeader) -> () {\n self.call(buffer, header);\n }\n}\n\nimpl FnMutMessageHandler {\n pub fn new() -> Self {\n Self {\n func: Self::noop,\n ctx: std::ptr::null_mut(),\n }\n }\n\n #[inline]\n /// Point this handler at `ctx` / `func` so the next [`call`](Self::call)\n /// dispatches into them.\n ///\n /// # Lifetime contract (caller must uphold)\n /// `ctx` is stored as a raw `*mut ()` that **escapes the borrow** \u{2014} the\n /// borrow checker does NOT keep it alive. The caller MUST ensure `ctx`\n /// outlives every subsequent `call` (i.e. `ctx` is dropped only after this\n /// handler is retired or re-`set`). Calling `call` after `ctx` is dropped is\n /// use-after-free. `ctx` must also not be moved while borrowed.\n pub fn set<T>(&mut self, ctx: &mut T, func: fn(&mut T, &[u8], AeronHeader)) -> &mut Self {\n self.func = Self::wrap::<T>(func);\n self.ctx = ctx as *mut T as *mut ();\n self\n }\n\n #[inline(always)]\n pub fn call(&mut self, msg: &[u8], header: AeronHeader) {\n (self.func)(self.ctx, msg, header);\n }\n\n /// Drop the borrowed `ctx` so a later stray [`call`](Self::call) is a safe no-op\n /// (the `noop` func is restored and the pointer nulled) rather than a use-after-free.\n ///\n /// [`AeronFragmentClosureAssembler::poll`] calls this before returning so the\n /// raw pointer never outlives the borrow it was created from.\n #[inline]\n pub fn clear(&mut self) {\n self.func = Self::noop;\n self.ctx = std::ptr::null_mut();\n }\n\n #[inline]\n fn wrap<T>(f: fn(&mut T, &[u8], AeronHeader)) -> fn(*mut (), &[u8], AeronHeader) {\n // SAFETY: `fn(&mut T,\u{2026})` and `fn(*mut(),\u{2026})` have the same ABI/representation\n unsafe { std::mem::transmute(f) }\n }\n\n fn noop(_: *mut (), _: &[u8], _: AeronHeader) {\n // default no-op handler\n }\n}\n\n/// A poll target whose raw fragments can be reassembled by [`AeronFragmentAssembler`].\n/// Implemented for [`AeronSubscription`] here and for archive types in rusteron-archive.\npub trait FragmentAssemblable {\n fn poll_with_assembler(\n &self,\n assembler: Option<&Handler<AeronFragmentAssembler>>,\n fragment_limit: usize,\n ) -> Result<i32, AeronCError>;\n}\n\nimpl FragmentAssemblable for AeronSubscription {\n #[inline]\n fn poll_with_assembler(\n &self,\n assembler: Option<&Handler<AeronFragmentAssembler>>,\n fragment_limit: usize,\n ) -> Result<i32, AeronCError> {\n self.poll(assembler, fragment_limit)\n }\n}\n\npub struct AeronFragmentClosureAssembler {\n assembler: Handler<AeronFragmentAssembler>,\n handler: Handler<FnMutMessageHandler>,\n}\n\nimpl AeronFragmentClosureAssembler {\n pub fn new() -> Result<Self, AeronCError> {\n let handler = Handler::new(FnMutMessageHandler::new());\n Ok(Self {\n assembler: Handler::new(AeronFragmentAssembler::new(Some(&handler))?),\n handler,\n })\n }\n\n /// Poll `pollable` (a subscription or persistent subscription), dispatching each\n /// reassembled message to `func`. `ctx` is borrowed only for the duration of the call.\n ///\n /// Returns the fragment count from the underlying poll.\n pub fn poll<P: FragmentAssemblable, T>(\n &mut self,\n pollable: &P,\n ctx: &mut T,\n func: fn(&mut T, &[u8], AeronHeader),\n fragment_limit: usize,\n ) -> Result<i32, AeronCError> {\n unsafe {\n self.handler.get_mut().set(ctx, func);\n }\n let result = pollable.poll_with_assembler(Some(&self.assembler), fragment_limit);\n unsafe {\n self.handler.get_mut().clear();\n }\n result\n }\n}\n\npub struct FnMutControlledMessageHandler {\n func: fn(*mut (), &[u8], AeronHeader) -> aeron_controlled_fragment_handler_action_t,\n ctx: *mut (),\n}\n\nimpl FnMutControlledMessageHandler {\n pub fn new() -> Self {\n Self {\n func: Self::noop,\n ctx: std::ptr::null_mut(),\n }\n }\n\n #[inline]\n /// Point this handler at `ctx` / `func` so the next [`call`](Self::call)\n /// dispatches into them.\n ///\n /// # Lifetime contract (caller must uphold)\n /// Same as [`FnMutMessageHandler::set`](super::FnMutMessageHandler::set):\n /// `ctx` is stored as a raw pointer that escapes the borrow. The caller MUST\n /// keep `ctx` alive (and unmoved) until this handler is retired or re-`set`;\n /// calling `call` after `ctx` is dropped is use-after-free.\n pub fn set<T>(\n &mut self,\n ctx: &mut T,\n func: fn(&mut T, &[u8], AeronHeader) -> aeron_controlled_fragment_handler_action_t,\n ) -> &mut Self {\n self.func = Self::wrap::<T>(func);\n self.ctx = ctx as *mut T as *mut ();\n self\n }\n\n #[inline(always)]\n pub fn call(&mut self, msg: &[u8], header: AeronHeader) -> aeron_controlled_fragment_handler_action_t {\n (self.func)(self.ctx, msg, header)\n }\n\n /// Drop the borrowed `ctx` (cf. [`FnMutMessageHandler::clear`]) so a later stray\n /// [`call`](Self::call) returns `CONTINUE` instead of dereferencing a stale pointer.\n #[inline]\n pub fn clear(&mut self) {\n self.func = Self::noop;\n self.ctx = std::ptr::null_mut();\n }\n\n #[inline]\n fn wrap<T>(\n f: fn(&mut T, &[u8], AeronHeader) -> aeron_controlled_fragment_handler_action_t,\n ) -> fn(*mut (), &[u8], AeronHeader) -> aeron_controlled_fragment_handler_action_t {\n unsafe { std::mem::transmute(f) }\n }\n\n fn noop(_: *mut (), _: &[u8], _: AeronHeader) -> aeron_controlled_fragment_handler_action_t {\n bindings::aeron_controlled_fragment_handler_action_en::AERON_ACTION_CONTINUE\n }\n}\n\nimpl AeronControlledFragmentHandlerCallback for FnMutControlledMessageHandler {\n fn handle_aeron_controlled_fragment_handler(\n &mut self,\n buffer: &[u8],\n header: AeronHeader,\n ) -> aeron_controlled_fragment_handler_action_t {\n self.call(buffer, header)\n }\n}\n\n/// A poll target whose raw fragments can be reassembled by [`AeronControlledFragmentAssembler`].\npub trait ControlledFragmentAssemblable {\n fn controlled_poll_with_assembler(\n &self,\n assembler: Option<&Handler<AeronControlledFragmentAssembler>>,\n fragment_limit: usize,\n ) -> Result<i32, AeronCError>;\n}\n\nimpl ControlledFragmentAssemblable for AeronSubscription {\n #[inline]\n fn controlled_poll_with_assembler(\n &self,\n assembler: Option<&Handler<AeronControlledFragmentAssembler>>,\n fragment_limit: usize,\n ) -> Result<i32, AeronCError> {\n self.controlled_poll(assembler, fragment_limit)\n }\n}\n\npub struct AeronControlledFragmentClosureAssembler {\n assembler: Handler<AeronControlledFragmentAssembler>,\n handler: Handler<FnMutControlledMessageHandler>,\n}\n\nimpl AeronControlledFragmentClosureAssembler {\n pub fn new() -> Result<Self, AeronCError> {\n let handler = Handler::new(FnMutControlledMessageHandler::new());\n Ok(Self {\n assembler: Handler::new(AeronControlledFragmentAssembler::new(Some(&handler))?),\n handler,\n })\n }\n\n /// Controlled poll of `pollable`, dispatching each (possibly reassembled) message to\n /// `func`. `ctx` is borrowed only for the duration of the call.\n pub fn poll<P: ControlledFragmentAssemblable, T>(\n &mut self,\n pollable: &P,\n ctx: &mut T,\n func: fn(&mut T, &[u8], AeronHeader) -> aeron_controlled_fragment_handler_action_t,\n fragment_limit: usize,\n ) -> Result<i32, AeronCError> {\n unsafe {\n self.handler.get_mut().set(ctx, func);\n }\n let result = pollable.controlled_poll_with_assembler(Some(&self.assembler), fragment_limit);\n unsafe {\n self.handler.get_mut().clear();\n }\n result\n }\n}\n\n/// Status transition tracker that emits only when the observed [`AeronStatus`]\n/// differs from the previously recorded state.\n///\n/// Mirrors the reactive side-channel pattern in wingfoil\'s `AeronStatusStream`,\n/// useful for driving application-level state machines (e.g. reconnect logic,\n/// UI indicators) that need to react to connection changes without polling.\n///\n/// # Example\n/// ```ignore\n/// // (illustrative \u{2014} `publication.status()` needs a live handle; see the\n/// // `aeron_custom_tests` unit tests for runnable pure-Rust assertions)\n/// let mut tracker = AeronStatusTracker::new();\n/// tracker.observe(publication.status()); // emits `Some(Disconnected)`\n/// // ... connection establishes ...\n/// tracker.observe(publication.status()); // emits `Some(Connected)`\n/// tracker.observe(publication.status()); // emits `None` (no change)\n/// ```\n#[derive(Debug, Clone, Copy, PartialEq, Eq)]\npub struct AeronStatusTracker {\n last_status: Option<AeronStatus>,\n}\n\nimpl AeronStatusTracker {\n /// Create a new tracker with no initial state.\n #[must_use]\n pub const fn new() -> Self {\n Self { last_status: None }\n }\n\n /// Observe a status, returning `Some(status)` if this is a transition from\n /// the previous state (or the first observation), `None` if the status is\n /// unchanged.\n pub fn observe(&mut self, status: AeronStatus) -> Option<AeronStatus> {\n if self.last_status != Some(status) {\n self.last_status = Some(status);\n Some(status)\n } else {\n None\n }\n }\n\n /// Reset the tracker, causing the next call to [`observe`](Self::observe)\n /// to emit even if the status matches the previously-observed value.\n pub fn reset(&mut self) {\n self.last_status = None;\n }\n\n /// The most recent status observed, if any.\n #[must_use]\n pub const fn last_status(&self) -> Option<AeronStatus> {\n self.last_status\n }\n}\n\nimpl Default for AeronStatusTracker {\n fn default() -> Self {\n Self::new()\n }\n}\n\n/// Validate a UDP channel URI endpoint (host:port) for use with Aeron.\n///\n/// This validates the shape and character set of endpoint strings passed to\n/// [`AeronUriStringBuilder::endpoint`] or used directly in channel URIs like\n/// `aeron:udp?endpoint=localhost:40123`. Aeron does not validate input upfront;\n/// this function catches malformed endpoints before they reach the C layer,\n/// providing clearer error messages than the generic C parse failures.\n///\n/// # Accepted forms\n/// - `hostname:port` \u{2014} host is alphanumeric plus `-` and `.`, port 0-65535\n/// - `IPv4:port` \u{2014} dotted decimal, port 0-65535\n/// - `[IPv6]:port` \u{2014} bracketed IPv6 literal, port required\n///\n/// # Errors\n/// Returns `Err(AeronCError)` with code `-1` for any validation failure:\n/// - empty string or missing port separator\n/// - port not `0..=65535` or not a decimal integer\n/// - host contains characters outside the safe allowlist (alphanumeric,\n/// `-`, `.`, `:` for IPv4, `[]` for IPv6 bracketing)\n/// - URI separator characters (`?`, `=`, `:`, `/`) appear unescaped\n///\n/// # Example\n/// ```ignore\n/// // (illustrative \u{2014} `aeron_custom.rs` is `include!`\'d into multiple crates, so\n/// // no single `use` path compiles across all of them; see the\n/// // `aeron_custom_tests` module for runnable pure-Rust assertions)\n/// assert!(validate_endpoint_for_aeron_udp(\"localhost:40123\").is_ok());\n/// assert!(validate_endpoint_for_aeron_udp(\"[::1]:40123\").is_ok());\n/// assert!(validate_endpoint_for_aeron_udp(\"localhost:99999\").is_err());\n/// assert!(validate_endpoint_for_aeron_udp(\"localhost?foo:8080\").is_err());\n/// ```\npub fn validate_endpoint_for_aeron_udp(endpoint: &str) -> Result<(), AeronCError> {\n if endpoint.is_empty() {\n return Err(AeronCError::from_code(-1));\n }\n\n // Reject URI separator characters that would break channel string parsing.\n // These are safe in a full URI (e.g. `aeron:udp?endpoint=...`) but not in\n // the endpoint value itself.\n if endpoint.contains(\'?\') || endpoint.contains(\'=\') || endpoint.contains(\'/\') {\n return Err(AeronCError::from_code(-1));\n }\n\n // IPv6 addresses are bracketed: `[IPv6]:port`. Extract the host part.\n if endpoint.starts_with(\'[\') {\n let end_bracket = endpoint.find(\']\').ok_or_else(|| AeronCError::from_code(-1))?;\n if end_bracket == 1 {\n // Empty `[]` is invalid.\n return Err(AeronCError::from_code(-1));\n }\n let host_part = &endpoint[1..end_bracket];\n // Find the port separator colon AFTER the closing bracket.\n let after_bracket = &endpoint[end_bracket..];\n let colon_offset = after_bracket.find(\':\').ok_or_else(|| AeronCError::from_code(-1))?;\n let colon_pos = end_bracket + colon_offset;\n if colon_offset != 1 || colon_pos + 1 >= endpoint.len() {\n // Port must immediately follow `]` and be non-empty.\n return Err(AeronCError::from_code(-1));\n }\n let port_str = &endpoint[colon_pos + 1..];\n validate_ipv6(host_part)?;\n validate_port(port_str)?;\n return Ok(());\n }\n\n // Unbracketed form: `host:port`. Split on the last `:` (IPv4 has at most\n // 3 colons for dotted decimals; we only want the port separator).\n let colon_pos = endpoint.rfind(\':\').ok_or_else(|| AeronCError::from_code(-1))?;\n if colon_pos == 0 || colon_pos + 1 >= endpoint.len() {\n // Require non-empty host and port.\n return Err(AeronCError::from_code(-1));\n }\n let host = &endpoint[..colon_pos];\n let port_str = &endpoint[colon_pos + 1..];\n\n validate_host(host)?;\n validate_port(port_str)?;\n Ok(())\n}\n\n/// Validate an IPv6 address (without brackets). Returns `Err` if the string\n/// is not a valid IPv6 literal.\nfn validate_ipv6(addr: &str) -> Result<(), AeronCError> {\n if addr.is_empty() {\n return Err(AeronCError::from_code(-1));\n }\n\n // IPv6 allows: hexadecimal digits, `:` (single or `::` for compression).\n // Minimal validation: ensure only valid characters and at least one colon.\n let has_colon = addr.bytes().any(|b| b == b\':\');\n if !has_colon {\n return Err(AeronCError::from_code(-1));\n }\n\n for ch in addr.bytes() {\n let is_hex = ch.is_ascii_hexdigit();\n let is_colon = ch == b\':\';\n if !(is_hex || is_colon) {\n return Err(AeronCError::from_code(-1));\n }\n }\n\n Ok(())\n}\n\n/// Validate an Aeron UDP host identifier (hostname or IPv4). Returns `Err` if\n/// the host contains characters outside the safe allowlist, is empty, or looks\n/// like an unbracketed IPv6 address.\nfn validate_host(host: &str) -> Result<(), AeronCError> {\n if host.is_empty() {\n return Err(AeronCError::from_code(-1));\n }\n\n // Reject unbracketed IPv6 addresses (they must use `[IPv6]:port` form).\n // IPv6 addresses have multiple consecutive colons or colons in positions\n // that don\'t match IPv4 dotted decimal.\n let colon_count = host.bytes().filter(|&b| b == b\':\').count();\n if colon_count > 1 {\n // Multiple colons means this is likely an IPv6 address without brackets.\n return Err(AeronCError::from_code(-1));\n }\n if colon_count == 1 {\n // Single colon: could be IPv4 (ok) or a short IPv6 form like `::1` (reject).\n // If there\'s a colon adjacent to another colon or at the start, it\'s IPv6.\n if host.contains(\"::\") || host.starts_with(\':\') || host.ends_with(\':\') {\n return Err(AeronCError::from_code(-1));\n }\n }\n\n // Safe character set: alphanumeric, `-`, `.`, `:` (IPv4 has a single colon\n // between octets; IPv6 brackets are handled by the caller).\n for ch in host.bytes() {\n let is_alnum = ch.is_ascii_alphanumeric();\n let is_safe = matches!(ch, b\'-\' | b\'.\' | b\':\');\n if !(is_alnum || is_safe) {\n return Err(AeronCError::from_code(-1));\n }\n }\n\n Ok(())\n}\n\n/// Validate a TCP/UDP port number. Returns `Err` if the port is not a\n/// decimal integer in `0..=65535`.\nfn validate_port(port: &str) -> Result<(), AeronCError> {\n if port.is_empty() {\n return Err(AeronCError::from_code(-1));\n }\n\n // Port must be all digits (no leading `-` or `+`).\n if !port.bytes().all(|b| b.is_ascii_digit()) {\n return Err(AeronCError::from_code(-1));\n }\n\n // Parse as u16; reject values that overflow 65535.\n let port_num = port.parse::<u32>().map_err(|_| AeronCError::from_code(-1))?;\n if port_num > 65535 {\n return Err(AeronCError::from_code(-1));\n }\n\n Ok(())\n}\n\n#[cfg(test)]\nmod aeron_custom_tests {\n use super::*;\n\n #[test]\n fn from_position_positive_is_ok_position() {\n assert_eq!(AeronOfferError::from_position(0), Ok(0));\n assert_eq!(AeronOfferError::from_position(12_345), Ok(12_345));\n }\n\n #[test]\n fn from_position_maps_every_sentinel_distinctly() {\n assert_eq!(AeronOfferError::from_position(-1), Err(AeronOfferError::NotConnected));\n assert_eq!(AeronOfferError::from_position(-2), Err(AeronOfferError::BackPressured));\n assert_eq!(AeronOfferError::from_position(-3), Err(AeronOfferError::AdminAction));\n assert_eq!(AeronOfferError::from_position(-4), Err(AeronOfferError::Closed));\n assert_eq!(\n AeronOfferError::from_position(-5),\n Err(AeronOfferError::MaxPositionExceeded)\n );\n }\n\n #[test]\n fn retryable_vs_fatal_classification() {\n assert!(AeronOfferError::NotConnected.is_retryable());\n assert!(AeronOfferError::BackPressured.is_retryable());\n assert!(AeronOfferError::AdminAction.is_retryable());\n assert!(AeronOfferError::Closed.is_fatal());\n assert!(AeronOfferError::MaxPositionExceeded.is_fatal());\n assert!(AeronOfferError::Error(AeronCError::from_code(-99)).is_fatal());\n }\n\n #[test]\n fn from_position_unknown_negative_preserves_code() {\n match AeronOfferError::from_position(-99) {\n Err(AeronOfferError::Error(e)) => assert_eq!(e.code, -99),\n other => panic!(\"expected Error variant, got {other:?}\"),\n }\n }\n\n #[test]\n fn status_from_error_maps_back_pressure_and_closed() {\n assert_eq!(\n AeronStatus::from_error(&AeronOfferError::BackPressured),\n Some(AeronStatus::BackPressured)\n );\n assert_eq!(\n AeronStatus::from_error(&AeronOfferError::Closed),\n Some(AeronStatus::Closed)\n );\n }\n\n #[test]\n fn status_from_error_returns_none_for_non_status_errors() {\n // PublicationMaxPositionExceeded (-5) is not a status-like transition.\n assert_eq!(AeronStatus::from_error(&AeronOfferError::MaxPositionExceeded), None);\n }\n\n #[test]\n fn defused_claim_drops_without_invoking_ffi() {\n // A zero-default `AeronBufferClaim` has a null inner pointer; calling\n // abort/commit FFI on it would be unsound. With `finalised = true` the\n // `Drop` backstop must skip the abort. Constructing directly is allowed\n // because the test sits in the same module as the private fields. The\n // mere fact that this drops without segfaulting is the assertion\n // (mirrors wingfoil\'s defused-claim test). `position()` reads only the\n // stored i64 field, so it is also FFI-free.\n let claim = AeronClaim {\n claim: AeronBufferClaim::default(),\n position: 7,\n finalised: true,\n };\n assert_eq!(claim.position(), 7);\n drop(claim);\n }\n\n // --- AeronStatusTracker tests ---\n\n #[test]\n fn status_tracker_emits_on_first_observation() {\n let mut tracker = AeronStatusTracker::new();\n assert_eq!(\n tracker.observe(AeronStatus::Disconnected),\n Some(AeronStatus::Disconnected)\n );\n }\n\n #[test]\n fn status_tracker_emits_on_transition() {\n let mut tracker = AeronStatusTracker::new();\n tracker.observe(AeronStatus::Disconnected);\n assert_eq!(tracker.observe(AeronStatus::Connected), Some(AeronStatus::Connected));\n }\n\n #[test]\n fn status_tracker_does_not_emit_on_same_status() {\n let mut tracker = AeronStatusTracker::new();\n tracker.observe(AeronStatus::Connected);\n assert_eq!(tracker.observe(AeronStatus::Connected), None);\n assert_eq!(tracker.observe(AeronStatus::Connected), None);\n }\n\n #[test]\n fn status_tracker_emits_again_after_reset() {\n let mut tracker = AeronStatusTracker::new();\n tracker.observe(AeronStatus::Connected);\n assert_eq!(tracker.observe(AeronStatus::Connected), None);\n\n tracker.reset();\n assert_eq!(tracker.observe(AeronStatus::Connected), Some(AeronStatus::Connected));\n }\n\n #[test]\n fn status_tracker_records_closed_transition() {\n let mut tracker = AeronStatusTracker::new();\n tracker.observe(AeronStatus::Connected);\n assert_eq!(tracker.observe(AeronStatus::Closed), Some(AeronStatus::Closed));\n // No further transitions from Closed.\n assert_eq!(tracker.observe(AeronStatus::Closed), None);\n }\n\n #[test]\n fn status_tracker_tracks_all_states() {\n let mut tracker = AeronStatusTracker::new();\n\n // Full lifecycle: Disconnected -> Connected -> BackPressured -> Connected -> Closed\n assert_eq!(\n tracker.observe(AeronStatus::Disconnected),\n Some(AeronStatus::Disconnected)\n );\n assert_eq!(tracker.observe(AeronStatus::Connected), Some(AeronStatus::Connected));\n assert_eq!(\n tracker.observe(AeronStatus::BackPressured),\n Some(AeronStatus::BackPressured)\n );\n assert_eq!(tracker.observe(AeronStatus::Connected), Some(AeronStatus::Connected));\n assert_eq!(tracker.observe(AeronStatus::Closed), Some(AeronStatus::Closed));\n }\n\n // --- validate_endpoint_for_aeron_udp tests ---\n\n #[test]\n fn validate_endpoint_accepts_hostname_with_port() {\n assert!(validate_endpoint_for_aeron_udp(\"localhost:40123\").is_ok());\n assert!(validate_endpoint_for_aeron_udp(\"localhost:0\").is_ok());\n assert!(validate_endpoint_for_aeron_udp(\"my-host.example.com:65535\").is_ok());\n }\n\n #[test]\n fn validate_endpoint_accepts_ipv4_with_port() {\n assert!(validate_endpoint_for_aeron_udp(\"127.0.0.1:8080\").is_ok());\n assert!(validate_endpoint_for_aeron_udp(\"192.168.1.1:0\").is_ok());\n assert!(validate_endpoint_for_aeron_udp(\"10.0.0.1:65535\").is_ok());\n }\n\n #[test]\n fn validate_endpoint_accepts_ipv6_bracketed_with_port() {\n assert!(validate_endpoint_for_aeron_udp(\"[::1]:8080\").is_ok());\n assert!(validate_endpoint_for_aeron_udp(\"[fe80::1]:9000\").is_ok());\n assert!(validate_endpoint_for_aeron_udp(\"[2001:db8::1]:0\").is_ok());\n assert!(validate_endpoint_for_aeron_udp(\"[2001:db8::1]:65535\").is_ok());\n }\n\n #[test]\n fn validate_endpoint_rejects_empty() {\n assert!(validate_endpoint_for_aeron_udp(\"\").is_err());\n }\n\n #[test]\n fn validate_endpoint_rejects_missing_port() {\n assert!(validate_endpoint_for_aeron_udp(\"localhost\").is_err());\n assert!(validate_endpoint_for_aeron_udp(\"127.0.0.1\").is_err());\n assert!(validate_endpoint_for_aeron_udp(\"[::1]\").is_err());\n assert!(validate_endpoint_for_aeron_udp(\"[::1]:\").is_err());\n assert!(validate_endpoint_for_aeron_udp(\":8080\").is_err()); // empty host\n }\n\n #[test]\n fn validate_endpoint_rejects_port_out_of_range() {\n assert!(validate_endpoint_for_aeron_udp(\"localhost:65536\").is_err());\n assert!(validate_endpoint_for_aeron_udp(\"localhost:99999\").is_err());\n assert!(validate_endpoint_for_aeron_udp(\"[::1]:65536\").is_err());\n }\n\n #[test]\n fn validate_endpoint_rejects_non_digit_port() {\n assert!(validate_endpoint_for_aeron_udp(\"localhost:abc\").is_err());\n assert!(validate_endpoint_for_aeron_udp(\"localhost:80a0\").is_err());\n assert!(validate_endpoint_for_aeron_udp(\"localhost:-1\").is_err());\n assert!(validate_endpoint_for_aeron_udp(\"localhost:+8080\").is_err());\n }\n\n #[test]\n fn validate_endpoint_rejects_unsafe_characters() {\n // URI separator characters.\n assert!(validate_endpoint_for_aeron_udp(\"localhost?foo:8080\").is_err());\n assert!(validate_endpoint_for_aeron_udp(\"localhost=bar:8080\").is_err());\n assert!(validate_endpoint_for_aeron_udp(\"localhost/path:8080\").is_err());\n\n // Characters outside the safe allowlist.\n assert!(validate_endpoint_for_aeron_udp(\"local_host:8080\").is_err()); // underscore\n assert!(validate_endpoint_for_aeron_udp(\"local host:8080\").is_err()); // space\n assert!(validate_endpoint_for_aeron_udp(\"local$host:8080\").is_err()); // $\n }\n\n #[test]\n fn validate_endpoint_rejects_malformed_ipv6() {\n // Missing closing bracket.\n assert!(validate_endpoint_for_aeron_udp(\"[::1:8080\").is_err());\n\n // Empty brackets.\n assert!(validate_endpoint_for_aeron_udp(\"[]:8080\").is_err());\n\n // Port missing after bracket.\n assert!(validate_endpoint_for_aeron_udp(\"[::1]\").is_err());\n\n // Host after bracket (port must immediately follow `]`).\n assert!(validate_endpoint_for_aeron_udp(\"[::1]x:8080\").is_err());\n\n // Bracket without colon in the right place.\n assert!(validate_endpoint_for_aeron_udp(\"[::1]8080\").is_err());\n }\n\n #[test]\n fn validate_endpoint_rejects_colon_in_wrong_position() {\n // Unbracketed with multiple colons is invalid (IPv4 has at most 3 colons,\n // but we require host:port, so only the last colon is the separator).\n assert!(validate_endpoint_for_aeron_udp(\"::1:8080\").is_err());\n\n // Colon at start is invalid (empty host).\n assert!(validate_endpoint_for_aeron_udp(\":8080\").is_err());\n }\n\n #[test]\n fn validate_endpoint_accepts_hyphenated_hostname() {\n assert!(validate_endpoint_for_aeron_udp(\"my-host-name:8080\").is_ok());\n assert!(validate_endpoint_for_aeron_udp(\"host-1:40123\").is_ok());\n }\n\n #[test]\n fn validate_endpoint_accepts_dotted_hostname() {\n assert!(validate_endpoint_for_aeron_udp(\"host.sub.example.com:8080\").is_ok());\n assert!(validate_endpoint_for_aeron_udp(\"a.b.c.d:40123\").is_ok());\n }\n}\n\n// Retryable / unrecoverable classification for Aeron errors. Lives in hand-written code rather\n// than the codegen `common.rs` because the generator drops methods that carry multi-line doc\n// comments. GenericError(-1) and Unknown(_) are intentionally neither \u{2014} `-1` is Aeron\'s catch-all\n// and could mean anything, so the caller must decide (treat as fatal if unsure).\nimpl AeronErrorType {\n /// Transient \u{2014} retry the operation (back off first): back-pressure, admin action, a full\n /// client buffer, or a polling timeout.\n pub fn is_retryable(&self) -> bool {\n self == &AeronErrorType::PublicationBackPressured\n || self == &AeronErrorType::PublicationAdminAction\n || self == &AeronErrorType::ClientErrorBufferFull\n || self == &AeronErrorType::TimedOut\n }\n\n /// Definitively terminal \u{2014} retrying will not help: the publication is closed / exhausted /\n /// errored, or the driver or client has timed out (effectively dead). Not exhaustive: an\n /// ambiguous code (`GenericError` / `Unknown`) is neither retryable nor unrecoverable.\n pub fn is_unrecoverable(&self) -> bool {\n self == &AeronErrorType::PublicationClosed\n || self == &AeronErrorType::PublicationMaxPositionExceeded\n || self == &AeronErrorType::PublicationError\n || self == &AeronErrorType::ClientErrorDriverTimeout\n || self == &AeronErrorType::ClientErrorClientTimeout\n || self == &AeronErrorType::ClientErrorConductorServiceTimeout\n }\n}\n\nimpl AeronCError {\n /// Transient failure \u{2014} retry the operation (back off first). See [`AeronErrorType::is_retryable`].\n pub fn is_retryable(&self) -> bool {\n self.kind().is_retryable()\n }\n\n /// Definitively terminal \u{2014} abort the operation. See [`AeronErrorType::is_unrecoverable`].\n /// Not exhaustive: ambiguous codes are neither retryable nor unrecoverable.\n pub fn is_unrecoverable(&self) -> bool {\n self.kind().is_unrecoverable()\n }\n}\n";