core/: Emit Event::Failed on aborted pending connection

core/src/connection/error.rs

@@ -71,6 +71,9 @@ pub enum PendingConnectionError<TTransErr> {
     /// An error occurred while negotiating the transport protocol(s).
     Transport(TransportError<TTransErr>),
 
+    /// Pending connection attempt has been aborted.
+    Aborted,
+
     /// The peer identity obtained on the connection did not
     /// match the one that was expected or is otherwise invalid.
     InvalidPeerId,
@@ -87,6 +90,7 @@ where
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             PendingConnectionError::IO(err) => write!(f, "Pending connection: I/O error: {}", err),
+            PendingConnectionError::Aborted => write!(f, "Pending connection: Aborted."),
             PendingConnectionError::Transport(err) => {
                 write!(f, "Pending connection: Transport error: {}", err)
             }
@@ -106,6 +110,7 @@ where
             PendingConnectionError::IO(err) => Some(err),
             PendingConnectionError::Transport(err) => Some(err),
             PendingConnectionError::InvalidPeerId => None,
+            PendingConnectionError::Aborted => None,
         }
     }
 }

core/src/connection/manager.rs

@@ -20,7 +20,7 @@
 
 use super::{
     handler::{THandlerError, THandlerInEvent, THandlerOutEvent},
-    Connected, ConnectedPoint, Connection, ConnectionError, ConnectionHandler, ConnectionLimit,
+    Connected, ConnectedPoint, ConnectionError, ConnectionHandler, ConnectionLimit,
     IntoConnectionHandler, PendingConnectionError, Substream,
 };
 use crate::{muxing::StreamMuxer, Executor};
@@ -277,40 +277,6 @@ impl<H: IntoConnectionHandler, TE> Manager<H, TE> {
         ConnectionId(task_id)
     }
 
-    /// Adds an existing connection to the manager.
-    pub fn add<M>(&mut self, conn: Connection<M, H::Handler>, info: Connected) -> ConnectionId
-    where
-        H: IntoConnectionHandler + Send + 'static,
-        H::Handler: ConnectionHandler<Substream = Substream<M>> + Send + 'static,
-        <H::Handler as ConnectionHandler>::OutboundOpenInfo: Send + 'static,
-        TE: error::Error + Send + 'static,
-        M: StreamMuxer + Send + Sync + 'static,
-        M::OutboundSubstream: Send + 'static,
-    {
-        let task_id = self.next_task_id;
-        self.next_task_id.0 += 1;
-
-        let (tx, rx) = mpsc::channel(self.task_command_buffer_size);
-        self.tasks.insert(
-            task_id,
-            TaskInfo {
-                sender: tx,
-                state: TaskState::Established(info),
-            },
-        );
-
-        let task: Pin<Box<Task<Pin<Box<future::Pending<_>>>, _, _, _>>> =
-            Box::pin(Task::established(task_id, self.events_tx.clone(), rx, conn));
-
-        if let Some(executor) = &mut self.executor {
-            executor.exec(task);
-        } else {
-            self.local_spawns.push(task);
-        }
-
-        ConnectionId(task_id)
-    }
-
     /// Gets an entry for a managed connection, if it exists.
     pub fn entry(&mut self, id: ConnectionId) -> Option<Entry<'_, THandlerInEvent<H>>> {
         if let hash_map::Entry::Occupied(task) = self.tasks.entry(id.0) {
@@ -496,18 +462,6 @@ impl<'a, I> EstablishedEntry<'a, I> {
         }
     }
 
-    // TODO: Needed?
-    // /// Instantly removes the entry from the manager, dropping
-    // /// the command channel to the background task of the connection,
-    // /// which will thus drop the connection asap without an orderly
-    // /// close or emitting another event.
-    // pub fn remove(self) -> Connected {
-    //     match self.task.remove().state {
-    //         TaskState::Established(c) => c,
-    //         TaskState::Pending => unreachable!("By Entry::new()"),
-    //     }
-    // }
-
     /// Returns the connection ID.
     pub fn id(&self) -> ConnectionId {
         ConnectionId(*self.task.key())

core/src/connection/manager/task.rs

@@ -131,24 +131,6 @@ where
             },
         }
     }
-
-    /// Create a task for an existing node we are already connected to.
-    pub fn established(
-        id: TaskId,
-        events: mpsc::Sender<Event<H, E>>,
-        commands: mpsc::Receiver<Command<THandlerInEvent<H>>>,
-        connection: Connection<M, H::Handler>,
-    ) -> Self {
-        Task {
-            id,
-            events,
-            commands: commands.fuse(),
-            state: State::Established {
-                connection,
-                event: None,
-            },
-        }
-    }
 }
 
 /// The state associated with the `Task` of a connection.
@@ -227,8 +209,16 @@ where
                         Poll::Pending => {}
                         Poll::Ready(None) => {
                             // The manager has dropped the task; abort.
-                            // TODO: Should we return the handler in this case?
-                            return Poll::Ready(());
+                            // Don't accept any further commands and terminate the
+                            // task with a final event.
+                            this.commands.get_mut().close();
+                            let event = Event::Failed {
+                                id,
+                                handler,
+                                error: PendingConnectionError::Aborted,
+                            };
+                            this.state = State::Terminating(event);
+                            continue 'poll;
                         }
                         Poll::Ready(Some(_)) => {
                             panic!("Task received command while the connection is pending.")
@@ -280,9 +270,10 @@ where
                                 continue 'poll;
                             }
                             Poll::Ready(None) => {
+                                todo!("Safe to assume that this should never happen?");
                                 // The manager has dropped the task or disappeared; abort.
                                 // TODO: Should we return the handler in this case?
-                                return Poll::Ready(());
+                                // return Poll::Ready(());
                             }
                         }
                     }

core/src/connection/pool.rs

@@ -387,8 +387,7 @@ impl<THandler: IntoConnectionHandler, TTransErr> Pool<THandler, TTransErr> {
                 if let Some(manager::Entry::Established(e)) = self.manager.entry(id) {
                     e.start_close(None);
                     // TODO: I removed the disconnected logic, thus depending on start_close to
-                    // eventually trigger a ConnectionClosed event. Make sure that is the case and
-                    // also that the num_established counters are kept consistent.
+                    // eventually trigger a ConnectionClosed event. Make sure that is the case.
                 }
                 self.counters.dec_established(endpoint);
             }