Start work on flyweights and driver proxy

Implemented flyweights four different ways, can't say I like any of them. This is at least a starting point.

aeron-rs/src/command/correlated_message.rs

@@ -0,0 +1,30 @@
+use crate::command::flyweight::Flyweight;
+use crate::concurrent::AtomicBuffer;
+
+pub struct CorrelatedMessageDefn {
+    pub(crate) client_id: i64,
+    pub(crate) correlation_id: i64,
+}
+
+impl<A> Flyweight<A, CorrelatedMessageDefn>
+where
+    A: AtomicBuffer,
+{
+    pub fn client_id(&self) -> i64 {
+        self.get_struct().client_id
+    }
+
+    pub fn put_client_id(&mut self, value: i64) -> &mut Self {
+        self.get_struct_mut().client_id = value;
+        self
+    }
+
+    pub fn correlation_id(&self) -> i64 {
+        self.get_struct().correlation_id
+    }
+
+    pub fn put_correlation_id(&mut self, value: i64) -> &mut Self {
+        self.get_struct_mut().correlation_id = value;
+        self
+    }
+}

aeron-rs/src/command/flyweight.rs

@@ -0,0 +1,47 @@
+use crate::concurrent::AtomicBuffer;
+use crate::util::{IndexT, Result};
+use std::marker::PhantomData;
+
+pub struct Flyweight<A, S>
+where
+    A: AtomicBuffer,
+{
+    pub(in crate::command) buffer: A,
+    base_offset: IndexT,
+    _phantom: PhantomData<S>,
+}
+
+impl<A, S> Flyweight<A, S>
+where
+    A: AtomicBuffer,
+    S: Sized,
+{
+    pub fn new(buffer: A, offset: IndexT) -> Result<Flyweight<A, S>> {
+        buffer.overlay::<S>(offset)?;
+        Ok(Flyweight {
+            buffer,
+            base_offset: offset,
+            _phantom: PhantomData,
+        })
+    }
+
+    pub(crate) fn get_struct(&self) -> &S {
+        // UNWRAP: Bounds check performed during initialization
+        self.buffer.overlay::<S>(self.base_offset).unwrap()
+    }
+
+    pub(crate) fn get_struct_mut(&mut self) -> &mut S {
+        // UNWRAP: Bounds check performed during initialization
+        self.buffer.overlay_mut::<S>(self.base_offset).unwrap()
+    }
+
+    pub(crate) fn bytes_at(&self, offset: IndexT) -> &[u8] {
+        let offset = (self.base_offset + offset) as usize;
+        // FIXME: Unwrap is unjustified here.
+        // C++ uses pointer arithmetic with no bounds checking, so I'm more comfortable
+        // with the Rust version at least panicking. Is the idea that we're safe because
+        // this is a crate-local (protected in C++) method?
+        self.buffer.bounds_check(offset as IndexT, 0).unwrap();
+        &self.buffer[offset..]
+    }
+}

aeron-rs/src/command/mod.rs

@@ -0,0 +1,3 @@
+pub mod correlated_message;
+pub mod flyweight;
+pub mod terminate_driver;

aeron-rs/src/command/terminate_driver.rs

