fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-05-28 18:21:20 +00:00
Code Issues Projects Releases Wiki Activity

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

Browse Source 
`Transport::listen_on` is an asynchronous operation. It returns immediately but the actual process of establishing a listening socket happens as part of `Transport::poll` which will return one or more `TransportEvent`s related to a particular `listen_on` call.

Currently, `listen_on` returns a `ListenerId` which allows the user of the `Transport` interface to correlate the events with a particular `listen_on` call. This "user" is the `Swarm` runtime. Currently, a user of libp2p establishes a new listening socket by talking to the `Swarm::listen_on` interface and it is not possible to do the same thing via the `NetworkBehaviour` trait.

Within the `NetworkBehaviour` trait, we emit _commands_ to the `Swarm` like `ToSwarm::Dial`. These commands don't have a "return value" like a synchronous function does and thus, if we were to add a `ToSwarm::ListenOn` command, it could not receive the `ListenerId` from the `Transport`.

To fix this and to be consistent with our [coding guidelines](https://github.com/libp2p/rust-libp2p/blob/master/docs/coding-guidelines.md#allow-correlating-asynchronous-responses-to-their-requests) we change the interface of `Transport::listen_on` to require the user to pass in a `ListenerId`. This will allow us to construct a command in a `NetworkBehaviour` that remembers this ID which enables precise tracking of which events containing a `ListenerId` correlate which a particular `listen_on` command.

This is especially important in the context of listening on wildcard addresses like `0.0.0.0` because we end up binding to multiple network interfaces and thus emit multiple events for a single `listen_on` call.

Pull-Request: #3567.
This commit is contained in:
Darius Clark 2023-05-14 05:42:51 -04:00 committed by GitHub
parent 9e625881d5
commit 5b32c8a0d2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
34 changed files with 289 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
core/CHANGELOG.md
View File

@ -1,11 +1,15 @@
## 0.40.0 - unreleased
- Allow `ListenerId` to be user-controlled, i.e. to be provided on `Transport::listen_on`.
See [PR 3567].
- Raise MSRV to 1.65.
See [PR 3715].
- Remove deprecated symbols related to upgrades.
See [PR 3867].
[PR 3567]: https://github.com/libp2p/rust-libp2p/pull/3567
[PR 3715]: https://github.com/libp2p/rust-libp2p/pull/3715
[PR 3867]: https://github.com/libp2p/rust-libp2p/pull/3867

10
core/src/either.rs
View File

@ -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)),
}),

30
core/src/transport.rs
View File

@ -31,6 +31,7 @@ use std::{
error::Error,
fmt,
pin::Pin,
sync::atomic::{AtomicUsize, Ordering},
task::{Context, Poll},
};
@ -55,6 +56,8 @@ pub use self::memory::MemoryTransport;
pub use self::optional::OptionalTransport;
pub use self::upgrade::Upgrade;
static NEXT_LISTENER_ID: AtomicUsize = AtomicUsize::new(1);
/// A transport provides connection-oriented communication between two peers
/// through ordered streams of data (i.e. connections).
///
@ -109,8 +112,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.
///
@ -241,18 +248,25 @@ pub trait Transport {
/// The ID of a single listener.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
pub struct ListenerId(u64);
pub struct ListenerId(usize);
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()
}
}
impl Default for ListenerId {
fn default() -> Self {
Self::new()
/// Creates a new `ListenerId`.
pub fn next() -> Self {
ListenerId(NEXT_LISTENER_ID.fetch_add(1, Ordering::SeqCst))
}
#[deprecated(note = "Use ` ListenerId::next` instead.")]
#[allow(clippy::should_implement_trait)]
pub fn default() -> Self {
Self::next()
}
}

8
core/src/transport/and_then.rs
View File

@ -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))
}

22
core/src/transport/boxed.rs
View File

@ -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 {

10
core/src/transport/choice.rs
View File

@ -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)),
};

6
core/src/transport/dummy.rs
View File

@ -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))
}

8
core/src/transport/global_only.rs
View File

