From 1418d068b080c7d02e5eb314d7d6a2b80c55fe7e Mon Sep 17 00:00:00 2001
From: Jedrzej Stuczynski <jedrzej.stuczynski@gmail.com>
Date: Thu, 27 Feb 2020 12:18:50 +0000
Subject: [PATCH] Mixnode packet forwarder

---
 mixnode/Cargo.toml                    |  1 +
 mixnode/src/node/packet_forwarding.rs | 48 +++++++++++++++++++++++++++
 2 files changed, 49 insertions(+)
 create mode 100644 mixnode/src/node/packet_forwarding.rs

diff --git a/mixnode/Cargo.toml b/mixnode/Cargo.toml
index b58ee4a113..ec249f8b62 100644
--- a/mixnode/Cargo.toml
+++ b/mixnode/Cargo.toml
@@ -24,6 +24,7 @@ addressing = {path = "../common/addressing" }
 config = {path = "../common/config"}
 crypto = {path = "../common/crypto"}
 directory-client = { path = "../common/clients/directory-client" }
+multi-tcp-client = { path = "../common/clients/multi-tcp-client" }
 pemstore = {path = "../common/pemstore"}
 
 ## will be moved to proper dependencies once released
diff --git a/mixnode/src/node/packet_forwarding.rs b/mixnode/src/node/packet_forwarding.rs
new file mode 100644
index 0000000000..15707a7b43
--- /dev/null
+++ b/mixnode/src/node/packet_forwarding.rs
@@ -0,0 +1,48 @@
+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>,
+    conn_tx: mpsc::UnboundedSender<(SocketAddr, Vec<u8>)>,
+    conn_rx: mpsc::UnboundedReceiver<(SocketAddr, Vec<u8>)>,
+}
+
+impl<'a: 'static> PacketForwarder<'a> {
+    pub(crate) async fn new(
+        initial_endpoints: Vec<SocketAddr>,
+        reconnection_backoff: Duration,
+        maximum_reconnection_backoff: Duration,
+    ) -> PacketForwarder<'a> {
+        let tcp_client_config = multi_tcp_client::Config::new(
+            initial_endpoints,
+            reconnection_backoff,
+            maximum_reconnection_backoff,
+        );
+
+        let (conn_tx, conn_rx) = mpsc::unbounded();
+
+        PacketForwarder {
+            tcp_client: multi_tcp_client::Client::new(tcp_client_config).await,
+            conn_tx,
+            conn_rx,
+        }
+    }
+
+    pub(crate) fn start(mut self, handle: &Handle) -> mpsc::UnboundedSender<(SocketAddr, Vec<u8>)> {
+        // TODO: what to do with the lost JoinHandle?
+        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),
+                }
+            }
+        });
+        sender_channel
+    }
+}