@@ -0,0 +1,68 @@
+use crate::command::correlated_message::CorrelatedMessageDefn;
+use crate::command::flyweight::Flyweight;
+use crate::concurrent::AtomicBuffer;
+use crate::util::IndexT;
+use std::mem::size_of;
+
+pub struct TerminateDriverDefn {
+    pub(crate) correlated_message: CorrelatedMessageDefn,
+    pub(crate) token_length: i32,
+}
+
+impl<A> Flyweight<A, TerminateDriverDefn>
+where
+    A: AtomicBuffer,
+{
+    pub fn client_id(&self) -> i64 {
+        self.get_struct().correlated_message.client_id
+    }
+
+    pub fn put_client_id(&mut self, value: i64) -> &mut Self {
+        self.get_struct_mut().correlated_message.client_id = value;
+        self
+    }
+
+    pub fn correlation_id(&self) -> i64 {
+        self.get_struct().correlated_message.correlation_id
+    }
+
+    pub fn put_correlation_id(&mut self, value: i64) -> &mut Self {
+        self.get_struct_mut().correlated_message.correlation_id = value;
+        self
+    }
+
+    pub fn token_length(&self) -> i32 {
+        self.get_struct().token_length
+    }
+
+    pub fn put_token_length(&mut self, value: i32) -> &mut Self {
+        self.get_struct_mut().token_length = value;
+        self
+    }
+
+    pub fn token_buffer(&self) -> &[u8] {
+        // QUESTION: Should I be slicing the buffer to `token_length`?
+        // C++ doesn't do anything, so I'm going to assume not.
+        &self.bytes_at(size_of::<TerminateDriverDefn>() as IndexT)
+    }
+
+    pub fn put_token_buffer(&mut self, token_buffer: &[u8]) -> &mut Self {
+        let token_length = token_buffer.len() as i32;
+        self.get_struct_mut().token_length = token_length;
+
+        if token_length > 0 {
+            // FIXME: Unwrap is unjustified here
+            // Currently just assume that people are going to be nice about the token buffer
+            // and not oversize it.
+            self.buffer
+                .put_slice(
+                    size_of::<TerminateDriverDefn>() as IndexT,
+                    &token_buffer,
+                    0,
+                    token_length,
+                )
+                .unwrap()
+        }
+        self
+    }
+}

aeron-rs/src/concurrent/mod.rs

@@ -12,6 +12,14 @@ use std::ptr::{read_volatile, write_volatile};
 use memmap::MmapMut;
 use std::ops::{Deref, DerefMut};
 
+fn bounds_check_slice(slice: &[u8], offset: IndexT, size: IndexT) -> Result<()> {
+    if offset < 0 || size < 0 || slice.len() as IndexT - offset < size {
+        Err(AeronError::OutOfBounds)
+    } else {
+        Ok(())
+    }
+}
+
 /// Atomic operations on slices of memory
 pub trait AtomicBuffer: Deref<Target = [u8]> + DerefMut<Target = [u8]> {
     /// Check that there are at least `size` bytes of memory available
@@ -27,11 +35,7 @@ pub trait AtomicBuffer: Deref<Target = [u8]> + DerefMut<Target = [u8]> {
     /// assert!(buffer.bounds_check(-1, 8).is_err());
     /// ```
     fn bounds_check(&self, offset: IndexT, size: IndexT) -> Result<()> {
-        if offset < 0 || size < 0 || self.deref().len() as IndexT - offset < size {
-            Err(AeronError::OutOfBounds)
-        } else {
-            Ok(())
-        }
+        bounds_check_slice(self.deref(), offset, size)
     }
 
     /// Overlay a struct on a buffer.
@@ -199,6 +203,24 @@ pub trait AtomicBuffer: Deref<Target = [u8]> + DerefMut<Target = [u8]> {
         self.overlay_mut::<i64>(offset).map(|i| *i = value)
     }
 
+    fn put_slice(
+        &mut self,
+        index: IndexT,
+        source: &[u8],
+        source_index: IndexT,
+        len: IndexT,
+    ) -> Result<()> {
+        self.bounds_check(index, len)?;
+        bounds_check_slice(source, source_index, len)?;
+
+        let index = index as usize;
+        let source_index = source_index as usize;
+        let len = len as usize;
+
+        self[index..index + len].copy_from_slice(&source[source_index..source_index + len]);
+        Ok(())
+    }
+
     /// Write the contents of one buffer to another. Does not perform any synchronization
     fn put_bytes<B>(
         &mut self,

aeron-rs/src/concurrent/ringbuffer.rs

@@ -25,6 +25,9 @@ pub mod buffer_descriptor {
     /// the start of the ring buffer metadata trailer.
     pub const CORRELATION_COUNTER_OFFSET: IndexT = (CACHE_LINE_LENGTH * 8) as IndexT;
 
+    /// Offset within the ring buffer trailer to the consumer heartbeat timestamp
+    pub const CONSUMER_HEARTBEAT_OFFSET: IndexT = (CACHE_LINE_LENGTH * 10) as IndexT;
+
     /// Total size of the ring buffer metadata trailer.
     pub const TRAILER_LENGTH: IndexT = (CACHE_LINE_LENGTH * 12) as IndexT;
 
@@ -125,6 +128,7 @@ where
     head_cache_position_index: IndexT,
     head_position_index: IndexT,
     correlation_id_counter_index: IndexT,
+    consumer_heartbeat_index: IndexT,
 }
 
 impl<A> ManyToOneRingBuffer<A>
@@ -143,6 +147,7 @@ where
             head_cache_position_index: capacity + buffer_descriptor::HEAD_CACHE_POSITION_OFFSET,
             head_position_index: capacity + buffer_descriptor::HEAD_POSITION_OFFSET,
             correlation_id_counter_index: capacity + buffer_descriptor::CORRELATION_COUNTER_OFFSET,
+            consumer_heartbeat_index: capacity + buffer_descriptor::CONSUMER_HEARTBEAT_OFFSET,
         })
     }
 
