feat(swarm): allow NetworkBehaviours to create and remove listeners

swarm-derive/src/lib.rs

@@ -732,6 +732,12 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> syn::Result<Toke
                 std::task::Poll::Ready(#network_behaviour_action::Dial { opts }) => {
                     return std::task::Poll::Ready(#network_behaviour_action::Dial { opts });
                 }
+                std::task::Poll::Ready(#network_behaviour_action::ListenOn { id, address }) => {
+                    return std::task::Poll::Ready(#network_behaviour_action::ListenOn { id, address });
+                }
+                std::task::Poll::Ready(#network_behaviour_action::RemoveListener { id }) => {
+                    return std::task::Poll::Ready(#network_behaviour_action::RemoveListener { id });
+                }
                 std::task::Poll::Ready(#network_behaviour_action::NotifyHandler { peer_id, handler, event }) => {
                     return std::task::Poll::Ready(#network_behaviour_action::NotifyHandler {
                         peer_id,

swarm/CHANGELOG.md

@@ -1,5 +1,8 @@
 ## 0.43.0 - unreleased
 
+- Allow `NetworkBehaviours` to create and remove listeners.
+  See [PR 3292].
+
 - Raise MSRV to 1.65.
   See [PR 3715].
 
@@ -58,6 +61,7 @@
 
 - Remove deprecated items. See [PR 3956].
 
+[PR 3292]: https://github.com/libp2p/rust-libp2p/pull/3292
 [PR 3605]: https://github.com/libp2p/rust-libp2p/pull/3605
 [PR 3651]: https://github.com/libp2p/rust-libp2p/pull/3651
 [PR 3715]: https://github.com/libp2p/rust-libp2p/pull/3715

swarm/src/behaviour.rs

@@ -250,6 +250,12 @@ pub enum ToSwarm<TOutEvent, TInEvent> {
     /// This allows a [`NetworkBehaviour`] to identify a connection that resulted out of its own dial request.
     Dial { opts: DialOpts },
 
+    /// Instructs the [`Swarm`](crate::Swarm) to listen on the provided address.
+    ListenOn { id: ListenerId, address: Multiaddr },
+
+    /// Instructs the [`Swarm`](crate::Swarm) to remove the listener.
+    RemoveListener { id: ListenerId },
+
     /// Instructs the `Swarm` to send an event to the handler dedicated to a
     /// connection with a peer.
     ///
@@ -324,6 +330,8 @@ impl<TOutEvent, TInEventOld> ToSwarm<TOutEvent, TInEventOld> {
         match self {
             ToSwarm::GenerateEvent(e) => ToSwarm::GenerateEvent(e),
             ToSwarm::Dial { opts } => ToSwarm::Dial { opts },
+            ToSwarm::ListenOn { id, address } => ToSwarm::ListenOn { id, address },
+            ToSwarm::RemoveListener { id } => ToSwarm::RemoveListener { id },
             ToSwarm::NotifyHandler {
                 peer_id,
                 handler,
@@ -353,6 +361,8 @@ impl<TOutEvent, THandlerIn> ToSwarm<TOutEvent, THandlerIn> {
         match self {
             ToSwarm::GenerateEvent(e) => ToSwarm::GenerateEvent(f(e)),
             ToSwarm::Dial { opts } => ToSwarm::Dial { opts },
+            ToSwarm::ListenOn { id, address } => ToSwarm::ListenOn { id, address },
+            ToSwarm::RemoveListener { id } => ToSwarm::RemoveListener { id },
             ToSwarm::NotifyHandler {
                 peer_id,
                 handler,

swarm/src/lib.rs

@@ -372,11 +372,17 @@ where
     /// 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 = ListenerId::next();
-        self.transport.listen_on(id, addr)?;
-        self.behaviour
-            .on_swarm_event(FromSwarm::NewListener(behaviour::NewListener {
-                listener_id: id,
-            }));
+
+        if let Err(e) = self.transport.listen_on(id, addr) {
+            self.behaviour
+                .on_swarm_event(FromSwarm::ListenerError(behaviour::ListenerError {
+                    listener_id: id,
+                    err: &e,
+                }));
+            return Err(e);
+        }
+
+        self.add_listener(id);
         Ok(id)
     }
 
@@ -544,6 +550,14 @@ where
         self.confirmed_external_addr.iter()
     }
 
+    /// TBD
+    fn add_listener(&mut self, listener_id: ListenerId) {
+        self.behaviour
+            .on_swarm_event(FromSwarm::NewListener(behaviour::NewListener {
+                listener_id,
+            }));
+    }
+
     /// Add a **confirmed** external address for the local node.
     ///
     /// This function should only be called with addresses that are guaranteed to be reachable.
@@ -1016,6 +1030,21 @@ where
                     });
                 }
             }
+            ToSwarm::ListenOn { id, address } => {
+                if let Err(e) = self.transport.listen_on(id, address) {
+                    self.behaviour.on_swarm_event(FromSwarm::ListenerError(
+                        behaviour::ListenerError {
+                            listener_id: id,
+                            err: &e,
+                        },
+                    ));
+                }
+
+                self.add_listener(id);
+            }
+            ToSwarm::RemoveListener { id } => {
+                self.remove_listener(id);
+            }
             ToSwarm::NotifyHandler {
                 peer_id,
                 handler,

swarm/tests/listener.rs

@@ -0,0 +1,146 @@
+use std::{
+    collections::{HashSet, VecDeque},
+    task::{Context, Poll},
+};
+
+use libp2p_core::{multiaddr::Protocol, transport::ListenerId, Endpoint, Multiaddr};
+use libp2p_identity::PeerId;
+use libp2p_swarm::{
+    derive_prelude::NewListener, dummy, ConnectionDenied, ConnectionId, FromSwarm, ListenerClosed,
+    ListenerError, NetworkBehaviour, NewListenAddr, PollParameters, Swarm, SwarmEvent, THandler,
+    THandlerInEvent, THandlerOutEvent, ToSwarm,
+};
+
+use libp2p_swarm_test::SwarmExt;
+
+#[derive(Default)]
+struct Behaviour {
+    events: VecDeque<ToSwarm<<Self as NetworkBehaviour>::ToSwarm, THandlerInEvent<Self>>>,
+    listeners: HashSet<ListenerId>,
+}
+
+impl Behaviour {
+    pub(crate) fn listen(&mut self, addr: Multiaddr) -> ListenerId {
+        let listener_id = ListenerId::next();
+        assert!(!self.listeners.contains(&listener_id));
+        self.events.push_back(ToSwarm::ListenOn {
+            id: listener_id,
+            address: addr,
+        });
+
+        self.listeners.insert(listener_id);
+
+        listener_id
+    }
+
+    pub(crate) fn stop_listening(&mut self, id: ListenerId) {
+        self.events.push_back(ToSwarm::RemoveListener { id });
+    }
+}
+
+impl NetworkBehaviour for Behaviour {
+    type ConnectionHandler = dummy::ConnectionHandler;
+    type ToSwarm = void::Void;
+
+    fn handle_established_inbound_connection(
+        &mut self,
+        _: ConnectionId,
+        _: PeerId,
+        _: &Multiaddr,
+        _: &Multiaddr,
+    ) -> Result<libp2p_swarm::THandler<Self>, ConnectionDenied> {
+        Ok(dummy::ConnectionHandler)
+    }
+
+    fn handle_established_outbound_connection(
+        &mut self,
+        _: ConnectionId,
+        _: PeerId,
+        _: &Multiaddr,
+        _: Endpoint,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        Ok(dummy::ConnectionHandler)
+    }
+
+    fn on_connection_handler_event(
+        &mut self,
+        _: PeerId,
+        _: ConnectionId,
+        _: THandlerOutEvent<Self>,
+    ) {
+    }
+
+    fn on_swarm_event(&mut self, event: FromSwarm<Self::ConnectionHandler>) {
+        match event {
+            FromSwarm::NewListener(NewListener { listener_id }) => {
+                assert!(self.listeners.contains(&listener_id));
+            }
+            FromSwarm::NewListenAddr(NewListenAddr { listener_id, .. }) => {
+                assert!(self.listeners.contains(&listener_id));
+            }
+            FromSwarm::ListenerError(ListenerError { listener_id, err }) => {
+                panic!("Error for listener {listener_id:?}: {err}");
+            }
+            FromSwarm::ListenerClosed(ListenerClosed {
+                listener_id,
+                reason,
+            }) => {
+                assert!(self.listeners.contains(&listener_id));
+                assert!(reason.is_ok());
+                self.listeners.remove(&listener_id);
+                assert!(!self.listeners.contains(&listener_id));
+            }
+            _ => {}
+        }
+    }
+
+    fn poll(
+        &mut self,
+        _: &mut Context<'_>,
+        _: &mut impl PollParameters,
+    ) -> Poll<ToSwarm<Self::ToSwarm, THandlerInEvent<Self>>> {
+        if let Some(event) = self.events.pop_front() {
+            return Poll::Ready(event);
+        }
+
+        Poll::Pending
+    }
+}
+
+#[async_std::test]
+async fn behaviour_listener() {
+    let mut swarm = Swarm::new_ephemeral(|_| Behaviour::default());
+    let addr: Multiaddr = Protocol::Memory(0).into();
+    let id = swarm.behaviour_mut().listen(addr.clone());
+
+    let address = swarm
+        .wait(|e| match e {
+            SwarmEvent::NewListenAddr {
+                listener_id,
+                address,
+            } => {
+                assert_eq!(listener_id, id);
+                Some(address)
+            }
+            _ => None,
+        })
+        .await;
+
+    swarm.behaviour_mut().stop_listening(id);
+
+    swarm
+        .wait(|e| match e {
+            SwarmEvent::ListenerClosed {
+                listener_id,
+                addresses,
+                reason,
+            } => {
+                assert_eq!(listener_id, id);
+                assert!(addresses.contains(&address));
+                assert!(reason.is_ok());
+                Some(())
+            }
+            _ => None,
+        })
+        .await;
+}