refactor(transport): wrap transport stats in Arc

Author: mempirate

msg-socket/src/req/mod.rs

@@ -186,7 +186,7 @@ pub(crate) struct SocketState<S> {
     /// The socket stats.
     pub(crate) stats: Arc<SocketStats<ReqStats>>,
     /// The transport stats. This is None until a connection is established.
-    pub(crate) transport: (watch::Sender<S>, watch::Receiver<S>),
+    pub(crate) transport: (watch::Sender<Arc<S>>, watch::Receiver<Arc<S>>),
 }
 
 // Manual clone implementation needed here because `S` is n`.

msg-socket/src/req/socket.rs

@@ -5,7 +5,7 @@ use tokio::{
     net::{ToSocketAddrs, lookup_host},
     sync::{
         mpsc, oneshot,
-        watch::{self, Ref},
+        watch::{self},
     },
 };
 
@@ -76,7 +76,7 @@ where
             options: Arc::new(options),
             state: SocketState {
                 stats: Arc::new(SocketStats::default()),
-                transport: watch::channel(T::Stats::default()),
+                transport: watch::channel(Arc::new(T::Stats::default())),
             },
             compressor: None,
             _marker: PhantomData,
@@ -95,8 +95,8 @@ where
     }
 
     /// Borrow the latest transport-level stats snapshot.
-    pub fn transport_stats(&self) -> Ref<'_, T::Stats> {
-        self.state.transport.1.borrow()
+    pub fn transport_stats(&self) -> Arc<T::Stats> {
+        self.state.transport.1.borrow().clone()
     }
 
     pub async fn request(&self, message: Bytes) -> Result<Bytes, ReqError> {

msg-transport/src/lib.rs

@@ -10,6 +10,7 @@ use std::{
     net::SocketAddr,
     path::PathBuf,
     pin::Pin,
+    sync::Arc,
     task::{Context, Poll},
     time::{Duration, Instant},
 };
@@ -46,7 +47,7 @@ where
     /// The inner IO object.
     inner: Io,
     /// The sender for the stats.
-    sender: watch::Sender<S>,
+    sender: watch::Sender<Arc<S>>,
     /// The next time the stats should be refreshed.
     next_refresh: Instant,
     /// The interval at which the stats should be refreshed.
@@ -129,7 +130,7 @@ where
     /// stats. The `sender` is used to send the latest stats to the caller.
     ///
     /// TODO: Specify configuration options.
-    pub fn new(inner: Io, sender: watch::Sender<S>) -> Self {
+    pub fn new(inner: Io, sender: watch::Sender<Arc<S>>) -> Self {
         Self {
             inner,
             sender,
@@ -145,7 +146,7 @@ where
         if self.next_refresh <= now {
             match S::try_from(&self.inner) {
                 Ok(stats) => {
-                    if let Err(e) = self.sender.send(stats) {
+                    if let Err(e) = self.sender.send(Arc::new(stats)) {
                         tracing::error!(err = ?e, "failed to update transport stats");
                     }
                 }