swarm/: include ListenerId in SwarmEvents (#2123)

Co-authored-by: Max Inden <mail@max-inden.de>

examples/chat-tokio.rs

@@ -156,8 +156,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
                 swarm.behaviour_mut().floodsub.publish(floodsub_topic.clone(), line.as_bytes());
             }
             event = swarm.select_next_some() => {
-                if let SwarmEvent::NewListenAddr(addr) = event {
+                if let SwarmEvent::NewListenAddr { address, .. } = event {
-                    println!("Listening on {:?}", addr);
+                    println!("Listening on {:?}", address);
                 }
             }
         }

examples/chat.rs

@@ -160,8 +160,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
         loop {
             match swarm.poll_next_unpin(cx) {
                 Poll::Ready(Some(event)) => {
-                    if let SwarmEvent::NewListenAddr(addr) = event {
+                    if let SwarmEvent::NewListenAddr { address, .. } = event {
-                        println!("Listening on {:?}", addr);
+                        println!("Listening on {:?}", address);
                     }
                 }
                 Poll::Ready(None) => return Poll::Ready(Ok(())),

examples/distributed-key-value-store.rs

@@ -175,8 +175,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
         loop {
             match swarm.poll_next_unpin(cx) {
                 Poll::Ready(Some(event)) => {
-                    if let SwarmEvent::NewListenAddr(addr) = event {
+                    if let SwarmEvent::NewListenAddr { address, .. } = event {
-                        println!("Listening on {:?}", addr);
+                        println!("Listening on {:?}", address);
                     }
                 }
                 Poll::Ready(None) => return Poll::Ready(Ok(())),

examples/gossipsub-chat.rs

@@ -162,8 +162,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
                         id,
                         peer_id
                     ),
-                    SwarmEvent::NewListenAddr(addr) => {
+                    SwarmEvent::NewListenAddr { address, .. } => {
-                        println!("Listening on {:?}", addr);
+                        println!("Listening on {:?}", address);
                     }
                     _ => {}
                 },

examples/ipfs-private.rs

@@ -287,8 +287,8 @@ fn main() -> Result<(), Box<dyn Error>> {
         loop {
             match swarm.poll_next_unpin(cx) {
                 Poll::Ready(Some(event)) => {
-                    if let SwarmEvent::NewListenAddr(addr) = event {
+                    if let SwarmEvent::NewListenAddr { address, .. } = event {
-                        println!("Listening on {:?}", addr);
+                        println!("Listening on {:?}", address);
                     }
                 }
                 Poll::Ready(None) => return Poll::Ready(Ok(())),

examples/ping.rs

@@ -79,7 +79,7 @@ fn main() -> Result<(), Box<dyn Error>> {
     block_on(future::poll_fn(move |cx| loop {
         match swarm.poll_next_unpin(cx) {
             Poll::Ready(Some(event)) => match event {
-                SwarmEvent::NewListenAddr(addr) => println!("Listening on {:?}", addr),
+                SwarmEvent::NewListenAddr{ address, .. } => println!("Listening on {:?}", address),
                 SwarmEvent::Behaviour(event) => println!("{:?}", event),
                 _ => {}
             },

protocols/identify/src/identify.rs

@@ -498,7 +498,7 @@ mod tests {
                 let swarm1_fut = swarm1.select_next_some();
                 pin_mut!(swarm1_fut);
                 match swarm1_fut.await {
-                    SwarmEvent::NewListenAddr(addr) => return addr,
+                    SwarmEvent::NewListenAddr { address, .. } => return address,
                     _ => {}
                 }
             }
@@ -577,7 +577,7 @@ mod tests {
                 let swarm1_fut = swarm1.select_next_some();
                 pin_mut!(swarm1_fut);
                 match swarm1_fut.await {
-                    SwarmEvent::NewListenAddr(addr) => return addr,
+                    SwarmEvent::NewListenAddr { address, .. } => return address,
                     _ => {}
                 }
             }

protocols/ping/tests/ping.rs

@@ -64,7 +64,7 @@ fn ping_pong() {
         let peer1 = async move {
             loop {
                 match swarm1.select_next_some().await {
-                    SwarmEvent::NewListenAddr(listener) => tx.send(listener).await.unwrap(),
+                    SwarmEvent::NewListenAddr { address, .. } => tx.send(address).await.unwrap(),
                     SwarmEvent::Behaviour(PingEvent { peer, result: Ok(PingSuccess::Ping { rtt }) }) => {
                         count1 -= 1;
                         if count1 == 0 {
@@ -137,7 +137,7 @@ fn max_failures() {
 
             loop {
                 match swarm1.select_next_some().await {
-                    SwarmEvent::NewListenAddr(listener) => tx.send(listener).await.unwrap(),
+                    SwarmEvent::NewListenAddr { address, .. } => tx.send(address).await.unwrap(),
                     SwarmEvent::Behaviour(PingEvent {
                         result: Ok(PingSuccess::Ping { .. }), ..
                     }) => {

protocols/relay/examples/relay.rs

@@ -140,8 +140,8 @@ fn main() -> Result<(), Box<dyn Error>> {
         loop {
             match swarm.poll_next_unpin(cx) {
                 Poll::Ready(Some(event)) => match event {
-                    SwarmEvent::NewListenAddr(addr) => {
+                    SwarmEvent::NewListenAddr { address, .. } => {
-                        print_listener_peer(&addr, &opt.mode, local_peer_id)
+                        print_listener_peer(&address, &opt.mode, local_peer_id)
                     }
                     _ => println!("{:?}", event),
                 },

protocols/relay/tests/lib.rs

@@ -73,7 +73,9 @@ fn src_connect_to_dst_listening_via_relay() {
     relay_swarm.listen_on(relay_addr.clone()).unwrap();
     spawn_swarm_on_pool(&pool, relay_swarm);
 
-    dst_swarm.listen_on(dst_listen_addr_via_relay.clone()).unwrap();
+    let dst_listener = dst_swarm
+        .listen_on(dst_listen_addr_via_relay.clone())
+        .unwrap();
 
     pool.run_until(async {
         // Destination Node dialing Relay.
@@ -93,7 +95,13 @@ fn src_connect_to_dst_listening_via_relay() {
         // Destination Node reporting listen address via relay.
         loop {
             match dst_swarm.select_next_some().await {
-                SwarmEvent::NewListenAddr(addr) if addr == dst_listen_addr_via_relay => break,
+                SwarmEvent::NewListenAddr {
+                    address,
+                    listener_id,
+                } if listener_id == dst_listener => {
+                    assert_eq!(address, dst_listen_addr_via_relay);
+                    break;
+                }
                 SwarmEvent::Behaviour(CombinedEvent::Ping(_)) => {}
                 SwarmEvent::Behaviour(CombinedEvent::Kad(KademliaEvent::RoutingUpdated {
                     ..
@@ -282,14 +290,20 @@ fn src_connect_to_dst_via_established_connection_to_relay() {
     relay_swarm.listen_on(relay_addr.clone()).unwrap();
     spawn_swarm_on_pool(&pool, relay_swarm);
 
-    dst_swarm.listen_on(dst_addr_via_relay.clone()).unwrap();
+    let dst_listener = dst_swarm.listen_on(dst_addr_via_relay.clone()).unwrap();
     // Wait for destination to listen via relay.
     pool.run_until(async {
         loop {
             match dst_swarm.select_next_some().await {
                 SwarmEvent::Dialing(_) => {}
                 SwarmEvent::ConnectionEstablished { .. } => {}
-                SwarmEvent::NewListenAddr(addr) if addr == dst_addr_via_relay => break,
+                SwarmEvent::NewListenAddr {
+                    address,
+                    listener_id,
+                } if listener_id == dst_listener => {
+                    assert_eq!(address, dst_addr_via_relay);
+                    break;
+                }
                 e => panic!("{:?}", e),
             }
         }
@@ -531,7 +545,7 @@ fn firewalled_src_discover_firewalled_dst_via_kad_and_connect_to_dst_via_routabl
     spawn_swarm_on_pool(&pool, relay_swarm);
 
     // Destination Node listen via Relay.
-    dst_swarm.listen_on(dst_addr_via_relay.clone()).unwrap();
+    let dst_listener = dst_swarm.listen_on(dst_addr_via_relay.clone()).unwrap();
 
     pool.run_until(async {
         // Destination Node dialing Relay.
@@ -556,7 +570,13 @@ fn firewalled_src_discover_firewalled_dst_via_kad_and_connect_to_dst_via_routabl
         // Destination Node reporting listen address via relay.
         loop {
             match dst_swarm.select_next_some().await {
-                SwarmEvent::NewListenAddr(addr) if addr == dst_addr_via_relay => break,
+                SwarmEvent::NewListenAddr {
+                    address,
+                    listener_id,
+                } if listener_id == dst_listener => {
+                    assert_eq!(address, dst_addr_via_relay);
+                    break;
+                }
                 SwarmEvent::Behaviour(CombinedEvent::Ping(_)) => {}
                 SwarmEvent::Behaviour(CombinedEvent::Kad(KademliaEvent::RoutingUpdated {
                     ..
@@ -715,14 +735,20 @@ fn inactive_connection_timeout() {
     relay_swarm.listen_on(relay_addr.clone()).unwrap();
     spawn_swarm_on_pool(&pool, relay_swarm);
 
-    dst_swarm.listen_on(dst_addr_via_relay.clone()).unwrap();
+    let new_listener = dst_swarm.listen_on(dst_addr_via_relay.clone()).unwrap();
     // Wait for destination to listen via relay.
     pool.run_until(async {
         loop {
             match dst_swarm.select_next_some().await {
                 SwarmEvent::Dialing(_) => {}
                 SwarmEvent::ConnectionEstablished { .. } => {}
-                SwarmEvent::NewListenAddr(addr) if addr == dst_addr_via_relay => break,
+                SwarmEvent::NewListenAddr {
+                    address,
+                    listener_id,
+                } if listener_id == new_listener => {
+                    assert_eq!(address, dst_addr_via_relay);
+                    break;
+                }
                 e => panic!("{:?}", e),
             }
         }
@@ -789,14 +815,20 @@ fn concurrent_connection_same_relay_same_dst() {
     relay_swarm.listen_on(relay_addr.clone()).unwrap();
     spawn_swarm_on_pool(&pool, relay_swarm);
 
-    dst_swarm.listen_on(dst_addr_via_relay.clone()).unwrap();
+    let dst_listener = dst_swarm.listen_on(dst_addr_via_relay.clone()).unwrap();
     // Wait for destination to listen via relay.
     pool.run_until(async {
         loop {
             match dst_swarm.select_next_some().await {
                 SwarmEvent::Dialing(_) => {}
                 SwarmEvent::ConnectionEstablished { .. } => {}
-                SwarmEvent::NewListenAddr(addr) if addr == dst_addr_via_relay => break,
+                SwarmEvent::NewListenAddr {
+                    address,
+                    listener_id,
+                } if listener_id == dst_listener => {
+                    assert_eq!(address, dst_addr_via_relay);
+                    break;
+                }
                 e => panic!("{:?}", e),
             }
         }
@@ -904,10 +936,10 @@ fn yield_incoming_connection_through_correct_listener() {
     relay_3_swarm.listen_on(relay_3_addr.clone()).unwrap();
     spawn_swarm_on_pool(&pool, relay_3_swarm);
 
-    dst_swarm.listen_on(relay_1_addr_incl_circuit.clone()).unwrap();
+    let dst_listener_via_relay_1 = dst_swarm.listen_on(relay_1_addr_incl_circuit.clone()).unwrap();
-    dst_swarm.listen_on(relay_2_addr_incl_circuit.clone()).unwrap();
+    let dst_listener_via_relay_2 = dst_swarm.listen_on(relay_2_addr_incl_circuit.clone()).unwrap();
     // Listen on own address in order for relay 3 to be able to connect to destination node.
-    dst_swarm.listen_on(dst_addr.clone()).unwrap();
+    let dst_listener = dst_swarm.listen_on(dst_addr.clone()).unwrap();
 
     // Wait for destination node to establish connections to relay 1 and 2.
     pool.run_until(async {
@@ -924,10 +956,18 @@ fn yield_incoming_connection_through_correct_listener() {
                         break;
                     }
                 }
-                SwarmEvent::NewListenAddr(addr)
+                SwarmEvent::NewListenAddr {
-                    if addr == relay_1_addr_incl_circuit
+                    address,
-                        || addr == relay_2_addr_incl_circuit
+                    listener_id,
-                        || addr == dst_addr => {}
+                } if listener_id == dst_listener_via_relay_2 => assert_eq!(address, relay_2_addr_incl_circuit),
+                SwarmEvent::NewListenAddr {
+                    address,
+                    listener_id,
+                } if listener_id == dst_listener_via_relay_1 => assert_eq!(address, relay_1_addr_incl_circuit),
+                SwarmEvent::NewListenAddr {
+                    address,
+                    listener_id,
+                } if listener_id == dst_listener => assert_eq!(address, dst_addr),
                 SwarmEvent::Behaviour(CombinedEvent::Ping(_)) => {}
                 e => panic!("{:?}", e),
             }
@@ -966,10 +1006,10 @@ fn yield_incoming_connection_through_correct_listener() {
                         unreachable!();
                     }
                 }
-                SwarmEvent::NewListenAddr(addr)
+                SwarmEvent::NewListenAddr { address, .. }
-                    if addr == relay_1_addr_incl_circuit
+                    if address == relay_1_addr_incl_circuit
-                        || addr == relay_2_addr_incl_circuit
+                        || address == relay_2_addr_incl_circuit
-                        || addr == dst_addr => {}
+                        || address == dst_addr => {}
                 SwarmEvent::Behaviour(CombinedEvent::Ping(PingEvent {
                     peer,
                     result: Ok(_),
@@ -1044,7 +1084,7 @@ fn yield_incoming_connection_through_correct_listener() {
     pool.run_until(async {
         loop {
             match dst_swarm.select_next_some().await {
-                SwarmEvent::NewListenAddr(addr) if addr == Protocol::P2pCircuit.into() => break,
+                SwarmEvent::NewListenAddr { address, .. } if address == Protocol::P2pCircuit.into() => break,
                 SwarmEvent::Behaviour(CombinedEvent::Ping(_)) => {}
                 SwarmEvent::Behaviour(CombinedEvent::Kad(KademliaEvent::RoutingUpdated {
                     ..

protocols/request-response/tests/ping.rs

@@ -98,7 +98,7 @@ fn ping_protocol() {
     let peer1 = async move {
         loop {
             match swarm1.select_next_some().await {
-                SwarmEvent::NewListenAddr(addr) => tx.send(addr).await.unwrap(),
+                SwarmEvent::NewListenAddr { address, .. }=> tx.send(address).await.unwrap(),
                 SwarmEvent::Behaviour(RequestResponseEvent::Message {
                     peer,
                     message: RequestResponseMessage::Request { request, channel, .. }
@@ -312,7 +312,7 @@ fn ping_protocol_throttled() {
     let peer1 = async move {
         for i in 1 .. {
             match swarm1.select_next_some().await {
-                SwarmEvent::NewListenAddr(addr) => tx.send(addr).await.unwrap(),
+                SwarmEvent::NewListenAddr { address, .. } => tx.send(address).await.unwrap(),
                 SwarmEvent::Behaviour(throttled::Event::Event(RequestResponseEvent::Message {
                     peer,
                     message: RequestResponseMessage::Request { request, channel, .. },

src/tutorial.rs

@@ -298,8 +298,8 @@
 //!     block_on(future::poll_fn(move |cx| loop {
 //!         match swarm.poll_next_unpin(cx) {
 //!             Poll::Ready(Some(event)) => {
-//!                 if let SwarmEvent::NewListenAddr(addr) = event {
+//!                 if let SwarmEvent::NewListenAddr { address, .. }= event {
-//!                     println!("Listening on {:?}", addr);
+//!                     println!("Listening on {:?}", address);
 //!                 }
 //!             },
 //!             Poll::Ready(None) => return Poll::Ready(()),

swarm/CHANGELOG.md

@@ -19,8 +19,13 @@
   `NetworkBehaviourAction::CloseConnection` to close connections to a specific
   peer via an `ExpandedSwarm` or `NetworkBehaviour`. See [PR 2110] for details.
 
+- Expose the `ListenerId` in `SwarmEvent`s that are associated with a listener.
+  
+  See [PR 2123] for details.
+
 [PR 2100]: https://github.com/libp2p/rust-libp2p/pull/2100
 [PR 2110]: https://github.com/libp2p/rust-libp2p/pull/2110/
+[PR 2123]: https://github.com/libp2p/rust-libp2p/pull/2123
 
 # 0.29.0 [2021-04-13]
 

swarm/src/lib.rs

@@ -229,11 +229,23 @@ pub enum SwarmEvent<TBvEv, THandleErr> {
         error: PendingConnectionError<io::Error>,
     },
     /// One of our listeners has reported a new local listening address.
-    NewListenAddr(Multiaddr),
+    NewListenAddr{
+        /// The listener that is listening on the new address.
+        listener_id: ListenerId,
+        /// The new address that is being listened on.
+        address: Multiaddr
+    },
     /// One of our listeners has reported the expiration of a listening address.
-    ExpiredListenAddr(Multiaddr),
+    ExpiredListenAddr{
+        /// The listener that is no longer listening on the address.
+        listener_id: ListenerId,
+        /// The expired address.
+        address: Multiaddr
+    },
     /// One of the listeners gracefully closed.
     ListenerClosed {
+        /// The listener that closed.
+        listener_id: ListenerId,
         /// The addresses that the listener was listening on. These addresses are now considered
         /// expired, similar to if a [`ExpiredListenAddr`](SwarmEvent::ExpiredListenAddr) event
         /// has been generated for each of them.
@@ -244,6 +256,8 @@ pub enum SwarmEvent<TBvEv, THandleErr> {
     },
     /// One of the listeners reported a non-fatal error.
     ListenerError {
+        /// The listener that errored.
+        listener_id: ListenerId,
         /// The listener error.
         error: io::Error,
     },
@@ -328,8 +342,10 @@ where TBehaviour: NetworkBehaviour<ProtocolsHandler = THandler>,
     }
 
     /// Starts listening on the given address.
-    ///
     /// Returns an error if the address is not supported.
+    ///
+    /// Listeners report their new listening addresses as [`SwarmEvent::NewListenAddr`].
+    /// Depending on the underlying transport, one listener may have multiple listening addresses.
     pub fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<io::Error>> {
         let id = self.network.listen_on(addr)?;
         self.behaviour.inject_new_listener(id);
@@ -588,13 +604,19 @@ where TBehaviour: NetworkBehaviour<ProtocolsHandler = THandler>,
                         this.listened_addrs.push(listen_addr.clone())
                     }
                     this.behaviour.inject_new_listen_addr(listener_id, &listen_addr);
-                    return Poll::Ready(SwarmEvent::NewListenAddr(listen_addr));
+                    return Poll::Ready(SwarmEvent::NewListenAddr {
+                        listener_id, 
+                        address: listen_addr
+                    });
                 }
                 Poll::Ready(NetworkEvent::ExpiredListenerAddress { listener_id, listen_addr }) => {
                     log::debug!("Listener {:?}; Expired address {:?}.", listener_id, listen_addr);
                     this.listened_addrs.retain(|a| a != &listen_addr);
                     this.behaviour.inject_expired_listen_addr(listener_id, &listen_addr);
-                    return Poll::Ready(SwarmEvent::ExpiredListenAddr(listen_addr));
+                    return Poll::Ready(SwarmEvent::ExpiredListenAddr{
+                        listener_id,
+                        address: listen_addr
+                    });
                 }
                 Poll::Ready(NetworkEvent::ListenerClosed { listener_id, addresses, reason }) => {
                     log::debug!("Listener {:?}; Closed by {:?}.", listener_id, reason);
@@ -606,6 +628,7 @@ where TBehaviour: NetworkBehaviour<ProtocolsHandler = THandler>,
                         Err(err) => Err(err),
                     });
                     return Poll::Ready(SwarmEvent::ListenerClosed {
+                        listener_id,
                         addresses,
                         reason,
                     });
@@ -613,6 +636,7 @@ where TBehaviour: NetworkBehaviour<ProtocolsHandler = THandler>,
                 Poll::Ready(NetworkEvent::ListenerError { listener_id, error }) => {
                     this.behaviour.inject_listener_error(listener_id, &error);
                     return Poll::Ready(SwarmEvent::ListenerError {
+                        listener_id,
                         error,
                     });
                 },