@ -276,8 +276,12 @@ impl<T: crate::Transport + Unpin> crate::Transport for Transport<T> {
type ListenerUpgrade = <T as crate::Transport>::ListenerUpgrade;
type Dial = <T as crate::Transport>::Dial;
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 {

8
core/src/transport/map.rs
View File

@ -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 {

10
core/src/transport/map_err.rs
View File

@ -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 {

55
core/src/transport/memory.rs
View File

@ -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

8
core/src/transport/optional.rs
View File

@ -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))
}

8
core/src/transport/timeout.rs
View File

@ -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))
}

16
core/src/transport/upgrade.rs
View File

@ -350,8 +350,12 @@ where
self.0.dial_as_listener(addr)
}
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
self.0.listen_on(addr)
fn listen_on(
&mut self,
id: ListenerId,
addr: Multiaddr,
) -> Result<(), TransportError<Self::Error>> {
self.0.listen_on(id, addr)
}
fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
@ -429,9 +433,13 @@ 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>> {
self.inner
.listen_on(addr)
.listen_on(id, addr)
.map_err(|err| err.map(TransportUpgradeError::Transport))
}

6
core/tests/transport_upgrade.rs
View File

@ -19,7 +19,7 @@
// DEALINGS IN THE SOFTWARE.
use futures::prelude::*;
use libp2p_core::transport::{MemoryTransport, Transport};
use libp2p_core::transport::{ListenerId, MemoryTransport, Transport};
use libp2p_core::upgrade::{self, InboundUpgrade, OutboundUpgrade, UpgradeInfo};
use libp2p_identity as identity;
use libp2p_mplex::MplexConfig;
@ -102,7 +102,9 @@ fn upgrade_pipeline() {
let listen_addr1 = Multiaddr::from(Protocol::Memory(random::<u64>()));
let listen_addr2 = listen_addr1.clone();
listener_transport.listen_on(listen_addr1).unwrap();
listener_transport
.listen_on(ListenerId::next(), listen_addr1)
.unwrap();
let server = async move {
loop {

5
muxers/mplex/benches/split_send_size.rs
View File

@ -27,6 +27,7 @@ use futures::future::poll_fn;
use futures::prelude::*;
use futures::{channel::oneshot, future::join};
use libp2p_core::muxing::StreamMuxerExt;
use libp2p_core::transport::ListenerId;
use libp2p_core::{multiaddr::multiaddr, muxing, transport, upgrade, Multiaddr, Transport};
use libp2p_identity as identity;
use libp2p_identity::PeerId;
@ -100,7 +101,9 @@ fn run(
payload: &Vec<u8>,
listen_addr: &Multiaddr,
) {
receiver_trans.listen_on(listen_addr.clone()).unwrap();
receiver_trans
.listen_on(ListenerId::next(), listen_addr.clone())
.unwrap();
let (addr_sender, addr_receiver) = oneshot::channel();
let mut addr_sender = Some(addr_sender);
let payload_len = payload.len();

4
protocols/ping/src/protocol.rs
View File

@ -88,7 +88,7 @@ mod tests {
use futures::StreamExt;
use libp2p_core::{
multiaddr::multiaddr,
transport::{memory::MemoryTransport, Transport},
transport::{memory::MemoryTransport, ListenerId, Transport},
};
use rand::{thread_rng, Rng};
use std::time::Duration;
@ -97,7 +97,7 @@ mod tests {
fn ping_pong() {
let mem_addr = multiaddr![Memory(thread_rng().gen::<u64>())];
let mut transport = MemoryTransport::new().boxed();
transport.listen_on(mem_addr).unwrap();
transport.listen_on(ListenerId::next(), mem_addr).unwrap();
let listener_addr = transport
.select_next_some()

13
protocols/relay/src/priv_client/transport.rs
View File

@ -70,7 +70,7 @@ use thiserror::Error;
/// 3. Listen for incoming relayed connections via specific relay.
///
/// ```
/// # use libp2p_core::{Multiaddr, multiaddr::{Protocol}, Transport, PeerId};
/// # use libp2p_core::{Multiaddr, multiaddr::{Protocol}, transport::ListenerId, Transport, PeerId};
/// # use libp2p_core::transport::memory::MemoryTransport;
/// # use libp2p_core::transport::choice::OrTransport;
/// # use libp2p_relay as relay;
@ -85,7 +85,7 @@ use thiserror::Error;
/// .with(Protocol::Memory(40)) // Relay address.
/// .with(Protocol::P2p(relay_id.into())) // Relay peer id.
/// .with(Protocol::P2pCircuit); // Signal to listen via remote relay node.
/// transport.listen_on(relay_addr).unwrap();
/// transport.listen_on(ListenerId::next(), relay_addr).unwrap();
/// ```
pub struct Transport {
to_behaviour: mpsc::Sender<TransportToBehaviourMsg>,
@ -111,7 +111,11 @@ impl libp2p_core::Transport for Transport {
type ListenerUpgrade = Ready<Result<Self::Output, Self::Error>>;
type Dial = BoxFuture<'static, Result<Connection, Error>>;
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
fn listen_on(
&mut self,
listener_id: ListenerId,
addr: Multiaddr,
) -> Result<(), TransportError<Self::Error>> {
let (relay_peer_id, relay_addr) = match parse_relayed_multiaddr(addr)? {
RelayedMultiaddr {
relay_peer_id: None,
@ -138,7 +142,6 @@ impl libp2p_core::Transport for Transport {
to_listener,
});
let listener_id = ListenerId::new();
let listener = Listener {
listener_id,
queued_events: Default::default(),
@ -146,7 +149,7 @@ impl libp2p_core::Transport for Transport {
is_closed: false,
};
self.listeners.push(listener);
Ok(listener_id)
Ok(())
}
fn remove_listener(&mut self, id: ListenerId) -> bool {

6
swarm/src/behaviour/listen_addresses.rs
View File

@ -34,7 +34,7 @@ impl ListenAddresses {
mod tests {
use super::*;
use crate::dummy;
use libp2p_core::multiaddr::Protocol;
use libp2p_core::{multiaddr::Protocol, transport::ListenerId};
use once_cell::sync::Lazy;
#[test]
@ -62,14 +62,14 @@ mod tests {
fn new_listen_addr() -> FromSwarm<'static, dummy::ConnectionHandler> {
FromSwarm::NewListenAddr(NewListenAddr {
listener_id: Default::default(),
listener_id: ListenerId::next(),
addr: &MEMORY_ADDR,
})
}
fn expired_listen_addr() -> FromSwarm<'static, dummy::ConnectionHandler> {
FromSwarm::ExpiredListenAddr(ExpiredListenAddr {
listener_id: Default::default(),
listener_id: ListenerId::next(),
addr: &MEMORY_ADDR,
})
}

7
swarm/src/lib.rs
View File

@ -474,7 +474,8 @@ where
/// 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.transport.listen_on(addr)?;
let id = ListenerId::next();
self.transport.listen_on(id, addr)?;
self.behaviour
.on_swarm_event(FromSwarm::NewListener(behaviour::NewListener {
listener_id: id,
@ -2222,7 +2223,9 @@ mod tests {
let mut transports = Vec::new();
for _ in 0..num_listen_addrs {
let mut transport = transport::MemoryTransport::default().boxed();
transport.listen_on("/memory/0".parse().unwrap()).unwrap();
transport
.listen_on(ListenerId::next(), "/memory/0".parse().unwrap())
.unwrap();
match transport.select_next_some().await {
TransportEvent::NewAddress { listen_addr, .. } => {

11
transports/dns/src/lib.rs
View File

@ -198,10 +198,14 @@ where
BoxFuture<'static, Result<Self::Output, Self::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>> {
self.inner
.lock()
.listen_on(addr)
.listen_on(id, addr)
.map_err(|e| e.map(DnsErr::Transport))
}
@ -599,8 +603,9 @@ mod tests {
fn listen_on(
&mut self,
_: ListenerId,
_: Multiaddr,
) -> Result<ListenerId, TransportError<Self::Error>> {
) -> Result<(), TransportError<Self::Error>> {
unreachable!()
}

4
transports/quic/src/lib.rs
View File

@ -32,7 +32,7 @@
//! # fn main() -> std::io::Result<()> {
//! #
//! use libp2p_quic as quic;
//! use libp2p_core::{Multiaddr, Transport};
//! use libp2p_core::{Multiaddr, Transport, transport::ListenerId};
//!
//! let keypair = libp2p_identity::Keypair::generate_ed25519();
//! let quic_config = quic::Config::new(&keypair);
@ -40,7 +40,7 @@
//! let mut quic_transport = quic::async_std::Transport::new(quic_config);
//!
//! let addr = "/ip4/127.0.0.1/udp/12345/quic-v1".parse().expect("address should be valid");
//! quic_transport.listen_on(addr).expect("listen error.");
//! quic_transport.listen_on(ListenerId::next(), addr).expect("listen error.");
//! #
//! # Ok(())
//! # }

21
transports/quic/src/transport.rs
View File

@ -98,10 +98,13 @@ impl<P: Provider> Transport for GenTransport<P> {
type ListenerUpgrade = Connecting;
type Dial = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
fn listen_on(
&mut self,
listener_id: ListenerId,
addr: Multiaddr,
) -> Result<(), TransportError<Self::Error>> {
let (socket_addr, version) = multiaddr_to_socketaddr(&addr, self.support_draft_29)
.ok_or(TransportError::MultiaddrNotSupported(addr))?;
let listener_id = ListenerId::new();
let listener = Listener::new(
listener_id,
socket_addr,
@ -120,7 +123,7 @@ impl<P: Provider> Transport for GenTransport<P> {
// New outbound connections will use the bidirectional (listener) endpoint.
self.dialer.remove(&socket_addr.ip().into());
Ok(listener_id)
Ok(())
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
@ -772,8 +775,9 @@ mod test {
// Run test twice to check that there is no unexpected behaviour if `Transport.listener`
// is temporarily empty.
for _ in 0..2 {
let id = transport
.listen_on("/ip4/0.0.0.0/udp/0/quic-v1".parse().unwrap())
let id = ListenerId::next();
transport
.listen_on(id, "/ip4/0.0.0.0/udp/0/quic-v1".parse().unwrap())
.unwrap();
// Copy channel to use it later.
@ -866,8 +870,11 @@ mod test {
.await;
// Start listening so that the dialer and driver are dropped.
let _ = transport
.listen_on("/ip4/0.0.0.0/udp/0/quic-v1".parse().unwrap())
transport
.listen_on(
ListenerId::next(),
"/ip4/0.0.0.0/udp/0/quic-v1".parse().unwrap(),
)
.unwrap();
assert!(!transport.dialer.contains_key(&SocketFamily::Ipv4));

14
transports/quic/tests/smoke.rs
View File

@ -115,9 +115,10 @@ async fn wrapped_with_delay() {
fn listen_on(
&mut self,
id: ListenerId,
addr: Multiaddr,
) -> Result<ListenerId, TransportError<Self::Error>> {
self.0.lock().unwrap().listen_on(addr)
) -> Result<(), TransportError<Self::Error>> {
self.0.lock().unwrap().listen_on(id, addr)
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
@ -327,7 +328,10 @@ async fn draft_29_support() {
let (_, mut d_transport) =
create_transport::<quic::tokio::Provider>(|cfg| cfg.support_draft_29 = false);
assert!(matches!(
d_transport.listen_on("/ip4/127.0.0.1/udp/0/quic".parse().unwrap()),
d_transport.listen_on(
ListenerId::next(),
"/ip4/127.0.0.1/udp/0/quic".parse().unwrap()
),
Err(TransportError::MultiaddrNotSupported(_))
));
let d_quic_v1_addr = start_listening(&mut d_transport, "/ip4/127.0.0.1/udp/0/quic-v1").await;
@ -509,7 +513,9 @@ fn create_transport<P: Provider>(
}
async fn start_listening(transport: &mut Boxed<(PeerId, StreamMuxerBox)>, addr: &str) -> Multiaddr {
transport.listen_on(addr.parse().unwrap()).unwrap();
transport
.listen_on(ListenerId::next(), addr.parse().unwrap())
.unwrap();
match transport.next().await {
Some(TransportEvent::NewAddress { listen_addr, .. }) => listen_addr,
e => panic!("{e:?}"),

6
transports/quic/tests/stream_compliance.rs
View File

@ -1,5 +1,6 @@
use futures::channel::oneshot;
use futures::StreamExt;
use libp2p_core::transport::ListenerId;
use libp2p_core::Transport;
use libp2p_quic as quic;
use std::time::Duration;
@ -23,7 +24,10 @@ async fn connected_peers() -> (quic::Connection, quic::Connection) {
let mut listener = new_transport().boxed();
listener
.listen_on("/ip4/127.0.0.1/udp/0/quic-v1".parse().unwrap())
.listen_on(
ListenerId::next(),
"/ip4/127.0.0.1/udp/0/quic-v1".parse().unwrap(),
)
.unwrap();
let listen_address = listener.next().await.unwrap().into_new_address().unwrap();

35
transports/tcp/src/lib.rs
View File

@ -248,7 +248,7 @@ impl Config {
/// let listen_addr2: Multiaddr = "/ip4/127.0.0.1/tcp/9002".parse().unwrap();
///
/// let mut tcp1 = libp2p_tcp::async_io::Transport::new(libp2p_tcp::Config::new().port_reuse(true)).boxed();
/// tcp1.listen_on( listen_addr1.clone()).expect("listener");
/// tcp1.listen_on(ListenerId::next(), listen_addr1.clone()).expect("listener");
/// match tcp1.select_next_some().await {
/// TransportEvent::NewAddress { listen_addr, .. } => {
/// println!("Listening on {:?}", listen_addr);
@ -259,7 +259,7 @@ impl Config {
/// }
///
/// let mut tcp2 = libp2p_tcp::async_io::Transport::new(libp2p_tcp::Config::new().port_reuse(true)).boxed();
/// tcp2.listen_on( listen_addr2).expect("listener");
/// tcp2.listen_on(ListenerId::next(), listen_addr2).expect("listener");
/// match tcp2.select_next_some().await {
/// TransportEvent::NewAddress { listen_addr, .. } => {
/// println!("Listening on {:?}", listen_addr);
@ -437,19 +437,22 @@ where
type Dial = Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send>>;
type ListenerUpgrade = Ready<Result<Self::Output, Self::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>> {
let socket_addr = if let Ok(sa) = multiaddr_to_socketaddr(addr.clone()) {
sa
} else {
return Err(TransportError::MultiaddrNotSupported(addr));
};
let id = ListenerId::new();
log::debug!("listening on {}", socket_addr);
let listener = self
.do_listen(id, socket_addr)
.map_err(TransportError::Other)?;
self.listeners.push(listener);
Ok(id)
Ok(())
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
@ -916,7 +919,7 @@ mod tests {
async fn listener<T: Provider>(addr: Multiaddr, mut ready_tx: mpsc::Sender<Multiaddr>) {
let mut tcp = Transport::<T>::default().boxed();
tcp.listen_on(addr).unwrap();
tcp.listen_on(ListenerId::next(), addr).unwrap();
loop {
match tcp.select_next_some().await {
TransportEvent::NewAddress { listen_addr, .. } => {
@ -985,7 +988,7 @@ mod tests {
async fn listener<T: Provider>(addr: Multiaddr, mut ready_tx: mpsc::Sender<Multiaddr>) {
let mut tcp = Transport::<T>::default().boxed();
tcp.listen_on(addr).unwrap();
tcp.listen_on(ListenerId::next(), addr).unwrap();
loop {
match tcp.select_next_some().await {
@ -1058,7 +1061,7 @@ mod tests {
port_reuse_rx: oneshot::Receiver<Protocol<'_>>,
) {
let mut tcp = Transport::<T>::new(Config::new()).boxed();
tcp.listen_on(addr).unwrap();
tcp.listen_on(ListenerId::next(), addr).unwrap();
loop {
match tcp.select_next_some().await {
TransportEvent::NewAddress { listen_addr, .. } => {
@ -1093,7 +1096,7 @@ mod tests {
) {
let dest_addr = ready_rx.next().await.unwrap();
let mut tcp = Transport::<T>::new(Config::new().port_reuse(true));
tcp.listen_on(addr).unwrap();
tcp.listen_on(ListenerId::next(), addr).unwrap();
match poll_fn(|cx| Pin::new(&mut tcp).poll(cx)).await {
TransportEvent::NewAddress { .. } => {
// Check that tcp and listener share the same port reuse SocketAddr
@ -1161,7 +1164,7 @@ mod tests {
async fn listen_twice<T: Provider>(addr: Multiaddr) {
let mut tcp = Transport::<T>::new(Config::new().port_reuse(true));
tcp.listen_on(addr).unwrap();
tcp.listen_on(ListenerId::next(), addr).unwrap();
match poll_fn(|cx| Pin::new(&mut tcp).poll(cx)).await {
TransportEvent::NewAddress {
listen_addr: addr1, ..
@ -1176,7 +1179,7 @@ mod tests {
assert_eq!(port_reuse_tcp, port_reuse_listener1);
// Listen on the same address a second time.
tcp.listen_on(addr1.clone()).unwrap();
tcp.listen_on(ListenerId::next(), addr1.clone()).unwrap();
match poll_fn(|cx| Pin::new(&mut tcp).poll(cx)).await {
TransportEvent::NewAddress {
listen_addr: addr2, ..
@ -1215,7 +1218,7 @@ mod tests {
async fn listen<T: Provider>(addr: Multiaddr) -> Multiaddr {
let mut tcp = Transport::<T>::default().boxed();
tcp.listen_on(addr).unwrap();
tcp.listen_on(ListenerId::next(), addr).unwrap();
tcp.select_next_some()
.await
.into_new_address()
@ -1252,13 +1255,13 @@ mod tests {
#[cfg(feature = "async-io")]
{
let mut tcp = async_io::Transport::default();
assert!(tcp.listen_on(addr.clone()).is_err());
assert!(tcp.listen_on(ListenerId::next(), addr.clone()).is_err());
}
#[cfg(feature = "tokio")]
{
let mut tcp = tokio::Transport::default();
assert!(tcp.listen_on(addr).is_err());
assert!(tcp.listen_on(ListenerId::next(), addr).is_err());
}
}
@ -1320,8 +1323,8 @@ mod tests {
async fn cycle_listeners<T: Provider>() -> bool {
let mut tcp = Transport::<T>::default().boxed();
let listener_id = tcp
.listen_on("/ip4/127.0.0.1/tcp/0".parse().unwrap())
let listener_id = ListenerId::next();
tcp.listen_on(listener_id, "/ip4/127.0.0.1/tcp/0".parse().unwrap())
.unwrap();
tcp.remove_listener(listener_id)
}

5
transports/tcp/src/provider/async_io.rs
View File

@ -32,14 +32,15 @@ use std::task::{Context, Poll};
///
/// ```rust
/// # use libp2p_tcp as tcp;
/// # use libp2p_core::Transport;
/// # use libp2p_core::{Transport, transport::ListenerId};
/// # use futures::future;
/// # use std::pin::Pin;
/// #
/// # #[async_std::main]
/// # async fn main() {
/// let mut transport = tcp::async_io::Transport::new(tcp::Config::default());
/// let id = transport.listen_on("/ip4/127.0.0.1/tcp/0".parse().unwrap()).unwrap();
/// let id = ListenerId::next();
/// transport.listen_on(id, "/ip4/127.0.0.1/tcp/0".parse().unwrap()).unwrap();
///
/// let addr = future::poll_fn(|cx| Pin::new(&mut transport).poll(cx)).await.into_new_address().unwrap();
///

4
transports/tcp/src/provider/tokio.rs
View File

@ -36,14 +36,14 @@ use std::task::{Context, Poll};
///
/// ```rust
/// # use libp2p_tcp as tcp;
/// # use libp2p_core::Transport;
/// # use libp2p_core::{Transport, transport::ListenerId};
/// # use futures::future;
/// # use std::pin::Pin;
/// #
/// # #[tokio::main]
/// # async fn main() {
/// let mut transport = tcp::tokio::Transport::new(tcp::Config::default());
/// let id = transport.listen_on("/ip4/127.0.0.1/tcp/0".parse().unwrap()).unwrap();
/// let id = transport.listen_on(ListenerId::next(), "/ip4/127.0.0.1/tcp/0".parse().unwrap()).unwrap();
///
/// let addr = future::poll_fn(|cx| Pin::new(&mut transport).poll(cx)).await.into_new_address().unwrap();
///

11
transports/uds/src/lib.rs
View File

@ -93,10 +93,10 @@ macro_rules! codegen {
fn listen_on(
&mut self,
id: ListenerId,
addr: Multiaddr,
) -> Result<ListenerId, TransportError<Self::Error>> {
) -> Result<(), TransportError<Self::Error>> {
if let Ok(path) = multiaddr_to_path(&addr) {
let id = ListenerId::new();
let listener = $build_listener(path)
.map_err(Err)
.map_ok(move |listener| {
@ -138,7 +138,7 @@ macro_rules! codegen {
.try_flatten_stream()
.boxed();
self.listeners.push_back((id, listener));
Ok(id)
Ok(())
} else {
Err(TransportError::MultiaddrNotSupported(addr))
}
@ -260,6 +260,7 @@ mod tests {
use futures::{channel::oneshot, prelude::*};
use libp2p_core::{
multiaddr::{Multiaddr, Protocol},
transport::ListenerId,
Transport,
};
use std::{self, borrow::Cow, path::Path};
@ -292,7 +293,7 @@ mod tests {
async_std::task::spawn(async move {
let mut transport = UdsConfig::new().boxed();
transport.listen_on(addr).unwrap();
transport.listen_on(ListenerId::next(), addr).unwrap();
let listen_addr = transport
.select_next_some()
@ -328,7 +329,7 @@ mod tests {
let mut uds = UdsConfig::new();
let addr = "/unix//foo/bar".parse::<Multiaddr>().unwrap();
assert!(uds.listen_on(addr).is_err());
assert!(uds.listen_on(ListenerId::next(), addr).is_err());
}
#[test]

9
transports/wasm-ext/src/lib.rs
View File

@ -198,7 +198,11 @@ impl Transport for ExtTransport {
type ListenerUpgrade = Ready<Result<Self::Output, Self::Error>>;
type Dial = Dial;
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
fn listen_on(
&mut self,
listener_id: ListenerId,
addr: Multiaddr,
) -> Result<(), TransportError<Self::Error>> {
let iter = self.inner.listen_on(&addr.to_string()).map_err(|err| {
if is_not_supported_error(&err) {
TransportError::MultiaddrNotSupported(addr)
@ -206,7 +210,6 @@ impl Transport for ExtTransport {
TransportError::Other(JsErr::from(err))
}
})?;
let listener_id = ListenerId::new();
let listen = Listen {
listener_id,
iterator: SendWrapper::new(iter),
@ -215,7 +218,7 @@ impl Transport for ExtTransport {
is_closed: false,
};
self.listeners.push(listen);
Ok(listener_id)
Ok(())
}
fn remove_listener(&mut self, id: ListenerId) -> bool {

18
transports/webrtc/src/tokio/transport.rs
View File

@ -80,9 +80,11 @@ impl libp2p_core::Transport for Transport {
type ListenerUpgrade = BoxFuture<'static, Result<Self::Output, Self::Error>>;
type Dial = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
let id = ListenerId::new();
fn listen_on(
&mut self,
id: ListenerId,
addr: Multiaddr,
) -> Result<(), TransportError<Self::Error>> {
let socket_addr =
parse_webrtc_listen_addr(&addr).ok_or(TransportError::MultiaddrNotSupported(addr))?;
let udp_mux = UDPMuxNewAddr::listen_on(socket_addr)
@ -93,7 +95,7 @@ impl libp2p_core::Transport for Transport {
.map_err(|e| TransportError::Other(Error::Io(e)))?,
);
Ok(id)
Ok(())
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
@ -596,8 +598,12 @@ mod tests {
// Run test twice to check that there is no unexpected behaviour if `QuicTransport.listener`
// is temporarily empty.
for _ in 0..2 {
let listener = transport
.listen_on("/ip4/0.0.0.0/udp/0/webrtc-direct".parse().unwrap())
let listener = ListenerId::next();
transport
.listen_on(
listener,
"/ip4/0.0.0.0/udp/0/webrtc-direct".parse().unwrap(),
)
.unwrap();
match poll_fn(|cx| Pin::new(&mut transport).as_mut().poll(cx)).await {
TransportEvent::NewAddress {

6
transports/webrtc/tests/smoke.rs
View File

@ -23,7 +23,7 @@ use futures::future::{BoxFuture, Either};
use futures::stream::StreamExt;
use futures::{future, ready, AsyncReadExt, AsyncWriteExt, FutureExt, SinkExt};
use libp2p_core::muxing::{StreamMuxerBox, StreamMuxerExt};
use libp2p_core::transport::{Boxed, TransportEvent};
use libp2p_core::transport::{Boxed, ListenerId, TransportEvent};
use libp2p_core::{Multiaddr, Transport};
use libp2p_identity::PeerId;
use libp2p_webrtc as webrtc;
@ -81,7 +81,9 @@ fn create_transport() -> (PeerId, Boxed<(PeerId, StreamMuxerBox)>) {
}
async fn start_listening(transport: &mut Boxed<(PeerId, StreamMuxerBox)>, addr: &str) -> Multiaddr {
transport.listen_on(addr.parse().unwrap()).unwrap();
transport
.listen_on(ListenerId::next(), addr.parse().unwrap())
.unwrap();
match transport.next().await {
Some(TransportEvent::NewAddress { listen_addr, .. }) => listen_addr,
e => panic!("{e:?}"),

12
transports/websocket/src/framed.rs
View File

@ -122,7 +122,11 @@ where
type ListenerUpgrade = BoxFuture<'static, Result<Self::Output, Self::Error>>;
type Dial = BoxFuture<'static, Result<Self::Output, Self::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>> {
let mut inner_addr = addr.clone();
let proto = match inner_addr.pop() {
Some(p @ Protocol::Wss(_)) => {
@ -139,10 +143,10 @@ where
return Err(TransportError::MultiaddrNotSupported(addr));
}
};
match self.transport.lock().listen_on(inner_addr) {
Ok(id) => {
match self.transport.lock().listen_on(id, inner_addr) {
Ok(()) => {
self.listener_protos.insert(id, proto);
Ok(id)
Ok(())
}
Err(e) => Err(e.map(Error::Transport)),
}

22
transports/websocket/src/lib.rs
View File

@ -64,7 +64,7 @@ use std::{
///
/// ```
/// # use futures::future;
/// # use libp2p_core::Transport;
/// # use libp2p_core::{transport::ListenerId, Transport};
/// # use libp2p_dns as dns;
/// # use libp2p_tcp as tcp;
/// # use libp2p_websocket as websocket;
@ -83,7 +83,7 @@ use std::{
/// let cert = websocket::tls::Certificate::new(rcgen_cert.serialize_der().unwrap());
/// transport.set_tls_config(websocket::tls::Config::new(priv_key, vec![cert]).unwrap());
///
/// let id = transport.listen_on("/ip4/127.0.0.1/tcp/0/wss".parse().unwrap()).unwrap();
/// let id = transport.listen_on(ListenerId::next(), "/ip4/127.0.0.1/tcp/0/wss".parse().unwrap()).unwrap();
///
/// let addr = future::poll_fn(|cx| Pin::new(&mut transport).poll(cx)).await.into_new_address().unwrap();
/// println!("Listening on {addr}");
@ -95,7 +95,7 @@ use std::{
///
/// ```
/// # use futures::future;
/// # use libp2p_core::Transport;
/// # use libp2p_core::{transport::ListenerId, Transport};
/// # use libp2p_dns as dns;
/// # use libp2p_tcp as tcp;
/// # use libp2p_websocket as websocket;
@ -108,7 +108,7 @@ use std::{
/// tcp::async_io::Transport::new(tcp::Config::default()),
/// );
///
/// let id = transport.listen_on("/ip4/127.0.0.1/tcp/0/ws".parse().unwrap()).unwrap();
/// let id = transport.listen_on(ListenerId::next(), "/ip4/127.0.0.1/tcp/0/ws".parse().unwrap()).unwrap();
///
/// let addr = future::poll_fn(|cx| Pin::new(&mut transport).poll(cx)).await.into_new_address().unwrap();
/// println!("Listening on {addr}");
@ -195,8 +195,12 @@ where
type ListenerUpgrade = MapFuture<InnerFuture<T::Output, T::Error>, WrapperFn<T::Output>>;
type Dial = MapFuture<InnerFuture<T::Output, T::Error>, WrapperFn<T::Output>>;
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 {
@ -293,7 +297,7 @@ where
mod tests {
use super::WsConfig;
use futures::prelude::*;
use libp2p_core::{multiaddr::Protocol, Multiaddr, Transport};
use libp2p_core::{multiaddr::Protocol, transport::ListenerId, Multiaddr, Transport};
use libp2p_identity::PeerId;
use libp2p_tcp as tcp;
@ -315,7 +319,9 @@ mod tests {
async fn connect(listen_addr: Multiaddr) {
let mut ws_config = new_ws_config().boxed();
ws_config.listen_on(listen_addr).expect("listener");
ws_config
.listen_on(ListenerId::next(), listen_addr)
.expect("listener");
let addr = ws_config
.next()