Close Channels on Disconnect

russh/src/client/mod.rs

@@ -882,9 +882,10 @@ impl Session {
             }
         }
         debug!("disconnected");
-        if self.common.disconnected {
-            stream_write.shutdown().await.map_err(crate::Error::from)?;
-        }
+        self.receiver.close();
+        self.inbound_channel_receiver.close();
+        stream_write.shutdown().await.map_err(crate::Error::from)?;
+
         Ok(())
     }
 

russh/src/lib.rs

@@ -465,3 +465,11 @@ impl ChannelParams {
         self.confirmed = true;
     }
 }
+
+pub(crate) async fn timeout(delay: Option<std::time::Duration>) {
+    if let Some(delay) = delay {
+        tokio::time::sleep(delay).await
+    } else {
+        futures::future::pending().await
+    };
+}

russh/src/server/encrypted.rs

@@ -46,7 +46,7 @@ impl Session {
         // Either this packet is a KEXINIT, in which case we start a key re-exchange.
 
         #[allow(clippy::unwrap_used)]
-        let mut enc = self.common.encrypted.as_mut().unwrap();
+        let enc = self.common.encrypted.as_mut().unwrap();
         if buf.first() == Some(&msg::KEXINIT) {
             debug!("Received rekeying request");
             // If we're not currently rekeying, but `buf` is a rekey request
@@ -142,7 +142,7 @@ impl Session {
         };
 
         #[allow(clippy::unwrap_used)]
-        let mut enc = self.common.encrypted.as_mut().unwrap();
+        let enc = self.common.encrypted.as_mut().unwrap();
         // If we've successfully read a packet.
         match enc.state {
             EncryptedState::WaitingAuthServiceRequest {