@@ -377,6 +382,10 @@ where
     pub fn max_msg_length(&self) -> IndexT {
         self.max_msg_length
     }
+
+    pub fn consumer_heartbeat_time(&self) -> Result<i64> {
+        self.buffer.get_i64_volatile(self.consumer_heartbeat_index)
+    }
 }
 
 impl<A> Deref for ManyToOneRingBuffer<A>

aeron-rs/src/control_protocol.rs

@@ -0,0 +1,107 @@
+use aeron_driver_sys::*;
+
+/// Construct a C-compatible enum out of a set of constants.
+/// Commonly used for types in Aeron that have fixed values via `#define`,
+/// but aren't actually enums (e.g. AERON_COMMAND_.*, AERON_ERROR_CODE_.*).
+/// Behavior is ultimately very similar to `num::FromPrimitive`.
+macro_rules! define_enum {
+    (
+        $(#[$outer:meta])*
+        pub enum $name:ident {$(
+            $(#[$inner:meta]),*
+            $left:ident = $right:ident,
+        )+}
+    ) => {
+        #[repr(u32)]
+        #[derive(Debug, PartialEq)]
+        $(#[$outer])*
+        pub enum $name {$(
+            $(#[$inner])*
+            $left = $right,
+        )*}
+
+        impl ::std::convert::TryFrom<u32> for $name {
+            type Error = ();
+            fn try_from(val: u32) -> Result<$name, ()> {
+                match val {
+                    $(v if v == $name::$left as u32 => Ok($name::$left)),*,
+                    _ => Err(())
+                }
+            }
+        }
+    }
+}
+
+define_enum!(
+    #[doc = "Commands sent from clients to the Media Driver"]
+    pub enum ClientCommand {
+        #[doc = "Add a Publication"]
+        AddPublication = AERON_COMMAND_ADD_PUBLICATION,
+        #[doc = "Remove a Publication"]
+        RemovePublication = AERON_COMMAND_REMOVE_PUBLICATION,
+        #[doc = "Add an Exclusive Publication"]
+        AddExclusivePublication = AERON_COMMAND_ADD_EXCLUSIVE_PUBLICATION,
+        #[doc = "Add a Subscriber"]
+        AddSubscription = AERON_COMMAND_ADD_SUBSCRIPTION,
+        #[doc = "Remove a Subscriber"]
+        RemoveSubscription = AERON_COMMAND_REMOVE_SUBSCRIPTION,
+        #[doc = "Keepalaive from Client"]
+        ClientKeepalive = AERON_COMMAND_CLIENT_KEEPALIVE,
+        #[doc = "Add Destination to an existing Publication"]
+        AddDestination = AERON_COMMAND_ADD_DESTINATION,
+        #[doc = "Remove Destination from an existing Publication"]
+        RemoveDestination = AERON_COMMAND_REMOVE_DESTINATION,
+        #[doc = "Add a Counter to the counters manager"]
+        AddCounter = AERON_COMMAND_ADD_COUNTER,
+        #[doc = "Remove a Counter from the counters manager"]
+        RemoveCounter = AERON_COMMAND_REMOVE_COUNTER,
+        #[doc = "Close indication from Client"]
+        ClientClose = AERON_COMMAND_CLIENT_CLOSE,
+        #[doc = "Add Destination for existing Subscription"]
+        AddRcvDestination = AERON_COMMAND_ADD_RCV_DESTINATION,
+        #[doc = "Remove Destination for existing Subscription"]
+        RemoveRcvDestination = AERON_COMMAND_REMOVE_RCV_DESTINATION,
+        #[doc = "Request the driver to terminate"]
+        TerminateDriver = AERON_COMMAND_TERMINATE_DRIVER,
+    }
+);
+
+define_enum!(
+    #[doc = "Responses from the Media Driver to client commands"]
+    pub enum DriverResponse {
+        #[doc = "Error Response as a result of attempting to process a client command operation"]
+        OnError = AERON_RESPONSE_ON_ERROR,
+        #[doc = "Subscribed Image buffers are available notification"]
+        OnAvailableImage = AERON_RESPONSE_ON_AVAILABLE_IMAGE,
+        #[doc = "New Publication buffers are ready notification"]
+        OnPublicationReady = AERON_RESPONSE_ON_PUBLICATION_READY,
+        #[doc = "Operation has succeeded"]
+        OnOperationSuccess = AERON_RESPONSE_ON_OPERATION_SUCCESS,
+        #[doc = "Inform client of timeout and removal of an inactive Image"]
+        OnUnavailableImage = AERON_RESPONSE_ON_UNAVAILABLE_IMAGE,
+        #[doc = "New Exclusive Publication buffers are ready notification"]
+        OnExclusivePublicationReady = AERON_RESPONSE_ON_EXCLUSIVE_PUBLICATION_READY,
+        #[doc = "New Subscription is ready notification"]
+        OnSubscriptionReady = AERON_RESPONSE_ON_SUBSCRIPTION_READY,
+        #[doc = "New counter is ready notification"]
+        OnCounterReady = AERON_RESPONSE_ON_COUNTER_READY,
+        #[doc = "Inform clients of removal of counter"]
+        OnUnavailableCounter = AERON_RESPONSE_ON_UNAVAILABLE_COUNTER,
+        #[doc = "Inform clients of client timeout"]
+        OnClientTimeout = AERON_RESPONSE_ON_CLIENT_TIMEOUT,
+    }
+);
+
+#[cfg(test)]
+mod tests {
+    use crate::control_protocol::ClientCommand;
+    use std::convert::TryInto;
+
+    #[test]
+    fn client_command_convert() {
+        assert_eq!(
+            Ok(ClientCommand::AddPublication),
+            ::aeron_driver_sys::AERON_COMMAND_ADD_PUBLICATION.try_into()
+        )
+    }
+}

aeron-rs/src/driver_proxy.rs

@@ -0,0 +1,73 @@
+use crate::command::flyweight::Flyweight;
+use crate::command::terminate_driver::TerminateDriverDefn;
+use crate::concurrent::ringbuffer::ManyToOneRingBuffer;
+use crate::concurrent::AtomicBuffer;
+use crate::control_protocol::ClientCommand;
+use crate::util::{AeronError, IndexT, Result};
+
+pub struct DriverProxy<A>
+where
+    A: AtomicBuffer,
+{
+    to_driver: ManyToOneRingBuffer<A>,
+    client_id: i64,
+}
+
+impl<A> DriverProxy<A>
+where
+    A: AtomicBuffer,
+{
+    pub fn new(to_driver: ManyToOneRingBuffer<A>) -> Self {
+        let client_id = to_driver.next_correlation_id();
+        DriverProxy {
+            to_driver,
+            client_id,
+        }
+    }
+
+    pub fn time_of_last_driver_keepalive(&self) -> Result<i64> {
+        self.to_driver.consumer_heartbeat_time()
+    }
+
+    pub fn client_id(&self) -> i64 {
+        self.client_id
+    }
+
+    pub fn terminate_driver(&mut self, _token_buffer: Option<&[u8]>) -> Result<()> {
+        let _client_id = self.client_id;
+        self.write_command_to_driver(|buffer: &mut [u8], _length: &mut IndexT| {
+            // FIXME: This method signature is ugly.
+            // UNWRAP: Buffer from `write_command` guaranteed to be long enough for `TerminateDriverDefn`
+            let _request: Flyweight<_, TerminateDriverDefn> = Flyweight::new(buffer, 0).unwrap();
+
+            // FIXME: Uncommenting this causes termination to not succeed
+            /*
+            request.put_client_id(client_id).put_correlation_id(-1);
+            token_buffer.map(|b| request.put_token_buffer(b));
+            *length = request.token_length();
+            */
+
+            ClientCommand::TerminateDriver
+        })
+    }
+
+    fn write_command_to_driver<F>(&mut self, filler: F) -> Result<()>
+    where
+        F: FnOnce(&mut [u8], &mut IndexT) -> ClientCommand,
+    {
+        // QUESTION: Can Rust align structs on stack?
+        // C++ does some fancy shenanigans I assume help the CPU cache?
+        let mut buffer = &mut [0u8; 512][..];
+        let mut length = buffer.len() as IndexT;
+        let msg_type_id = filler(&mut buffer, &mut length);
+
+        if !self
+            .to_driver
+            .write(msg_type_id as i32, &buffer, 0, length)?
+        {
+            Err(AeronError::IllegalState)
+        } else {
+            Ok(())
+        }
+    }
+}

aeron-rs/src/lib.rs

@@ -1,13 +1,15 @@
 //! [Aeron](https://github.com/real-logic/aeron) client for Rust
-#![deny(missing_docs)]
 
 #[cfg(target_endian = "big")]
 compile_error!("Aeron is only supported on little-endian architectures");
 
 pub mod cnc_descriptor;
+pub mod command;
 pub mod concurrent;
 pub mod context;
+pub mod control_protocol;
 pub mod driver;
+pub mod driver_proxy;
 pub mod util;
 
 const fn sematic_version_compose(major: u8, minor: u8, patch: u8) -> i32 {

aeron-rs/tests/broadcast_receiver.rs

@@ -1,6 +1,4 @@
-use aeron_rs::concurrent::broadcast::{
-    buffer_descriptor, record_descriptor, BroadcastReceiver,
-};
+use aeron_rs::concurrent::broadcast::{buffer_descriptor, record_descriptor, BroadcastReceiver};
 use aeron_rs::concurrent::AtomicBuffer;
 use aeron_rs::util::bit::align;
 use aeron_rs::util::IndexT;

aeron-rs/tests/many_to_one_ring_buffer.rs

@@ -1,7 +1,5 @@
 /// Tests based on the C++ tests included with Aeron
-use aeron_rs::concurrent::ringbuffer::{
-    buffer_descriptor, record_descriptor, ManyToOneRingBuffer,
-};
+use aeron_rs::concurrent::ringbuffer::{buffer_descriptor, record_descriptor, ManyToOneRingBuffer};
 use aeron_rs::concurrent::AtomicBuffer;
 use aeron_rs::util::bit::align;
 use aeron_rs::util::IndexT;