Constant CUSTOM_AERON_CODE
Source 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 =\n unsafe { std::ffi::CStr::from_bytes_with_nul_unchecked(b\"aeron:ipc\\0\") };\n\nunsafe impl Send for AeronCountersReader {}\nunsafe impl Send for AeronSubscription {}\nunsafe impl Sync for AeronSubscription {}\nunsafe impl Send for AeronPublication {}\nunsafe impl Sync for AeronPublication {}\nunsafe impl Send for AeronExclusivePublication {}\nunsafe impl Sync for AeronExclusivePublication {}\nunsafe impl Send for AeronCounter {}\nunsafe impl Sync for AeronCounter {}\n\nimpl AeronCnc {\n /// Note this allocates the rust component on stack but the C aeron_cnc_t struct is still on the heap,\n /// as Aeron does the allocation.\n #[inline]\n pub fn read_on_partial_stack(\n aeron_dir: &std::ffi::CString,\n mut handler: impl FnMut(&mut AeronCnc),\n ) -> Result<(), AeronCError> {\n let cnc = ManagedCResource::initialise(move |cnc| unsafe {\n aeron_cnc_init(cnc, aeron_dir.as_ptr(), 0)\n })?;\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 /// **Deprecated**: allocate on the heap. Use `new_on_heap` instead.\n #[deprecated(since = \"0.1.122\", note = \"Use `new_on_heap` instead\")]\n #[inline]\n pub fn new(aeron_dir: &str) -> Result<AeronCnc, AeronCError> {\n Self::new_on_heap(aeron_dir)\n }\n\n /// Note this allocates on the heap, cannot be stored this on stack. As Aeron will do the allocation.\n /// Try to use `read_on_partial_stack` which performs less allocations\n #[inline]\n pub fn new_on_heap(aeron_dir: &str) -> Result<AeronCnc, AeronCError> {\n let c_string = std::ffi::CString::new(aeron_dir).expect(\"CString conversion failed\");\n let resource = ManagedCResource::new(\n move |cnc| unsafe { aeron_cnc_init(cnc, c_string.as_ptr(), 0) },\n Some(Box::new(move |cnc| unsafe {\n aeron_cnc_close(*cnc);\n 0\n })),\n false,\n None,\n )?;\n\n let result = Self {\n inner: CResource::OwnedOnHeap(std::rc::Rc::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\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\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).expect(\"CString::new failed\");\n let mapped_file = std::rc::Rc::new(std::cell::RefCell::new(aeron_mapped_file_t {\n addr: std::ptr::null_mut(),\n length: 0,\n }));\n let mapped_file2 = std::rc::Rc::clone(&mapped_file);\n let resource = ManagedCResource::new(\n move |ctx| {\n let result = unsafe {\n aeron_cnc_map_file_and_load_metadata(\n aeron_dir.as_ptr(),\n mapped_file.borrow_mut().deref_mut() 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| unsafe {\n aeron_unmap(mapped_file2.borrow_mut().deref_mut() as *mut aeron_mapped_file_t)\n })),\n false,\n None,\n )?;\n\n let result = Self {\n inner: CResource::OwnedOnHeap(std::rc::Rc::new(resource)),\n };\n Ok(result)\n }\n\n #[inline]\n /// allocates on stack\n pub fn read_from_file(\n aeron_dir: &std::ffi::CString,\n mut handler: impl FnMut(Self),\n ) -> 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\nimpl AeronSubscription {\n pub fn close_with_no_args(&mut self) -> Result<(), AeronCError> {\n self.close(Handlers::no_notification_handler())?;\n Ok(())\n }\n}\n\nimpl AeronPublication {\n pub fn close_with_no_args(&self) -> Result<(), AeronCError> {\n self.close(Handlers::no_notification_handler())?;\n Ok(())\n }\n\n /// sometimes when you first connect, is_connected = true, but you get backpressure as position is 0\n /// this will check if both publication is connected and position > 0\n #[inline]\n pub fn is_ready(&self) -> bool {\n self.is_connected() && self.position_limit() != 0\n }\n}\n\nimpl AeronExclusivePublication {\n pub fn close_with_no_args(&self) -> Result<(), AeronCError> {\n self.close(Handlers::no_notification_handler())?;\n Ok(())\n }\n\n /// sometimes when you first connect, is_connected = true, but you get backpressure as position is 0\n /// this will check if both publication is connected and position > 0\n #[inline]\n pub fn is_ready(&self) -> bool {\n self.is_connected() && self.position_limit() != 0\n }\n}\n\nimpl AeronCounter {\n pub fn close_with_no_args(&self) -> Result<(), AeronCError> {\n self.close(Handlers::no_notification_handler())?;\n Ok(())\n }\n}\n\nimpl AeronCounter {\n #[inline]\n pub fn addr_atomic(&self) -> &std::sync::atomic::AtomicI64 {\n unsafe { std::sync::atomic::AtomicI64::from_ptr(self.addr()) }\n }\n}\n\nimpl AeronSubscription {\n pub fn async_add_destination(\n &mut 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 pub fn add_destination(\n &mut self,\n client: &Aeron,\n destination: &std::ffi::CStr,\n timeout: std::time::Duration,\n ) -> Result<(), AeronCError> {\n let result = self.async_add_destination(client, destination)?;\n if result\n .aeron_subscription_async_destination_poll()\n .unwrap_or_default()\n > 0\n {\n return Ok(());\n }\n let time = std::time::Instant::now();\n while time.elapsed() < timeout {\n if result\n .aeron_subscription_async_destination_poll()\n .unwrap_or_default()\n > 0\n {\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 AeronExclusivePublication {\n pub fn async_add_destination(\n &mut self,\n client: &Aeron,\n destination: &std::ffi::CStr,\n ) -> Result<AeronAsyncDestination, AeronCError> {\n AeronAsyncDestination::aeron_exclusive_publication_async_add_destination(\n client,\n self,\n destination,\n )\n }\n\n pub fn add_destination(\n &mut self,\n client: &Aeron,\n destination: &std::ffi::CStr,\n timeout: std::time::Duration,\n ) -> Result<(), AeronCError> {\n let result = self.async_add_destination(client, destination)?;\n if result\n .aeron_subscription_async_destination_poll()\n .unwrap_or_default()\n > 0\n {\n return Ok(());\n }\n let time = std::time::Instant::now();\n while time.elapsed() < timeout {\n if result\n .aeron_subscription_async_destination_poll()\n .unwrap_or_default()\n > 0\n {\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 &mut 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 pub fn add_destination(\n &mut self,\n client: &Aeron,\n destination: &std::ffi::CStr,\n timeout: std::time::Duration,\n ) -> Result<(), AeronCError> {\n let result = self.async_add_destination(client, destination)?;\n if result\n .aeron_subscription_async_destination_poll()\n .unwrap_or_default()\n > 0\n {\n return Ok(());\n }\n let time = std::time::Instant::now();\n while time.elapsed() < timeout {\n if result\n .aeron_subscription_async_destination_poll()\n .unwrap_or_default()\n > 0\n {\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).expect(\"CString::new failed\");\n builder.init_on_string(&s)?;\n Ok(builder)\n }\n}\n\nconst PARSE_CSTR_ERROR_CODE: i32 = -132131;\n\nimpl AeronUriStringBuilder {\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_string(&self, key: &std::ffi::CStr, value: &str) -> Result<&Self, AeronCError> {\n let value = std::ffi::CString::new(value)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n self.put(&key, &value)?;\n Ok(self)\n }\n\n pub fn put_strings(&self, key: &str, value: &str) -> Result<&Self, AeronCError> {\n let key = std::ffi::CString::new(key)\n .map_err(|_| AeronCError::from_code(PARSE_CSTR_ERROR_CODE))?;\n let value = std::ffi::CString::new(value)\n .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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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_string(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 =\n 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\nimpl AeronCountersReader {\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(\n &self,\n counter_id: i32,\n max_length: usize,\n ) -> 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(\n &self,\n counter_id: i32,\n dst: &mut String,\n ) -> Result<(), AeronCError> {\n unsafe {\n let capacity = dst.capacity();\n let vec = dst.as_mut_vec();\n vec.set_len(capacity);\n self.counter_label(counter_id, vec.as_mut_ptr() as *mut _, capacity)?;\n let mut len = 0;\n loop {\n if len == capacity {\n break;\n }\n let val = vec[len];\n if val == 0 {\n break;\n }\n len += 1;\n }\n vec.set_len(len);\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(\n &self,\n counter_id: i32,\n dst: &mut Vec<u8>,\n ) -> Result<(), AeronCError> {\n let mut key_ptr: *mut u8 = std::ptr::null_mut();\n unsafe {\n let result = bindings::aeron_counters_reader_metadata_key(\n self.get_inner(),\n counter_id,\n &mut key_ptr,\n );\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(\n context: &AeronContext,\n timeout: std::time::Duration,\n ) -> 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> Handler<T> {\n pub fn leak_with_fragment_assembler(\n handler: T,\n ) -> Result<(Handler<AeronFragmentAssembler>, Handler<T>), AeronCError> {\n let handler = Handler::leak(handler);\n Ok((\n Handler::leak(AeronFragmentAssembler::new(Some(&handler))?),\n handler,\n ))\n }\n}\n\nimpl<T: AeronControlledFragmentHandlerCallback> Handler<T> {\n pub fn leak_with_controlled_fragment_assembler(\n handler: T,\n ) -> Result<(Handler<AeronControlledFragmentAssembler>, Handler<T>), AeronCError> {\n let handler = Handler::leak(handler);\n Ok((\n Handler::leak(AeronControlledFragmentAssembler::new(Some(&handler))?),\n handler,\n ))\n }\n}\n\nimpl AeronBufferClaim {\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 unsafe { &mut *self.frame_header.cast::<aeron_header_values_frame_t>() }\n }\n}\n\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 /// SAFETY: you must make sure ctx lives longer than when `call` method is invoked\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 #[inline]\n fn wrap<T>(\n f: fn(&mut T, &[u8], AeronHeader)\n ) -> 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\npub struct AeronFragmentClosureAssembler {\n assembler: AeronFragmentAssembler,\n handler: Handler<FnMutMessageHandler>,\n assembler_handler: Handler<AeronFragmentAssembler>,\n}\n\nimpl AeronFragmentClosureAssembler {\n pub fn new() -> Result<Self, AeronCError> {\n let handler = Handler::leak(FnMutMessageHandler::new());\n Ok(Self {\n assembler: AeronFragmentAssembler::new(Some(&handler))?,\n handler,\n assembler_handler: Handler {\n raw_ptr: std::ptr::null_mut(),\n should_drop: false,\n },\n })\n }\n\n pub fn process<T>(&mut self, ctx: &mut T, func: fn(&mut T, &[u8], AeronHeader)) -> Option<&Handler<AeronFragmentAssembler>> {\n self.handler.set(ctx, func);\n self.assembler_handler.raw_ptr = &mut self.assembler as *mut _;\n Some(&self.assembler_handler)\n }\n}\nimpl Drop for AeronFragmentClosureAssembler {\n fn drop(&mut self) {\n self.handler.release();\n }\n}\n";