General clean-up of the single stream connections code

src/libp2p/collection/multi_stream.rs

@@ -168,6 +168,15 @@ where
                 outbound_substreams_reverse,
             } => {
                 let event = match established.pull_event() {
+                    Some(established::Event::NewOutboundSubstreamsForbidden) => {
+                        // TODO: handle properly
+                        self.connection = MultiStreamConnectionTaskInner::ShutdownWaitingAck {
+                            start_shutdown_message_to_send: Some(None),
+                            shutdown_finish_message_sent: false,
+                            initiator: ShutdownInitiator::Coordinator,
+                        };
+                        Some(ConnectionToCoordinatorInner::StartShutdown(None))
+                    }
                     Some(established::Event::InboundError(err)) => {
                         Some(ConnectionToCoordinatorInner::InboundError(err))
                     }

src/libp2p/collection/single_stream.rs

@@ -488,6 +488,20 @@ where
                     }
 
                     match event {
+                        Some(established::Event::NewOutboundSubstreamsForbidden) => {
+                            // TODO: handle properly
+                            self.pending_messages.push_back(
+                                ConnectionToCoordinatorInner::StartShutdown(Some(
+                                    ShutdownCause::CleanShutdown,
+                                )),
+                            );
+                            self.pending_messages
+                                .push_back(ConnectionToCoordinatorInner::ShutdownFinished);
+                            self.connection = SingleStreamConnectionTaskInner::ShutdownWaitingAck {
+                                initiator: ShutdownInitiator::Remote,
+                            };
+                            return;
+                        }
                         Some(established::Event::InboundError(err)) => {
                             self.pending_messages
                                 .push_back(ConnectionToCoordinatorInner::InboundError(err));

src/libp2p/connection/established.rs

@@ -69,6 +69,11 @@ impl SubstreamId {
 #[must_use]
 #[derive(Debug)]
 pub enum Event<TRqUd, TNotifUd> {
+    /// The connection is now in a mode where opening new substreams (i.e. starting requests
+    /// and opening notifications substreams) is forbidden, but the remote is still able to open
+    /// substreams and messages on existing substreams are still allowed to be sent and received.
+    NewOutboundSubstreamsForbidden,
+
     /// Received an incoming substream, but this substream has produced an error.
     ///
     /// > **Note**: This event exists only for diagnostic purposes. No action is expected in