Fix tests and move tests::util::TestHandler to connection::util::Test…

…Handler

core/src/connection.rs

@@ -22,7 +22,7 @@ mod error;
 pub(crate) mod handler;
 mod listeners;
 mod substream;
-
+pub mod util;
 pub(crate) mod pool;
 
 pub use error::{

core/src/connection/pool.rs

@@ -969,7 +969,12 @@ impl<THandler: IntoConnectionHandler> PendingConnection<'_, THandler> {
     /// Aborts the connection attempt, closing the connection.
     pub fn abort(mut self) {
         if let Some(notifier) = self.entry.get_mut().abort_notifier.take() {
-            notifier.send(task::PendingConnectionCommand::Abort);
+            match notifier.send(task::PendingConnectionCommand::Abort) {
+                Ok(()) => {}
+                Err(e) => {
+                    log::debug!("Failed to pending connection: {:?}", e);
+                }
+            }
         }
     }
 }
@@ -1357,3 +1362,49 @@ impl<'a, K: 'a, V: 'a> EntryExt<'a, K, V> for hash_map::Entry<'a, K, V> {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use crate::{
+        connection::PendingOutboundConnectionError,
+        transport::dummy::DummyTransport,
+    };
+    use futures::{executor::block_on, future::poll_fn};
+
+    #[test]
+    fn aborting_pending_connection_surfaces_error() {
+        let mut pool = Pool::new(
+            PeerId::random(),
+            PoolConfig::default(),
+            ConnectionLimits::default(),
+        );
+        let target_peer = PeerId::random();
+
+        pool.add_outgoing(
+            DummyTransport::default(),
+            ["/ip4/127.0.0.1/tcp/1234".parse::<Multiaddr>().unwrap()].into_iter(),
+            Some(target_peer),
+            crate::connection::util::TestHandler(),
+        )
+        .expect("Failed to add an outgoing connection to test pool.");
+
+        // Disconnect from the peer, thus aborting all pending connections.
+        pool.disconnect(&target_peer);
+
+        // Poll the pool until the pending connection is aborted.
+        block_on(poll_fn(|cx| match pool.poll(cx) {
+            Poll::Pending => return Poll::Pending,
+            Poll::Ready(PoolEvent::PendingOutboundConnectionError {
+                error: PendingOutboundConnectionError::Aborted,
+                ..
+            }) => {
+                return Poll::Ready(());
+            }
+            Poll::Ready(_) => {
+                panic!("We should see an aborted error, nothing else.")
+            }
+        }));
+    }
+}

core/src/connection/util.rs

@@ -0,0 +1,38 @@
+use std::{io, task::{Poll, Context}};
+
+use multiaddr::Multiaddr;
+
+use crate::muxing::StreamMuxerBox;
+
+use super::{ConnectionHandler, Substream, SubstreamEndpoint, ConnectionHandlerEvent};
+
+
+#[derive(Debug)]
+pub struct TestHandler();
+
+impl ConnectionHandler for TestHandler {
+    type InEvent = ();
+    type OutEvent = ();
+    type Error = io::Error;
+    type Substream = Substream<StreamMuxerBox>;
+    type OutboundOpenInfo = ();
+
+    fn inject_substream(
+        &mut self,
+        _: Self::Substream,
+        _: SubstreamEndpoint<Self::OutboundOpenInfo>,
+    ) {
+    }
+
+    fn inject_event(&mut self, _: Self::InEvent) {}
+
+    fn inject_address_change(&mut self, _: &Multiaddr) {}
+
+    fn poll(
+        &mut self,
+        _: &mut Context<'_>,
+    ) -> Poll<Result<ConnectionHandlerEvent<Self::OutboundOpenInfo, Self::OutEvent>, Self::Error>>
+    {
+        Poll::Pending
+    }
+}

core/tests/concurrent_dialing.rs

@@ -27,12 +27,13 @@ use libp2p_core::{
     multiaddr::Protocol,
     network::{NetworkConfig, NetworkEvent},
     ConnectedPoint,
+    connection::util::TestHandler,
 };
 use quickcheck::*;
 use rand07::Rng;
 use std::num::NonZeroU8;
 use std::task::Poll;
-use util::{test_network, TestHandler};
+use util::{test_network};
 
 #[test]
 fn concurrent_dialing() {

core/tests/connection_limits.rs

@@ -23,13 +23,13 @@ mod util;
 use futures::{future::poll_fn, ready};
 use libp2p_core::multiaddr::{multiaddr, Multiaddr};
 use libp2p_core::{
-    connection::PendingConnectionError,
+    connection::{PendingConnectionError, util::TestHandler},
     network::{ConnectionLimits, DialError, NetworkConfig, NetworkEvent},
     PeerId,
 };
 use quickcheck::*;
 use std::task::Poll;
-use util::{test_network, TestHandler};
+use util::test_network;
 
 #[test]
 fn max_outgoing() {

core/tests/network_dial_error.rs

@@ -23,14 +23,14 @@ mod util;
 use futures::prelude::*;
 use libp2p_core::multiaddr::multiaddr;
 use libp2p_core::{
-    connection::PendingConnectionError,
+    connection::{PendingConnectionError, util::TestHandler},
     multiaddr::Protocol,
     network::{NetworkConfig, NetworkEvent},
     PeerId,
 };
 use rand::seq::SliceRandom;
 use std::{io, task::Poll};
-use util::{test_network, TestHandler};
+use util::test_network;
 
 #[test]
 fn deny_incoming_connec() {

core/tests/util.rs

@@ -2,7 +2,7 @@
 
 use futures::prelude::*;
 use libp2p_core::{
-    connection::{ConnectionHandler, ConnectionHandlerEvent, Substream, SubstreamEndpoint},
+    connection::util::TestHandler,
     identity,
     muxing::{StreamMuxer, StreamMuxerBox},
     network::{Network, NetworkConfig},
@@ -32,36 +32,6 @@ pub fn test_network(cfg: NetworkConfig) -> TestNetwork {
     TestNetwork::new(transport, local_public_key.into(), cfg)
 }
 
-#[derive(Debug)]
-pub struct TestHandler();
-
-impl ConnectionHandler for TestHandler {
-    type InEvent = ();
-    type OutEvent = ();
-    type Error = io::Error;
-    type Substream = Substream<StreamMuxerBox>;
-    type OutboundOpenInfo = ();
-
-    fn inject_substream(
-        &mut self,
-        _: Self::Substream,
-        _: SubstreamEndpoint<Self::OutboundOpenInfo>,
-    ) {
-    }
-
-    fn inject_event(&mut self, _: Self::InEvent) {}
-
-    fn inject_address_change(&mut self, _: &Multiaddr) {}
-
-    fn poll(
-        &mut self,
-        _: &mut Context<'_>,
-    ) -> Poll<Result<ConnectionHandlerEvent<Self::OutboundOpenInfo, Self::OutEvent>, Self::Error>>
-    {
-        Poll::Pending
-    }
-}
-
 pub struct CloseMuxer<M> {
     state: CloseMuxerState<M>,
 }