refactor: use tokio mpsc for controller/selector channel

src/controller.rs

@@ -5,8 +5,9 @@ use std::{
     time::Duration,
 };
 
-use flume::{Receiver, SendError, Sender};
+use flume::{SendError, Sender};
 use thiserror::Error;
+use tokio::sync::mpsc::UnboundedReceiver;
 
 #[cfg(feature = "irtt")]
 use crate::measurement::irtt::IrttBinding;
@@ -28,7 +29,7 @@ pub enum ControllerError {
 }
 
 pub struct Controller {
-    reflector_rx: Receiver<Vec<SocketAddr>>,
+    reflector_rx: UnboundedReceiver<Vec<SocketAddr>>,
     icmp_binding_tx: Sender<Vec<IcmpBinding>>,
     #[cfg(feature = "irtt")]
     irtt_binding_tx: Sender<Vec<IrttBinding>>,
@@ -43,7 +44,7 @@ pub struct Controller {
 
 impl Controller {
     pub fn new(
-        reflector_rx: Receiver<Vec<SocketAddr>>,
+        reflector_rx: UnboundedReceiver<Vec<SocketAddr>>,
         icmp_binding_tx: Sender<Vec<IcmpBinding>>,
         #[cfg(feature = "irtt")] irtt_binding_tx: Sender<Vec<IrttBinding>>,
         reflectors: HashMap<SocketAddr, ReflectorDefinition>,
@@ -69,7 +70,7 @@ impl Controller {
 
     pub async fn run(mut self, initial_reflectors: Vec<SocketAddr>) -> Result<(), ControllerError> {
         self.update_reflectors(initial_reflectors).await?;
-        while let Ok(reflectors) = self.reflector_rx.recv_async().await {
+        while let Some(reflectors) = self.reflector_rx.recv().await {
             self.update_reflectors(reflectors).await?;
         }
         Ok(())

src/main.rs

@@ -352,7 +352,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
     #[cfg(feature = "irtt")]
     let (irtt_binding_tx, irtt_binding_rx) = flume::unbounded();
     let (measurement_tx, measurement_rx) = tokio::sync::mpsc::unbounded_channel();
-    let (reflector_tx, reflector_rx) = flume::unbounded();
+    let (reflector_tx, reflector_rx) = tokio::sync::mpsc::unbounded_channel();
     let (reselect_tx, reselect_rx) = flume::bounded(1);
     let (selection_snapshot_tx, selection_snapshot_rx) = if reselection_enabled {
         let (tx, rx) = flume::unbounded();

src/reflector_selector.rs

@@ -11,13 +11,14 @@ use crate::measurement::model::MeasurementStream;
 use crate::metrics::{Metric, MetricsSender};
 use crate::settings::Settings;
 use crate::util::RwLockExt;
-use flume::{Receiver, RecvError, RecvTimeoutError, Selector, Sender};
+use flume::{Receiver, RecvError, RecvTimeoutError, Selector};
 use log::{debug, info};
 use std::net::SocketAddr;
 use std::sync::atomic::Ordering;
 use std::sync::{Arc, RwLock};
 use std::thread::sleep;
 use std::time::{Duration, Instant};
+use tokio::sync::mpsc::UnboundedSender;
 
 enum SelectorEvent {
     Trigger(Result<bool, RecvError>),
@@ -115,7 +116,7 @@ pub struct ReflectorSelector {
     pub settings: Settings,
     pub snapshot_rx: Receiver<ControlSnapshot>,
     pub reflector_peers_lock: Arc<RwLock<Vec<SocketAddr>>>,
-    pub reflector_tx: Sender<Vec<SocketAddr>>,
+    pub reflector_tx: UnboundedSender<Vec<SocketAddr>>,
     pub reflector_pool: Vec<SocketAddr>,
     pub endpoint_streams: Arc<RwLock<std::collections::HashMap<SocketAddr, MeasurementStream>>>,
     pub trigger_channel: Receiver<bool>,