Add user-data to if task.

Author: dvc94ch

transports/tcp/src/if_task.rs

@@ -13,17 +13,17 @@ pub enum IfTaskEvent {
 }
 
 #[derive(Debug)]
-enum Command {
-    RegisterListener(SocketAddr, mpsc::UnboundedSender<IfTaskEvent>),
-    PortReuseSocket(SocketAddr, oneshot::Sender<Option<SocketAddr>>),
+enum Command<T> {
+    RegisterListener(SocketAddr, T, mpsc::UnboundedSender<IfTaskEvent>),
+    PortReuseSocket(SocketAddr, oneshot::Sender<Option<T>>),
 }
 
 #[derive(Clone, Debug)]
-pub struct IfTaskHandle {
-    tx: mpsc::UnboundedSender<Command>,
+pub struct IfTaskHandle<T> {
+    tx: mpsc::UnboundedSender<Command<T>>,
 }
 
-impl IfTaskHandle {
+impl<T: Clone + Send + 'static> IfTaskHandle<T> {
     pub async fn new() -> io::Result<Self> {
         // channel is unbounded so that `register_listener` doesn't need to be async.
         let (tx, rx) = mpsc::unbounded();
@@ -40,14 +40,15 @@ impl IfTaskHandle {
     pub fn register_listener(
         &self,
         socket_addr: &SocketAddr,
+        user_data: T,
     ) -> mpsc::UnboundedReceiver<IfTaskEvent> {
         let (tx, rx) = mpsc::unbounded();
-        let cmd = Command::RegisterListener(*socket_addr, tx);
+        let cmd = Command::RegisterListener(*socket_addr, user_data, tx);
         self.tx.unbounded_send(cmd).expect("task paniced");
         rx
     }
 
-    pub async fn port_reuse_socket(&self, socket_addr: &SocketAddr) -> Option<SocketAddr> {
+    pub async fn port_reuse_socket(&self, socket_addr: &SocketAddr) -> Option<T> {
         let (tx, rx) = oneshot::channel();
         let cmd = Command::PortReuseSocket(*socket_addr, tx);
         self.tx.unbounded_send(cmd).expect("task paniced");
@@ -56,17 +57,17 @@ impl IfTaskHandle {
 }
 
 #[derive(Debug)]
-struct Task {
+struct Task<T> {
     /// Command receiver.
-    rx: mpsc::UnboundedReceiver<Command>,
+    rx: mpsc::UnboundedReceiver<Command<T>>,
     /// Interface watcher.
     watcher: IfWatcher,
     /// Listening addresses.
-    listening_addrs: Vec<(SocketAddr, mpsc::UnboundedSender<IfTaskEvent>)>,
+    listening_addrs: Vec<(SocketAddr, T, mpsc::UnboundedSender<IfTaskEvent>)>,
 }
 
-impl Task {
-    async fn new(rx: mpsc::UnboundedReceiver<Command>) -> io::Result<Self> {
+impl<T: Clone + Send + 'static> Task<T> {
+    async fn new(rx: mpsc::UnboundedReceiver<Command<T>>) -> io::Result<Self> {
         Ok(Self {
             rx,
             watcher: IfWatcher::new().await?,
@@ -86,7 +87,7 @@ impl Task {
                             continue;
                         }
                     };
-                    self.listening_addrs.retain(|(addr, tx)| {
+                    self.listening_addrs.retain(|(addr, _, tx)| {
                         match event {
                             IfEvent::Up(inet) => {
                                 if let Some(addr) = iface_match(&inet, &addr) {
@@ -107,20 +108,20 @@ impl Task {
                 }
                 cmd = rx.next() => {
                     match cmd {
-                        Some(Command::RegisterListener(socket_addr, tx)) => {
+                        Some(Command::RegisterListener(socket_addr, user_data, tx)) => {
                             for iface in self.watcher.iter() {
                                 if let Some(addr) = iface_match(iface, &socket_addr) {
                                     tx.unbounded_send(IfTaskEvent::NewAddress(addr)).ok();
                                 }
                             }
-                            self.listening_addrs.push((socket_addr, tx));
+                            self.listening_addrs.push((socket_addr, user_data, tx));
                         }
                         Some(Command::PortReuseSocket(socket_addr, tx)) => {
                             let mut reuse_socket = None;
-                            for (addr, _) in &self.listening_addrs {
+                            for (addr, user_data, _) in &self.listening_addrs {
                                 if addr.ip().is_ipv4() == socket_addr.ip().is_ipv4()
                                     && addr.ip().is_loopback() == socket_addr.ip().is_loopback() {
-                                    reuse_socket = Some(*addr);
+                                    reuse_socket = Some(user_data.clone());
                                     break;
                                 }
                             }

transports/tcp/src/lib.rs

@@ -69,7 +69,7 @@ pub struct TcpConfig {
     /// Port reuse.
     port_reuse: bool,
     /// The task handle.
-    handle: IfTaskHandle,
+    handle: IfTaskHandle<SocketAddr>,
 }
 
 impl TcpConfig {
@@ -140,7 +140,7 @@ impl TcpConfig {
         let listener = Async::new(socket.into_tcp_listener())?;
         let socket_addr = listener.get_ref().local_addr()?;
 
-        let rx = self.handle.register_listener(&socket_addr);
+        let rx = self.handle.register_listener(&socket_addr, socket_addr);
         let listen_stream = TcpListenStream {
             config: self,
             listener,