Use new dialer trait in network and swarm.

dvc94ch · dvc94ch · commit e8dbdecab3f9 · 2020-10-09T19:38:51.000+02:00
diff --git a/core/src/connection/listeners.rs b/core/src/connection/listeners.rs
@@ -20,7 +20,7 @@
 
 //! Manage listening on multiple multiaddresses at once.
 
-use crate::{Multiaddr, Transport, transport::{TransportError, ListenerEvent}};
+use crate::{address_translation, Dialer, Multiaddr, Transport, transport::{TransportError, ListenerEvent}};
 use futures::{prelude::*, task::Context, task::Poll};
 use log::debug;
 use smallvec::SmallVec;
@@ -232,6 +232,42 @@ where
         self.listeners.iter().flat_map(|l| l.addresses.iter())
     }
 
+    /// Perform address translation.
+    pub fn address_translation(&self, observed_addr: &Multiaddr) -> Vec<Multiaddr> {
+        let mut addrs = Vec::with_capacity(4 * self.listeners.len());
+        for listener in &self.listeners {
+            if listener.dialer.requires_address_translation() {
+                for addr in &listener.addresses {
+                    if let Some(new_addr) = address_translation(addr, observed_addr) {
+                        if addrs.iter().find(|addr| *addr == &new_addr).is_none() {
+                            addrs.push(new_addr);
+                        }
+                    }
+                }
+            } else {
+                if addrs.iter().find(|addr| *addr == observed_addr).is_none() {
+                    addrs.push(observed_addr.clone());
+                }
+            }
+        }
+        addrs
+    }
+
+    /// Returns a dialer for an address.
+    pub fn dialer_for_addr(&self, addr: &Multiaddr) -> TTrans::Dialer {
+        if self.listen_addrs().find(|addr2| *addr2 == addr).is_some() {
+            return self.transport.dialer();
+        }
+
+        self.listeners.iter().filter_map(|listener| {
+            listener.addresses.iter().find(|address| {
+                address.can_dial(addr)
+            }).map(|_| listener.dialer.clone())
+        })
+        .next()
+        .unwrap_or_else(|| self.transport.dialer())
+    }
+
     /// Provides an API similar to `Stream`, except that it cannot end.
     pub fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<ListenersEvent<TTrans>> {
         // We remove each element from `listeners` one by one and add them back.
diff --git a/core/src/network.rs b/core/src/network.rs
@@ -29,7 +29,6 @@ use crate::{
     Executor,
     Multiaddr,
     PeerId,
-    address_translation,
     connection::{
         ConnectionId,
         ConnectionLimit,
@@ -199,21 +198,8 @@ where
     /// * `observed_addr` - should be an address a remote observes you as, which can be obtained for
     /// example with the identify protocol.
     ///
-    pub fn address_translation<'a>(&'a self, observed_addr: &'a Multiaddr)
-        -> impl Iterator<Item = Multiaddr> + 'a
-    where
-        TMuxer: 'a,
-        THandler: 'a,
-    {
-        let mut addrs: Vec<_> = self.listen_addrs()
-            .filter_map(move |server| address_translation(server, observed_addr))
-            .collect();
-
-        // remove duplicates
-        addrs.sort_unstable();
-        addrs.dedup();
-
-        addrs.into_iter()
+    pub fn address_translation(&self, observed_addr: &Multiaddr) -> Vec<Multiaddr> {
+        self.listeners.address_translation(observed_addr)
     }
 
     /// Returns the peer id of the local node.
@@ -240,7 +226,7 @@ where
         TPeerId: Send + 'static,
     {
         let info = OutgoingInfo { address, peer_id: None };
-        match self.transport().dialer().dial(address.clone()) {
+        match self.listeners.dialer_for_addr(address).dial(address.clone()) {
             Ok(f) => {
                 let f = f.map_err(|err| PendingConnectionError::Transport(TransportError::Other(err)));
                 self.pool.add_outgoing(f, handler, info)
@@ -435,8 +421,8 @@ where
                 let dialing = &mut self.dialing;
                 let (next, event) = on_connection_failed(dialing, id, endpoint, error, handler);
                 if let Some(dial) = next {
-                    let transport = self.listeners.transport().clone();
-                    if let Err(e) = dial_peer_impl(transport, pool, dialing, dial) {
+                    let dialer = self.listeners.dialer_for_addr(&dial.address);
+                    if let Err(e) = dial_peer_impl(dialer, pool, dialing, dial) {
                         log::warn!("Dialing aborted: {:?}", e);
                     }
                 }
@@ -481,7 +467,8 @@ where
         TOutEvent: Send + 'static,
         TPeerId: Send + 'static,
     {
-        dial_peer_impl(self.transport().clone(), &mut self.pool, &mut self.dialing, opts)
+        let dialer = self.listeners.dialer_for_addr(&opts.address);
+        dial_peer_impl(dialer, &mut self.pool, &mut self.dialing, opts)
     }
 }
 
@@ -495,9 +482,9 @@ struct DialingOpts<TPeerId, THandler> {
 }
 
 /// Standalone implementation of `Network::dial_peer` for more granular borrowing.
-fn dial_peer_impl<TMuxer, TInEvent, TOutEvent, THandler, TTrans, TConnInfo, TPeerId>(
-    transport: TTrans,
-    pool: &mut Pool<TInEvent, TOutEvent, THandler, TTrans::Error,
+fn dial_peer_impl<TMuxer, TInEvent, TOutEvent, THandler, TDialer, TConnInfo, TPeerId>(
+    dialer: TDialer,
+    pool: &mut Pool<TInEvent, TOutEvent, THandler, TDialer::Error,
         <THandler::Handler as ConnectionHandler>::Error, TConnInfo, TPeerId>,
     dialing: &mut FnvHashMap<TPeerId, SmallVec<[peer::DialingState; 10]>>,
     opts: DialingOpts<TPeerId, THandler>
@@ -511,17 +498,17 @@ where
         InEvent = TInEvent,
         OutEvent = TOutEvent,
     > + Send + 'static,
-    TTrans: Transport<Output = (TConnInfo, TMuxer)>,
-    TTrans::Dial: Send + 'static,
-    TTrans::Error: error::Error + Send + 'static,
+    TDialer: Dialer<Output = (TConnInfo, TMuxer)>,
+    TDialer::Dial: Send + 'static,
+    TDialer::Error: error::Error + Send + 'static,
     TMuxer: StreamMuxer + Send + Sync + 'static,
     TMuxer::OutboundSubstream: Send + 'static,
     TInEvent: Send + 'static,
     TOutEvent: Send + 'static,
     TPeerId: Eq + Hash + Send + Clone + 'static,
     TConnInfo: ConnectionInfo<PeerId = TPeerId> + Send + 'static,
 {
-    let result = match transport.dialer().dial(opts.address.clone()) {
+    let result = match dialer.dial(opts.address.clone()) {
         Ok(fut) => {
             let fut = fut.map_err(|e| PendingConnectionError::Transport(TransportError::Other(e)));
             let info = OutgoingInfo { address: &opts.address, peer_id: Some(&opts.peer) };
diff --git a/core/src/transport.rs b/core/src/transport.rs
@@ -49,7 +49,7 @@ pub use self::memory::MemoryTransport;
 pub use self::optional::OptionalTransport;
 pub use self::upgrade::Upgrade;
 
-pub trait Dialer {
+pub trait Dialer: Clone {
     /// The result of a connection setup process, including protocol upgrades.
     ///
     /// Typically the output contains at least a handle to a data stream (i.e. a
@@ -71,6 +71,8 @@ pub trait Dialer {
     fn dial(self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>>
     where
         Self: Sized;
+
+    fn requires_address_translation(&self) -> bool { false }
 }
 
 /// A transport provides connection-oriented communication between two peers
diff --git a/core/tests/network_dial_error.rs b/core/tests/network_dial_error.rs
@@ -59,6 +59,85 @@ fn new_network(cfg: NetworkConfig) -> TestNetwork {
     TestNetwork::new(transport, local_public_key.into(), cfg)
 }
 
+#[test]
+fn port_reuse() {
+    // Checks whether reusing a port on a swarm works.
+
+    let mut swarm1 = new_network(NetworkConfig::default());
+    let mut swarm2 = new_network(NetworkConfig::default());
+
+    swarm1.listen_on("/ip4/127.0.0.1/tcp/0".parse().unwrap()).unwrap();
+    swarm2.listen_on("/ip4/127.0.0.1/tcp/0".parse().unwrap()).unwrap();
+
+    let address1 = async_std::task::block_on(future::poll_fn(|cx| {
+        if let Poll::Ready(NetworkEvent::NewListenerAddress { listen_addr, .. }) = swarm1.poll(cx) {
+            Poll::Ready(listen_addr)
+        } else {
+            panic!("Was expecting the listen address to be reported")
+        }
+    }));
+
+    let address2 = async_std::task::block_on(future::poll_fn(|cx| {
+        if let Poll::Ready(NetworkEvent::NewListenerAddress { listen_addr, .. }) = swarm2.poll(cx) {
+            Poll::Ready(listen_addr)
+        } else {
+            panic!("Was expecting the listen address to be reported")
+        }
+    }));
+
+    swarm2
+        .peer(swarm1.local_peer_id().clone())
+        .dial(address1.clone(), Vec::new(), TestHandler())
+        .unwrap();
+
+    let mut got_incoming = false;
+    let mut got_outgoing = false;
+    async_std::task::block_on(future::poll_fn(|cx| -> Poll<Result<(), io::Error>> {
+        loop {
+            match swarm1.poll(cx) {
+                Poll::Ready(NetworkEvent::IncomingConnection { connection, .. }) => {
+                    assert_eq!(&connection.local_addr, &address1);
+                    assert_eq!(&connection.send_back_addr, &address2);
+                    swarm1.accept(connection, TestHandler()).unwrap();
+                },
+                Poll::Ready(NetworkEvent::ConnectionEstablished {
+                    connection,
+                    ..
+                }) => {
+                    assert_eq!(connection.peer_id(), swarm2.local_peer_id());
+                    got_incoming = true;
+                    if got_outgoing {
+                        return Poll::Ready(Ok(()));
+                    }
+                },
+                Poll::Ready(NetworkEvent::ConnectionClosed { .. }) => continue,
+                Poll::Ready(e) => panic!("{:?}", e),
+                Poll::Pending => break,
+            }
+        }
+
+        loop {
+            match swarm2.poll(cx) {
+                Poll::Ready(NetworkEvent::ConnectionEstablished {
+                    connection,
+                    ..
+                }) => {
+                    assert_eq!(connection.peer_id(), swarm1.local_peer_id());
+                    got_outgoing = true;
+                    if got_incoming {
+                        return Poll::Ready(Ok(()));
+                    }
+                },
+                Poll::Ready(NetworkEvent::ConnectionClosed { .. }) => continue,
+                Poll::Ready(e) => panic!("{:?}", e),
+                Poll::Pending => break,
+            }
+        }
+
+        Poll::Pending
+    })).unwrap();
+}
+
 #[test]
 fn deny_incoming_connec() {
     // Checks whether refusing an incoming connection on a swarm triggers the correct events.
diff --git a/transports/dns/src/lib.rs b/transports/dns/src/lib.rs
@@ -303,6 +303,7 @@ mod tests {
             }
         }
 
+        #[derive(Clone)]
         struct CustomDialer;
 
         impl Dialer for CustomDialer {
diff --git a/transports/tcp/src/lib.rs b/transports/tcp/src/lib.rs
@@ -243,6 +243,10 @@ impl Dialer for TcpDialer {
 
         Ok(Box::pin(self.do_dial(socket_addr)))
     }
+
+    fn requires_address_translation(&self) -> bool {
+        self.local_socket_addr.is_none()
+    }
 }
 
 impl Transport for TcpConfig {

Original file line number	Diff line number	Diff line change
`@@ -303,6 +303,7 @@ mod tests {`
`303`	`303`	`}`
`304`	`304`	`}`
`305`	`305`
	`306`	`+ #[derive(Clone)]`
`306`	`307`	`struct CustomDialer;`
`307`	`308`
`308`	`309`	`impl Dialer for CustomDialer {`
Original file line number	Diff line number	Diff line change
`@@ -243,6 +243,10 @@ impl Dialer for TcpDialer {`
`243`	`243`
`244`	`244`	`Ok(Box::pin(self.do_dial(socket_addr)))`
`245`	`245`	`}`
	`246`	`+`
	`247`	`+ fn requires_address_translation(&self) -> bool {`
	`248`	`+ self.local_socket_addr.is_none()`
	`249`	`+ }`
`246`	`250`	`}`
`247`	`251`
`248`	`252`	`impl Transport for TcpConfig {`