edge testing for aeron

Author: nathantippy

core/src/distributed/aeron_publish.rs

@@ -25,8 +25,13 @@ pub async fn run(context: SteadyContext
              , aeron:Arc<futures_util::lock::Mutex<Aeron>>
              , state: SteadyState<AeronPublishSteadyState>) -> Result<(), Box<dyn Error>> {
 
-    internal_behavior(context.into_monitor([&rx], [])
-                      , rx, aeron_connect, stream_id, aeron, state).await
+    let cmd = context.into_monitor([&rx], []);
+    if cfg!(not(test)) {
+        internal_behavior(cmd, rx, aeron_connect, stream_id, aeron, state).await
+    } else {
+        cmd.simulated_behavior(vec!(&TestEquals(rx))).await
+    }
+
 }
 
 async fn internal_behavior<C: SteadyCommander>(mut cmd: C

core/src/distributed/aeron_subscribe.rs

@@ -29,10 +29,13 @@ pub async fn run(context: SteadyContext
                  , stream_id: i32
                  , aeron:Arc<futures_util::lock::Mutex<Aeron>>
                  , state: SteadyState<AeronSubscribeSteadyState>) -> Result<(), Box<dyn Error>> {
-    internal_behavior(context.into_monitor([], [&tx])
-                      , tx, aeron_connect, stream_id, aeron, state).await
+    let cmd = context.into_monitor([], [&tx]);
+    if cfg!(not(test)) {
+        internal_behavior(cmd, tx, aeron_connect, stream_id, aeron, state).await
+    } else {
+        cmd.simulated_behavior(vec!(&TestEcho(tx))).await
+    }
 }
-
 async fn internal_behavior<C: SteadyCommander>(mut cmd: C
                  , tx: SteadyStreamTx<StreamSessionMessage>
                  , aeron_channel: Channel

core/src/distributed/aeron_subscribe_bundle.rs

@@ -32,7 +32,17 @@ pub async fn run<const GIRTH:usize,>(context: SteadyContext
                                      , stream_id: i32
                                      , aeron:Arc<futures_util::lock::Mutex<Aeron>>
                                      , state: SteadyState<AeronSubscribeSteadyState>) -> Result<(), Box<dyn Error>> {
-    internal_behavior(context.into_monitor([], tx.control_meta_data()), tx, aeron_connect, stream_id, aeron, state).await
+
+    let cmd = context.into_monitor([], tx.control_meta_data());
+    if cfg!(not(test)) {
+        internal_behavior(cmd, tx, aeron_connect, stream_id, aeron, state).await
+    } else {
+        let te:Vec<_> = tx.iter()
+            .map(|f| TestEcho(f.clone()) ).collect();
+        let sims:Vec<_> = te.iter()
+            .map(|f| f as &dyn IntoSimRunner<_>).collect();
+        cmd.simulated_behavior(sims).await
+    }
 }
 
 //