still looking

Author: nathantippy

core/src/distributed/distributed_stream.rs

@@ -5,7 +5,7 @@
 //! `aeron_subscribe`.
 
 use crate::core_tx::TxCore;
-use crate::{abstract_executor, channel_builder::ChannelBuilder, monitor::{RxMetaData, TxMetaData}, Rx, SteadyCommander, Tx};
+use crate::{abstract_executor, channel_builder::ChannelBuilder, Rx, SteadyCommander, Tx};
 use ahash::AHashMap;
 use async_ringbuf::wrap::AsyncWrap;
 use async_ringbuf::AsyncRb;
@@ -867,15 +867,29 @@ impl<T: StreamItem> LazyStreamRx<T> {
     }
 }
 
+//TODO: these two methods control streams showing items not bytes, this may need to be corrected.
+
 impl<T: StreamItem> RxMetaDataProvider for SteadyStreamRx<T> {
-    fn meta_data(&self) -> RxMetaData {
-        self.meta_data()
+    fn meta_data(&self) -> Arc<ChannelMetaData> {
+        match self.try_lock() {
+            Some(guard) => guard.item_channel.channel_meta_data.meta_data(),
+            None => {
+                let guard = abstract_executor::block_on(self.lock());
+                guard.item_channel.channel_meta_data.meta_data()
+            }
+        }
     }
 }
 
 impl<T: StreamItem> TxMetaDataProvider for SteadyStreamTx<T> {
-    fn meta_data(&self) -> TxMetaData {
-        self.meta_data()
+    fn meta_data(&self) -> Arc<ChannelMetaData> {
+        match self.try_lock() {
+            Some(guard) => guard.item_channel.channel_meta_data.meta_data(),
+            None => {
+                let guard = abstract_executor::block_on(self.lock());
+                guard.item_channel.channel_meta_data.meta_data()
+            }
+        }
     }
 }
 

core/src/monitor.rs

@@ -947,20 +947,22 @@ pub(crate) mod monitor_tests {
     }
 
     // Test for wait_shutdown
-    // #[async_std::test]
-    // async fn test_wait_shutdown() {
-    //     let context = test_steady_context();
-    //     let monitor = context.into_monitor([], []);
-    //
-    //     // Simulate shutdown
-    //     {
-    //         let mut liveliness = monitor.runtime_state.write();
-    //         liveliness.request_shutdown();
-    //     }
-    //
-    //     let result = monitor.wait_shutdown().await;
-    //     assert!(result);
-    // }
+    #[async_std::test]
+    async fn test_wait_shutdown() {
+        let context = test_steady_context();
+        error!("expecting hang 1");
+        let monitor = context.into_monitor([], []);
+       error!("expecting hang 2");
+        // Simulate shutdown
+        {
+            let mut liveliness = monitor.runtime_state.write();
+            liveliness.request_shutdown();
+        }
+        error!("expecting hang 3");
+
+        let result = monitor.wait_shutdown().await;
+        assert!(result);
+    }
 
     // Test for wait_periodic
     #[async_std::test]

core/src/steady_rx.rs

@@ -372,7 +372,7 @@ impl<T> Rx<T> {
     }
 
     //  difficult to move because we have dual iterators and peek
-    pub(crate) async fn shared_peek_async_iter_timeout(&mut self, wait_for_count: usize, timeout: Option<Duration>) -> impl Iterator<Item = &T> {
+    pub(crate) async fn _shared_peek_async_iter_timeout(&mut self, wait_for_count: usize, timeout: Option<Duration>) -> impl Iterator<Item = &T> {
         let mut one_down = &mut self.oneshot_shutdown;
         if !one_down.is_terminated() {
             let mut operation = &mut self.rx.wait_occupied(wait_for_count);

core/src/telemetry/metrics_server.rs

@@ -41,9 +41,9 @@ pub(crate) async fn run(context: SteadyContext, rx: SteadyRx<DiagramData>) -> Re
                             , steady_config::telemetry_server_port()));
 
     let frame_rate_ms = context.frame_rate_ms;
-    let mut ctrl = context;
+    let ctrl = context;
     #[cfg(feature = "telemetry_on_telemetry")]
-    let mut ctrl = ctrl.into_monitor([&rx], []);
+    let ctrl = ctrl.into_monitor([&rx], []);
 
     internal_behavior(ctrl, frame_rate_ms, rx, addr).await
 }