refactor: use tokio watch channel for bindings

src/controller.rs

@@ -5,9 +5,11 @@ use std::{
     time::Duration,
 };
 
-use flume::{SendError, Sender};
 use thiserror::Error;
-use tokio::sync::mpsc::UnboundedReceiver;
+use tokio::sync::{
+    mpsc::UnboundedReceiver,
+    watch::{self, error::SendError},
+};
 
 #[cfg(feature = "irtt")]
 use crate::measurement::irtt::IrttBinding;
@@ -30,9 +32,9 @@ pub enum ControllerError {
 
 pub struct Controller {
     reflector_rx: UnboundedReceiver<Vec<SocketAddr>>,
-    icmp_binding_tx: Sender<Vec<IcmpBinding>>,
+    icmp_binding_tx: watch::Sender<Vec<IcmpBinding>>,
     #[cfg(feature = "irtt")]
-    irtt_binding_tx: Sender<Vec<IrttBinding>>,
+    irtt_binding_tx: watch::Sender<Vec<IrttBinding>>,
     reflectors: HashMap<SocketAddr, ReflectorDefinition>,
     active_streams: Arc<RwLock<Vec<MeasurementStream>>>,
     endpoint_streams: Arc<RwLock<HashMap<SocketAddr, MeasurementStream>>>,
@@ -45,8 +47,8 @@ pub struct Controller {
 impl Controller {
     pub fn new(
         reflector_rx: UnboundedReceiver<Vec<SocketAddr>>,
-        icmp_binding_tx: Sender<Vec<IcmpBinding>>,
+        icmp_binding_tx: watch::Sender<Vec<IcmpBinding>>,
-        #[cfg(feature = "irtt")] irtt_binding_tx: Sender<Vec<IrttBinding>>,
+        #[cfg(feature = "irtt")] irtt_binding_tx: watch::Sender<Vec<IrttBinding>>,
         reflectors: HashMap<SocketAddr, ReflectorDefinition>,
         active_streams: Arc<RwLock<Vec<MeasurementStream>>>,
         endpoint_streams: Arc<RwLock<HashMap<SocketAddr, MeasurementStream>>>,
@@ -137,9 +139,9 @@ impl Controller {
             .endpoint_streams
             .write()
             .expect("endpoint stream lock poisoned") = endpoint_streams;
-        self.icmp_binding_tx.send_async(icmp).await?;
+        self.icmp_binding_tx.send(icmp)?;
         #[cfg(feature = "irtt")]
-        self.irtt_binding_tx.send_async(irtt).await?;
+        self.irtt_binding_tx.send(irtt)?;
         Ok(())
     }
 }

src/main.rs

@@ -348,9 +348,9 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
 
     let (control_snapshot_tx, control_snapshot_rx) = watch::channel(None);
     let (error_tx, mut error_rx) = tokio::sync::mpsc::unbounded_channel::<anyhow::Error>();
-    let (icmp_binding_tx, icmp_binding_rx) = flume::unbounded();
+    let (icmp_binding_tx, icmp_binding_rx) = watch::channel(Vec::new());
     #[cfg(feature = "irtt")]
-    let (irtt_binding_tx, irtt_binding_rx) = flume::unbounded();
+    let (irtt_binding_tx, irtt_binding_rx) = watch::channel(Vec::new());
     let (measurement_tx, measurement_rx) = tokio::sync::mpsc::unbounded_channel();
     let (reflector_tx, reflector_rx) = tokio::sync::mpsc::unbounded_channel();
     let (reselect_tx, reselect_rx) = flume::bounded(1);

src/measurement/icmp.rs

@@ -5,7 +5,6 @@ use std::{
     time::{Duration, Instant, SystemTime},
 };
 
-use flume::Receiver;
 use icmp_socket2::{
     AsyncIcmpSocket, IcmpSocket4, Icmpv4Message, Icmpv4Packet,
     packet::{IcmpPacketBuildError, WithEchoRequest, WithTimestampRequest},
@@ -13,7 +12,7 @@ use icmp_socket2::{
 };
 use log::debug;
 use thiserror::Error;
-use tokio::sync::mpsc::UnboundedSender;
+use tokio::sync::{mpsc::UnboundedSender, watch};
 
 use crate::{
     measurement::model::{
@@ -25,6 +24,7 @@ use crate::{
     time::{ClockSample, Realtime},
 };
 
+#[derive(Clone)]
 pub struct IcmpBinding {
     pub stream: MeasurementStream,
     pub schedule: ProbeSchedule,
@@ -104,7 +104,7 @@ pub struct IcmpEngine {
 
     origin: Instant,
 
-    binding_rx: Receiver<Vec<IcmpBinding>>,
+    binding_rx: watch::Receiver<Vec<IcmpBinding>>,
     observation_tx: UnboundedSender<MeasurementEvent>,
 }
 
@@ -125,7 +125,7 @@ impl IcmpEngine {
         origin: Instant,
         #[allow(unused)] settings: &NetworkSettings,
         observation_tx: UnboundedSender<MeasurementEvent>,
-        binding_rx: Receiver<Vec<IcmpBinding>>,
+        binding_rx: watch::Receiver<Vec<IcmpBinding>>,
     ) -> Result<Self, IcmpError> {
         #[allow(unused_mut)]
         let mut socket = IcmpSocket4::new()?;
@@ -446,10 +446,10 @@ impl IcmpEngine {
             let deadline = self.next_deadline();
 
             tokio::select! {
-                bindings = self.binding_rx.recv_async() => {
+                bindings = self.binding_rx.changed() => {
-                    let bindings = bindings
+                    bindings.map_err(|_| IcmpError::BindingChannelClosed)?;
-                        .map_err(|_| IcmpError::BindingChannelClosed)?;
 
+                    let bindings = self.binding_rx.borrow_and_update().clone();
                     self.set_bindings(bindings)?;
                 }
 

src/measurement/irtt.rs

@@ -1,6 +1,5 @@
 use std::{collections::HashMap, net::SocketAddr};
 
-use flume::Receiver;
 use irtt_client::{
     ClientConfig, ClientEvent, NegotiationPolicy, OpenPolicy,
     managed::{
@@ -12,7 +11,7 @@ use irtt_client::{
 };
 use log::{debug, trace, warn};
 use thiserror::Error;
-use tokio::sync::{broadcast::error::RecvError, mpsc::UnboundedSender};
+use tokio::sync::{broadcast::error::RecvError, mpsc::UnboundedSender, watch};
 
 use crate::SHUTDOWN;
 use crate::measurement::model::{
@@ -51,6 +50,7 @@ pub enum IrttError {
     ManagedAbandoned,
 }
 
+#[derive(Clone)]
 pub struct IrttBinding {
     stream: MeasurementStream,
     port: u16,
@@ -69,7 +69,7 @@ pub struct IrttEngine {
     bindings: HashMap<MeasurementStreamId, IrttBinding>,
     targets: Vec<ManagedTargetConfig>,
 
-    binding_rx: Receiver<Vec<IrttBinding>>,
+    binding_rx: watch::Receiver<Vec<IrttBinding>>,
     observation_tx: UnboundedSender<MeasurementEvent>,
 }
 
@@ -78,7 +78,7 @@ impl IrttEngine {
         interval: std::time::Duration,
         settings: &NetworkSettings,
         observation_tx: UnboundedSender<MeasurementEvent>,
-        binding_rx: Receiver<Vec<IrttBinding>>,
+        binding_rx: watch::Receiver<Vec<IrttBinding>>,
     ) -> Result<(ManagedClientTask, Self), IrttError> {
         let config = managed_client_config(interval, settings);
         let (task, handle) = ManagedClient::task(config, Vec::new())?;
@@ -158,10 +158,10 @@ impl IrttEngine {
         retry.tick().await;
         loop {
             tokio::select! {
-                bindings = self.binding_rx.recv_async() => {
+                bindings = self.binding_rx.changed() => {
-                    let bindings = bindings
+                    bindings.map_err(|_| IrttError::BindingChannelClosed)?;
-                        .map_err(|_| IrttError::BindingChannelClosed)?;
 
+                    let bindings = self.binding_rx.borrow_and_update().clone();
                     self.set_bindings(bindings).await?;
                 }