feat(transport): allow ListenerId to be user-controlled

core/src/either.rs

@@ -154,14 +154,18 @@ where
         }
     }
 
-    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
+    fn listen_on(
+        &mut self,
+        id: ListenerId,
+        addr: Multiaddr,
+    ) -> Result<(), TransportError<Self::Error>> {
         use TransportError::*;
         match self {
-            Either::Left(a) => a.listen_on(addr).map_err(|e| match e {
+            Either::Left(a) => a.listen_on(id, addr).map_err(|e| match e {
                 MultiaddrNotSupported(addr) => MultiaddrNotSupported(addr),
                 Other(err) => Other(Either::Left(err)),
             }),
-            Either::Right(b) => b.listen_on(addr).map_err(|e| match e {
+            Either::Right(b) => b.listen_on(id, addr).map_err(|e| match e {
                 MultiaddrNotSupported(addr) => MultiaddrNotSupported(addr),
                 Other(err) => Other(Either::Right(err)),
             }),

core/src/transport.rs

@@ -31,6 +31,7 @@ use std::{
     error::Error,
     fmt,
     pin::Pin,
+    sync::atomic::{AtomicU64, Ordering},
     task::{Context, Poll},
 };
 
@@ -54,6 +55,8 @@ pub use self::memory::MemoryTransport;
 pub use self::optional::OptionalTransport;
 pub use self::upgrade::Upgrade;
 
+static NEXT_LISTENER_ID: AtomicU64 = AtomicU64::new(1);
+
 /// A transport provides connection-oriented communication between two peers
 /// through ordered streams of data (i.e. connections).
 ///
@@ -108,8 +111,12 @@ pub trait Transport {
     /// obtained from [dialing](Transport::dial).
     type Dial: Future<Output = Result<Self::Output, Self::Error>>;
 
-    /// Listens on the given [`Multiaddr`] for inbound connections.
-    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>>;
+    /// Listens on the given [`Multiaddr`] for inbound connections with a provided [`ListenerId`].
+    fn listen_on(
+        &mut self,
+        id: ListenerId,
+        addr: Multiaddr,
+    ) -> Result<(), TransportError<Self::Error>>;
 
     /// Remove a listener.
     ///
@@ -243,15 +250,22 @@ pub trait Transport {
 pub struct ListenerId(u64);
 
 impl ListenerId {
+    #[deprecated(note = "Renamed to ` ListenerId::next`.")]
+    #[allow(clippy::new_without_default)]
     /// Creates a new `ListenerId`.
     pub fn new() -> Self {
-        ListenerId(rand::random())
+        ListenerId::next()
+    }
+
+    /// Creates a new `ListenerId`.
+    pub fn next() -> Self {
+        ListenerId(NEXT_LISTENER_ID.fetch_add(1, Ordering::SeqCst))
     }
-}
 
-impl Default for ListenerId {
-    fn default() -> Self {
-        Self::new()
+    #[deprecated(note = "Use ` ListenerId::next` instead.")]
+    #[allow(clippy::should_implement_trait)]
+    pub fn default() -> Self {
+        Self::next()
     }
 }
 

core/src/transport/and_then.rs

@@ -54,9 +54,13 @@ where
     type ListenerUpgrade = AndThenFuture<T::ListenerUpgrade, C, F>;
     type Dial = AndThenFuture<T::Dial, C, F>;
 
-    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
+    fn listen_on(
+        &mut self,
+        id: ListenerId,
+        addr: Multiaddr,
+    ) -> Result<(), TransportError<Self::Error>> {
         self.transport
-            .listen_on(addr)
+            .listen_on(id, addr)
             .map_err(|err| err.map(Either::Left))
     }
 

core/src/transport/boxed.rs

@@ -52,7 +52,11 @@ type Dial<O> = Pin<Box<dyn Future<Output = io::Result<O>> + Send>>;
 type ListenerUpgrade<O> = Pin<Box<dyn Future<Output = io::Result<O>> + Send>>;
 
 trait Abstract<O> {
-    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<io::Error>>;
+    fn listen_on(
+        &mut self,
+        id: ListenerId,
+        addr: Multiaddr,
+    ) -> Result<(), TransportError<io::Error>>;
     fn remove_listener(&mut self, id: ListenerId) -> bool;
     fn dial(&mut self, addr: Multiaddr) -> Result<Dial<O>, TransportError<io::Error>>;
     fn dial_as_listener(&mut self, addr: Multiaddr) -> Result<Dial<O>, TransportError<io::Error>>;
@@ -70,8 +74,12 @@ where
     T::Dial: Send + 'static,
     T::ListenerUpgrade: Send + 'static,
 {
-    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<io::Error>> {
-        Transport::listen_on(self, addr).map_err(|e| e.map(box_err))
+    fn listen_on(
+        &mut self,
+        id: ListenerId,
+        addr: Multiaddr,
+    ) -> Result<(), TransportError<io::Error>> {
+        Transport::listen_on(self, id, addr).map_err(|e| e.map(box_err))
     }
 
     fn remove_listener(&mut self, id: ListenerId) -> bool {
@@ -123,8 +131,12 @@ impl<O> Transport for Boxed<O> {
     type ListenerUpgrade = ListenerUpgrade<O>;
     type Dial = Dial<O>;
 
-    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
-        self.inner.listen_on(addr)
+    fn listen_on(
+        &mut self,
+        id: ListenerId,
+        addr: Multiaddr,
+    ) -> Result<(), TransportError<Self::Error>> {
+        self.inner.listen_on(id, addr)
     }
 
     fn remove_listener(&mut self, id: ListenerId) -> bool {

core/src/transport/choice.rs

@@ -46,13 +46,17 @@ where
     type ListenerUpgrade = EitherFuture<A::ListenerUpgrade, B::ListenerUpgrade>;
     type Dial = EitherFuture<A::Dial, B::Dial>;
 
-    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
-        let addr = match self.0.listen_on(addr) {
+    fn listen_on(
+        &mut self,
+        id: ListenerId,
+        addr: Multiaddr,
+    ) -> Result<(), TransportError<Self::Error>> {
+        let addr = match self.0.listen_on(id, addr) {
             Err(TransportError::MultiaddrNotSupported(addr)) => addr,
             res => return res.map_err(|err| err.map(Either::Left)),
         };
 
-        let addr = match self.1.listen_on(addr) {
+        let addr = match self.1.listen_on(id, addr) {
             Err(TransportError::MultiaddrNotSupported(addr)) => addr,
             res => return res.map_err(|err| err.map(Either::Right)),
         };

core/src/transport/dummy.rs

@@ -59,7 +59,11 @@ impl<TOut> Transport for DummyTransport<TOut> {
     type ListenerUpgrade = futures::future::Pending<Result<Self::Output, io::Error>>;
     type Dial = futures::future::Pending<Result<Self::Output, io::Error>>;
 
-    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
+    fn listen_on(
+        &mut self,
+        _id: ListenerId,
+        addr: Multiaddr,
+    ) -> Result<(), TransportError<Self::Error>> {
         Err(TransportError::MultiaddrNotSupported(addr))
     }
 

core/src/transport/map.rs

@@ -61,8 +61,12 @@ where
     type ListenerUpgrade = MapFuture<T::ListenerUpgrade, F>;
     type Dial = MapFuture<T::Dial, F>;
 
-    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
-        self.transport.listen_on(addr)
+    fn listen_on(
+        &mut self,
+        id: ListenerId,
+        addr: Multiaddr,
+    ) -> Result<(), TransportError<Self::Error>> {
+        self.transport.listen_on(id, addr)
     }
 
     fn remove_listener(&mut self, id: ListenerId) -> bool {

core/src/transport/map_err.rs

@@ -50,9 +50,15 @@ where
     type ListenerUpgrade = MapErrListenerUpgrade<T, F>;
     type Dial = MapErrDial<T, F>;
 
-    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
+    fn listen_on(
+        &mut self,
+        id: ListenerId,
+        addr: Multiaddr,
+    ) -> Result<(), TransportError<Self::Error>> {
         let map = self.map.clone();
-        self.transport.listen_on(addr).map_err(|err| err.map(map))
+        self.transport
+            .listen_on(id, addr)
+            .map_err(|err| err.map(map))
     }
 
     fn remove_listener(&mut self, id: ListenerId) -> bool {

core/src/transport/memory.rs

@@ -179,7 +179,11 @@ impl Transport for MemoryTransport {
     type ListenerUpgrade = Ready<Result<Self::Output, Self::Error>>;
     type Dial = DialFuture;
 
-    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
+    fn listen_on(
+        &mut self,
+        id: ListenerId,
+        addr: Multiaddr,
+    ) -> Result<(), TransportError<Self::Error>> {
         let port = if let Ok(port) = parse_memory_addr(&addr) {
             port
         } else {
@@ -191,7 +195,6 @@ impl Transport for MemoryTransport {
             None => return Err(TransportError::Other(MemoryTransportError::Unreachable)),
         };
 
-        let id = ListenerId::new();
         let listener = Listener {
             id,
             port,
@@ -201,7 +204,7 @@ impl Transport for MemoryTransport {
         };
         self.listeners.push_back(Box::pin(listener));
 
-        Ok(id)
+        Ok(())
     }
 
     fn remove_listener(&mut self, id: ListenerId) -> bool {
@@ -457,30 +460,40 @@ mod tests {
         let addr_1: Multiaddr = "/memory/1639174018481".parse().unwrap();
         let addr_2: Multiaddr = "/memory/8459375923478".parse().unwrap();
 
-        let listener_id_1 = transport.listen_on(addr_1.clone()).unwrap();
+        let listener_id_1 = ListenerId::next();
+
+        transport.listen_on(listener_id_1, addr_1.clone()).unwrap();
         assert!(
             transport.remove_listener(listener_id_1),
             "Listener doesn't exist."
         );
 
-        let listener_id_2 = transport.listen_on(addr_1.clone()).unwrap();
-        let listener_id_3 = transport.listen_on(addr_2.clone()).unwrap();
+        let listener_id_2 = ListenerId::next();
+        transport.listen_on(listener_id_2, addr_1.clone()).unwrap();
+        let listener_id_3 = ListenerId::next();
+        transport.listen_on(listener_id_3, addr_2.clone()).unwrap();
 
-        assert!(transport.listen_on(addr_1.clone()).is_err());
-        assert!(transport.listen_on(addr_2.clone()).is_err());
+        assert!(transport
+            .listen_on(ListenerId::next(), addr_1.clone())
+            .is_err());
+        assert!(transport
+            .listen_on(ListenerId::next(), addr_2.clone())
+            .is_err());
 
         assert!(
             transport.remove_listener(listener_id_2),
             "Listener doesn't exist."
         );
-        assert!(transport.listen_on(addr_1).is_ok());
-        assert!(transport.listen_on(addr_2.clone()).is_err());
+        assert!(transport.listen_on(ListenerId::next(), addr_1).is_ok());
+        assert!(transport
+            .listen_on(ListenerId::next(), addr_2.clone())
+            .is_err());
 
         assert!(
             transport.remove_listener(listener_id_3),
             "Listener doesn't exist."
         );
-        assert!(transport.listen_on(addr_2).is_ok());
+        assert!(transport.listen_on(ListenerId::next(), addr_2).is_ok());
     }
 
     #[test]
@@ -489,8 +502,11 @@ mod tests {
         assert!(transport
             .dial("/memory/810172461024613".parse().unwrap())
             .is_err());
-        let _listener = transport
-            .listen_on("/memory/810172461024613".parse().unwrap())
+        transport
+            .listen_on(
+                ListenerId::next(),
+                "/memory/810172461024613".parse().unwrap(),
+            )
             .unwrap();
         assert!(transport
             .dial("/memory/810172461024613".parse().unwrap())
@@ -504,7 +520,8 @@ mod tests {
 
         let mut transport = MemoryTransport::default().boxed();
         futures::executor::block_on(async {
-            let listener_id = transport.listen_on(addr.clone()).unwrap();
+            let listener_id = ListenerId::next();
+            transport.listen_on(listener_id, addr.clone()).unwrap();
             let reported_addr = transport
                 .select_next_some()
                 .await
@@ -539,7 +556,7 @@ mod tests {
         let mut t1 = MemoryTransport::default().boxed();
 
         let listener = async move {
-            t1.listen_on(t1_addr.clone()).unwrap();
+            t1.listen_on(ListenerId::next(), t1_addr.clone()).unwrap();
             let upgrade = loop {
                 let event = t1.select_next_some().await;
                 if let Some(upgrade) = event.into_incoming() {
@@ -577,7 +594,9 @@ mod tests {
         let mut listener_transport = MemoryTransport::default().boxed();
 
         let listener = async move {
-            listener_transport.listen_on(listener_addr.clone()).unwrap();
+            listener_transport
+                .listen_on(ListenerId::next(), listener_addr.clone())
+                .unwrap();
             loop {
                 if let TransportEvent::Incoming { send_back_addr, .. } =
                     listener_transport.select_next_some().await
@@ -614,7 +633,9 @@ mod tests {
         let mut listener_transport = MemoryTransport::default().boxed();
 
         let listener = async move {
-            listener_transport.listen_on(listener_addr.clone()).unwrap();
+            listener_transport
+                .listen_on(ListenerId::next(), listener_addr.clone())
+                .unwrap();
             loop {
                 if let TransportEvent::Incoming { send_back_addr, .. } =
                     listener_transport.select_next_some().await

core/src/transport/optional.rs

@@ -60,9 +60,13 @@ where
     type ListenerUpgrade = T::ListenerUpgrade;
     type Dial = T::Dial;
 
-    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
+    fn listen_on(
+        &mut self,
+        id: ListenerId,
+        addr: Multiaddr,
+    ) -> Result<(), TransportError<Self::Error>> {
         if let Some(inner) = self.0.as_mut() {
-            inner.listen_on(addr)
+            inner.listen_on(id, addr)
         } else {
             Err(TransportError::MultiaddrNotSupported(addr))
         }

core/src/transport/timeout.rs

@@ -85,9 +85,13 @@ where
     type ListenerUpgrade = Timeout<InnerTrans::ListenerUpgrade>;
     type Dial = Timeout<InnerTrans::Dial>;
 
-    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
+    fn listen_on(
+        &mut self,
+        id: ListenerId,
+        addr: Multiaddr,
+    ) -> Result<(), TransportError<Self::Error>> {
         self.inner
-            .listen_on(addr)
+            .listen_on(id, addr)
             .map_err(|err| err.map(TransportTimeoutError::Other))
     }