Init + Uptime timers

src/transport/middle_proxy/send.rs

@@ -11,7 +11,7 @@ use tracing::{debug, warn};
 use crate::config::MeRouteNoWriterMode;
 use crate::error::{ProxyError, Result};
 use crate::network::IpFamily;
-use crate::protocol::constants::RPC_CLOSE_EXT_U32;
+use crate::protocol::constants::{RPC_CLOSE_CONN_U32, RPC_CLOSE_EXT_U32};
 
 use super::MePool;
 use super::codec::WriterCommand;
@@ -476,6 +476,37 @@ impl MePool {
         Ok(())
     }
 
+    pub async fn send_close_conn(self: &Arc<Self>, conn_id: u64) -> Result<()> {
+        if let Some(w) = self.registry.get_writer(conn_id).await {
+            let mut p = Vec::with_capacity(12);
+            p.extend_from_slice(&RPC_CLOSE_CONN_U32.to_le_bytes());
+            p.extend_from_slice(&conn_id.to_le_bytes());
+            match w.tx.try_send(WriterCommand::DataAndFlush(p)) {
+                Ok(()) => {}
+                Err(TrySendError::Full(cmd)) => {
+                    let _ = tokio::time::timeout(Duration::from_millis(50), w.tx.send(cmd)).await;
+                }
+                Err(TrySendError::Closed(_)) => {
+                    debug!(conn_id, "ME close_conn skipped: writer channel closed");
+                }
+            }
+        } else {
+            debug!(conn_id, "ME close_conn skipped (writer missing)");
+        }
+
+        self.registry.unregister(conn_id).await;
+        Ok(())
+    }
+
+    pub async fn shutdown_send_close_conn_all(self: &Arc<Self>) -> usize {
+        let conn_ids = self.registry.active_conn_ids().await;
+        let total = conn_ids.len();
+        for conn_id in conn_ids {
+            let _ = self.send_close_conn(conn_id).await;
+        }
+        total
+    }
+
     pub fn connection_count(&self) -> usize {
         self.conn_count.load(Ordering::Relaxed)
     }