websocket: Decrement metrics on connection closed

Signed-off-by: Alexandru Vasile <[email protected]>

Author: lexnv

src/transport/websocket/connection.rs

@@ -436,8 +436,7 @@ impl WebSocketConnection {
         })
     }
 
-    /// Start connection event loop.
-    pub(crate) async fn start(mut self) -> crate::Result<()> {
+    async fn start_inner(mut self) -> crate::Result<()> {
         self.protocol_set
             .report_connection_established(self.peer, self.endpoint)
             .await?;
@@ -602,4 +601,14 @@ impl WebSocketConnection {
             }
         }
     }
+
+    /// Start connection event loop.
+    pub(crate) async fn start(mut self) -> crate::Result<()> {
+        self.start_inner().await.inspect_err(|_| {
+            if let Some(metrics) = &self.metrics {
+                // All pending substreams must be decremented because the connection is closing.
+                metrics.pending_substreams_num.sub(self.pending_substreams.len() as u64);
+            }
+        })
+    }
 }