diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index c73aade059..54309f02c5 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -86,12 +86,8 @@ impl MixNode { fn start_packet_forwarder(&mut self) -> mpsc::UnboundedSender<(SocketAddr, Vec)> { info!("Starting packet forwarder..."); - - // this can later be replaced with topology information - let initial_addresses = vec![]; self.runtime .block_on(packet_forwarding::PacketForwarder::new( - initial_addresses, self.config.get_packet_forwarding_initial_backoff(), self.config.get_packet_forwarding_maximum_backoff(), )) diff --git a/mixnode/src/node/packet_forwarding.rs b/mixnode/src/node/packet_forwarding.rs index 8335322081..6fb507225c 100644 --- a/mixnode/src/node/packet_forwarding.rs +++ b/mixnode/src/node/packet_forwarding.rs @@ -1,24 +1,21 @@ use futures::channel::mpsc; use futures::StreamExt; -use log::*; use std::net::SocketAddr; use std::time::Duration; use tokio::runtime::Handle; -pub(crate) struct PacketForwarder<'a> { - tcp_client: multi_tcp_client::Client<'a>, +pub(crate) struct PacketForwarder { + tcp_client: multi_tcp_client::Client, conn_tx: mpsc::UnboundedSender<(SocketAddr, Vec)>, conn_rx: mpsc::UnboundedReceiver<(SocketAddr, Vec)>, } -impl PacketForwarder<'static> { +impl PacketForwarder { pub(crate) async fn new( - initial_endpoints: Vec, initial_reconnection_backoff: Duration, maximum_reconnection_backoff: Duration, - ) -> PacketForwarder<'static> { + ) -> PacketForwarder { let tcp_client_config = multi_tcp_client::Config::new( - initial_endpoints, initial_reconnection_backoff, maximum_reconnection_backoff, ); @@ -26,7 +23,7 @@ impl PacketForwarder<'static> { let (conn_tx, conn_rx) = mpsc::unbounded(); PacketForwarder { - tcp_client: multi_tcp_client::Client::new(tcp_client_config).await, + tcp_client: multi_tcp_client::Client::start_new(tcp_client_config).await, conn_tx, conn_rx, } @@ -37,10 +34,7 @@ impl PacketForwarder<'static> { let sender_channel = self.conn_tx.clone(); handle.spawn(async move { while let Some((address, packet)) = self.conn_rx.next().await { - match self.tcp_client.send(address, &packet).await { - Err(e) => warn!("Failed to forward packet to {:?} - {:?}", address, e), - Ok(_) => trace!("Forwarded packet to {:?}", address), - } + self.tcp_client.send(address, packet).await; } }); sender_channel diff --git a/nym-client/src/client/mix_traffic.rs b/nym-client/src/client/mix_traffic.rs index c71a7c3be6..1b27bb7d17 100644 --- a/nym-client/src/client/mix_traffic.rs +++ b/nym-client/src/client/mix_traffic.rs @@ -18,45 +18,33 @@ impl MixMessage { } // TODO: put our TCP client here -pub(crate) struct MixTrafficController<'a> { - tcp_client: multi_tcp_client::Client<'a>, +pub(crate) struct MixTrafficController { + tcp_client: multi_tcp_client::Client, mix_rx: MixMessageReceiver, } -impl MixTrafficController<'static> { +impl MixTrafficController { pub(crate) async fn new( - initial_endpoints: Vec, initial_reconnection_backoff: Duration, maximum_reconnection_backoff: Duration, mix_rx: MixMessageReceiver, ) -> Self { let tcp_client_config = multi_tcp_client::Config::new( - initial_endpoints, initial_reconnection_backoff, maximum_reconnection_backoff, ); MixTrafficController { - tcp_client: multi_tcp_client::Client::new(tcp_client_config).await, + tcp_client: multi_tcp_client::Client::start_new(tcp_client_config).await, mix_rx, } } async fn on_message(&mut self, mix_message: MixMessage) { debug!("Got a mix_message for {:?}", mix_message.0); - match self - .tcp_client - .send(mix_message.0, &mix_message.1.to_bytes()) - .await - { - Ok(_) => trace!("sent a mix message"), - // TODO: should there be some kind of threshold of failed messages - // that if reached, the application blows? - Err(e) => error!( - "We failed to send the packet to {} - {:?}", - mix_message.0, e - ), - }; + self.tcp_client + .send(mix_message.0, mix_message.1.to_bytes()) + .await; } pub(crate) async fn run(&mut self) { diff --git a/nym-client/src/client/mod.rs b/nym-client/src/client/mod.rs index d80f512850..dd889faf54 100644 --- a/nym-client/src/client/mod.rs +++ b/nym-client/src/client/mod.rs @@ -212,11 +212,8 @@ impl NymClient { // controller for sending sphinx packets to mixnet (either real traffic or cover traffic) fn start_mix_traffic_controller(&mut self, mix_rx: MixMessageReceiver) { info!("Starting mix trafic controller..."); - // TODO: possible optimisation: set the initial endpoints to all known mixes from layer 1 - let initial_mix_endpoints = Vec::new(); self.runtime .block_on(MixTrafficController::new( - initial_mix_endpoints, self.config.get_packet_forwarding_initial_backoff(), self.config.get_packet_forwarding_maximum_backoff(), mix_rx,