refactor: standardize channel naming with tx and rx suffixes

Rename internal channel senders and receivers to follow standard Rust conventions for improved code consistency and readability.

src/baseliner.rs

@@ -24,8 +24,8 @@ pub struct Baseliner {
     pub owd_recent: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>,
     pub reselect_trigger: Sender<bool>,
     pub start_time: Instant,
-    pub stats_receiver: Receiver<PingReply>,
-    pub metrics_sender: SyncSender<Metric>,
+    pub stats_rx: Receiver<PingReply>,
+    pub metrics_tx: SyncSender<Metric>,
 }
 
 fn ewma_factor(tick: f64, dur: f64) -> f64 {
@@ -49,7 +49,7 @@ impl Baseliner {
         let clock = Time::new(ClockId::Realtime);
 
         loop {
-            let time_data = self.stats_receiver.recv()?;
+            let time_data = self.stats_rx.recv()?;
 
             let mut owd_baseline_map = self.owd_baseline.lock_anyhow()?;
             let mut owd_recent_map = self.owd_recent.lock_anyhow()?;
@@ -107,7 +107,7 @@ impl Baseliner {
                     "Reflector {} has OWD > 5 seconds more than baseline, triggering reselection",
                     time_data.reflector
                 );
-                let _ = self.metrics_sender.try_send(Metric::Event {
+                let _ = self.metrics_tx.try_send(Metric::Event {
                     name: "reselection",
                     reason: "anomaly",
                     reflector: Some(time_data.reflector),
@@ -136,7 +136,7 @@ impl Baseliner {
                 }
             }
 
-            let _ = self.metrics_sender.try_send(Metric::Baseline {
+            let _ = self.metrics_tx.try_send(Metric::Baseline {
                 reflector: time_data.reflector,
                 baseline_up_ewma: owd_baseline.up_ewma,
                 baseline_down_ewma: owd_baseline.down_ewma,

src/main.rs

@@ -79,24 +79,24 @@ fn main() -> anyhow::Result<()> {
         }
     }
 
-    let (baseliner_stats_sender, baseliner_stats_receiver) = channel();
-    let (reselect_sender, reselect_receiver) = channel();
+    let (baseliner_stats_tx, baseliner_stats_rx) = channel();
+    let (reselect_tx, reselect_rx) = channel();
 
-    let (metrics_sender, metrics_receiver) = sync_channel(1000);
+    let (metrics_tx, metrics_rx) = sync_channel(1000);
 
     let ping_tx = config
         .observability_export_ping_metrics
-        .then(|| metrics_sender.clone());
+        .then(|| metrics_tx.clone());
 
     let baseline_tx = config
         .observability_export_baseline_metrics
-        .then(|| metrics_sender.clone());
+        .then(|| metrics_tx.clone());
 
     let event_tx = config
         .observability_export_events
-        .then(|| metrics_sender.clone());
+        .then(|| metrics_tx.clone());
 
-    let (mut pinger_receiver, mut pinger_sender) = match config.measurement_type {
+    let (mut ping_listener, mut ping_sender) = match config.measurement_type {
         MeasurementType::Icmp => (
             Box::new(PingerICMPEchoListener {}) as Box<dyn PingListener + Send>,
             Box::new(PingerICMPEchoSender {}) as Box<dyn PingSender + Send>,
@@ -114,10 +114,10 @@ fn main() -> anyhow::Result<()> {
         config: config.clone(),
         owd_baseline: owd_baseline.clone(),
         owd_recent: owd_recent.clone(),
-        reselect_trigger: reselect_sender.clone(),
+        reselect_trigger: reselect_tx.clone(),
         start_time: start_t,
-        stats_receiver: baseliner_stats_receiver,
-        metrics_sender: metrics_sender.clone(),
+        stats_rx: baseliner_stats_rx,
+        metrics_tx: metrics_tx.clone(),
     };
 
     let down_qdisc = Netlink::qdisc_from_ifname(config.download_interface.as_str())?;
@@ -146,17 +146,17 @@ fn main() -> anyhow::Result<()> {
     let (error_tx, error_rx) = channel::<anyhow::Error>();
 
     let err_tx = error_tx.clone();
-    let metrics_sender_clone = metrics_sender.clone();
+    let metrics_tx_clone = metrics_tx.clone();
     let reflector_peers_lock_clone = reflector_peers_lock.clone();
     thread::Builder::new()
         .name("receiver".to_string())
         .spawn(move || {
-            if let Err(e) = pinger_receiver.listen(
+            if let Err(e) = ping_listener.listen(
                 id,
                 config.measurement_type,
                 reflector_peers_lock_clone,
-                baseliner_stats_sender,
-                metrics_sender_clone,
+                baseliner_stats_tx,
+                metrics_tx_clone,
             ) {
                 let _ = err_tx.send(e);
             }
@@ -176,7 +176,7 @@ fn main() -> anyhow::Result<()> {
     thread::Builder::new()
         .name("sender".to_string())
         .spawn(move || {
-            if let Err(e) = pinger_sender.send(
+            if let Err(e) = ping_sender.send(
                 id,
                 config.measurement_type,
                 reflector_peers_lock_clone,
@@ -192,8 +192,8 @@ fn main() -> anyhow::Result<()> {
             owd_recent: owd_recent.clone(),
             reflector_peers_lock: reflector_peers_lock.clone(),
             reflector_pool,
-            trigger_channel: reselect_receiver,
-            metrics_sender: metrics_sender.clone(),
+            trigger_channel: reselect_rx,
+            metrics_tx: metrics_tx.clone(),
         };
         let err_tx = error_tx.clone();
         thread::Builder::new()
@@ -228,10 +228,10 @@ fn main() -> anyhow::Result<()> {
         owd_baseline,
         owd_recent,
         reflector_peers_lock,
-        reselect_sender,
+        reselect_tx,
         dl_direction,
         ul_direction,
-        metrics_sender.clone(),
+        metrics_tx.clone(),
     )?;
 
     debug!(
@@ -255,7 +255,7 @@ fn main() -> anyhow::Result<()> {
 
     let metrics = Metrics {
         config,
-        metrics_receiver,
+        metrics_rx,
     };
 
     let err_tx = error_tx.clone();

src/metrics.rs

@@ -120,7 +120,7 @@ pub enum Metric {
 
 pub struct Metrics {
     pub config: Config,
-    pub metrics_receiver: Receiver<Metric>,
+    pub metrics_rx: Receiver<Metric>,
 }
 
 impl Metrics {
@@ -153,7 +153,7 @@ impl Metrics {
 
         loop {
             // block until first metric or timeout
-            match self.metrics_receiver.recv_timeout(timeout) {
+            match self.metrics_rx.recv_timeout(timeout) {
                 Ok(metric) => batch.push(metric),
                 Err(RecvTimeoutError::Timeout) => {
                     if !batch.is_empty() {
@@ -167,7 +167,7 @@ impl Metrics {
 
             // greedily drain without blocking
             loop {
-                match self.metrics_receiver.try_recv() {
+                match self.metrics_rx.try_recv() {
                     Ok(metric) => batch.push(metric),
                     Err(_) => break,
                 }

src/pinger.rs

@@ -54,8 +54,8 @@ pub trait PingListener {
         id: u16,
         type_: MeasurementType,
         reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
-        stats_sender: Sender<PingReply>,
-        metrics_sender: SyncSender<Metric>,
+        stats_tx: Sender<PingReply>,
+        metrics_tx: SyncSender<Metric>,
     ) -> anyhow::Result<()> {
         let socket = &mut open_socket(type_)?;
 
@@ -84,7 +84,7 @@ pub trait PingListener {
                 }
             };
 
-            let _ = metrics_sender.try_send(Metric::Ping {
+            let _ = metrics_tx.try_send(Metric::Ping {
                 reflector: reply.reflector,
                 measurement_type: reply.measurement_type,
                 rtt: reply.rtt,
@@ -106,7 +106,7 @@ pub trait PingListener {
                 reply.up_time,
                 reply.down_time
             );
-            if let Err(e) = stats_sender.send(reply) {
+            if let Err(e) = stats_tx.send(reply) {
                 warn!("Stats channel closed, stopping listener: {}", e);
                 break Ok(());
             }

src/ratecontroller.rs

@@ -109,7 +109,7 @@ pub struct Ratecontroller {
     state_dl: State,
     state_ul: State,
     up_direction: StatsDirection,
-    metrics_sender: SyncSender<Metric>,
+    metrics_tx: SyncSender<Metric>,
 }
 
 impl Ratecontroller {
@@ -248,7 +248,7 @@ impl Ratecontroller {
         reselect_trigger: Sender<bool>,
         down_direction: StatsDirection,
         up_direction: StatsDirection,
-        metrics_sender: SyncSender<Metric>,
+        metrics_tx: SyncSender<Metric>,
     ) -> anyhow::Result<Self> {
         let dl_qdisc = Netlink::qdisc_from_ifname(config.download_interface.as_str())?;
         let dl_safe_rates =
@@ -269,7 +269,7 @@ impl Ratecontroller {
             state_dl: State::new(dl_qdisc, cur_rx, dl_safe_rates),
             state_ul: State::new(ul_qdisc, cur_tx, ul_safe_rates),
             up_direction,
-            metrics_sender,
+            metrics_tx,
         })
     }
 

src/reflector_selector.rs

@@ -15,7 +15,7 @@ pub struct ReflectorSelector {
     pub reflector_peers_lock: Arc<RwLock<Vec<IpAddr>>>,
     pub reflector_pool: Vec<IpAddr>,
     pub trigger_channel: Receiver<bool>,
-    pub metrics_sender: SyncSender<Metric>,
+    pub metrics_tx: SyncSender<Metric>,
 }
 
 impl ReflectorSelector {