Optimize the codes

rmqtt · Feb 29, 2024 · da08d5e · da08d5e
1 parent 3b8e3ff
commit da08d5e
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 8 deletions.
diff --git a/rmqtt/src/broker/default.rs b/rmqtt/src/broker/default.rs
@@ -1146,7 +1146,7 @@ impl DefaultSharedSubscription {
 impl SharedSubscription for &'static DefaultSharedSubscription {}
 
 pub struct DefaultRetainStorage {
-    messages: RwLock<RetainTree<TimedValue<Retain>>>,
+    pub messages: RwLock<RetainTree<TimedValue<Retain>>>,
 }
 
 impl DefaultRetainStorage {
@@ -1157,7 +1157,7 @@ impl DefaultRetainStorage {
     }
 
     #[inline]
-    pub async fn remove_expired_messages(&self) {
+    pub async fn remove_expired_messages(&self) -> usize {
         let mut messages = self.messages.write().await;
         messages.retain(usize::MAX, |tv| {
             if tv.is_expired() {
@@ -1166,7 +1166,7 @@ impl DefaultRetainStorage {
             } else {
                 true
             }
-        });
+        })
     }
 
     #[inline]
@@ -1226,12 +1226,12 @@ impl RetainStorage for &'static DefaultRetainStorage {
     }
 
     #[inline]
-    fn count(&self) -> isize {
+    async fn count(&self) -> isize {
         Runtime::instance().stats.retaineds.count()
     }
 
     #[inline]
-    fn max(&self) -> isize {
+    async fn max(&self) -> isize {
         Runtime::instance().stats.retaineds.max()
     }
 }

diff --git a/rmqtt/src/broker/mod.rs b/rmqtt/src/broker/mod.rs
@@ -282,10 +282,10 @@ pub trait RetainStorage: Sync + Send {
     async fn get(&self, topic_filter: &TopicFilter) -> Result<Vec<(TopicName, Retain)>>;
 
     ///
-    fn count(&self) -> isize;
+    async fn count(&self) -> isize;
 
     ///
-    fn max(&self) -> isize;
+    async fn max(&self) -> isize;
 }
 
 #[async_trait]

diff --git a/rmqtt/src/broker/stats.rs b/rmqtt/src/broker/stats.rs
@@ -216,6 +216,12 @@ impl Stats {
             self.message_storages.max_max(message_mgr.max().await);
         }
 
+        {
+            let retain = Runtime::instance().extends.retain().await;
+            self.retaineds.current_set(retain.count().await);
+            self.retaineds.max_max(retain.max().await);
+        }
+
         #[cfg(feature = "debug")]
         let shared = Runtime::instance().extends.shared().await;
 

diff --git a/rmqtt/src/node.rs b/rmqtt/src/node.rs
@@ -51,7 +51,11 @@ impl Node {
                 .unwrap();
             let runner = async {
                 if let Err(e) = Server::new().listen_and_serve().await {
-                    log::error!("listen and serve failure, {:?}", e);
+                    log::error!(
+                        "listen and serve failure, {:?}, laddr: {:?}",
+                        e,
+                        Runtime::instance().settings.rpc.server_addr
+                    );
                 }
             };
             rt.block_on(runner)