transport: Remove start event loop and rely on start

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

Author: lexnv

src/transport/manager/mod.rs

@@ -1107,10 +1107,7 @@ impl TransportManager {
                             "connection accepted and protocols notified",
                         );
 
-                        return Some(TransportEvent::ConnectionEstablished {
-                            peer,
-                            endpoint,
-                        });
+                        return Some(TransportEvent::ConnectionEstablished { peer, endpoint });
                     }
                     Err(error) => {
                         tracing::debug!(

src/transport/mod.rs

@@ -199,7 +199,10 @@ pub(crate) trait Transport: Stream + Unpin + Send {
     /// and all installed protocols have been notified via their event channels.
     /// This ensures that by the time the caller receives a ConnectionEstablished event,
     /// protocols are ready to handle substream operations.
-    fn accept(&mut self, connection_id: ConnectionId) -> crate::Result<BoxFuture<'static, crate::Result<()>>>;
+    fn accept(
+        &mut self,
+        connection_id: ConnectionId,
+    ) -> crate::Result<BoxFuture<'static, crate::Result<()>>>;
 
     /// Accept pending connection.
     fn accept_pending(&mut self, connection_id: ConnectionId) -> crate::Result<()>;

src/transport/quic/connection.rs

@@ -231,18 +231,9 @@ impl QuicConnection {
         })
     }
 
-    /// Start event loop for [`QuicConnection`].
-    pub async fn start(mut self) -> crate::Result<()> {
-        self.protocol_set
-            .report_connection_established(self.peer, self.endpoint.clone())
-            .await?;
-
-        self.start_event_loop().await
-    }
-
     /// Start the connection event loop without notifying protocols.
     /// This is used when protocols have already been notified during accept().
-    pub(crate) async fn start_event_loop(mut self) -> crate::Result<()> {
+    pub(crate) async fn start(mut self) -> crate::Result<()> {
         loop {
             tokio::select! {
                 event = self.connection.accept_bi() => match event {

src/transport/quic/mod.rs

@@ -306,7 +306,10 @@ impl Transport for QuicTransport {
         Ok(())
     }
 
-    fn accept(&mut self, connection_id: ConnectionId) -> crate::Result<BoxFuture<'static, crate::Result<()>>> {
+    fn accept(
+        &mut self,
+        connection_id: ConnectionId,
+    ) -> crate::Result<BoxFuture<'static, crate::Result<()>>> {
         let (connection, endpoint) = self
             .pending_open
             .remove(&connection_id)
@@ -327,9 +330,7 @@ impl Transport for QuicTransport {
 
         Ok(Box::pin(async move {
             // First, notify all protocols about the connection establishment
-            protocol_set
-                .report_connection_established(peer, endpoint_clone)
-                .await?;
+            protocol_set.report_connection_established(peer, endpoint_clone).await?;
 
             // After protocols are notified, spawn the connection event loop
             executor.run(Box::pin(async move {
@@ -341,7 +342,7 @@ impl Transport for QuicTransport {
                     bandwidth_sink,
                     substream_open_timeout,
                 )
-                .start_event_loop()
+                .start()
                 .await;
             }));
 

src/transport/tcp/connection.rs

@@ -730,18 +730,9 @@ impl TcpConnection {
         }
     }
 
-    /// Start connection event loop.
-    pub(crate) async fn start(mut self) -> crate::Result<()> {
-        self.protocol_set
-            .report_connection_established(self.peer, self.endpoint.clone())
-            .await?;
-
-        self.start_event_loop().await
-    }
-
     /// Start the connection event loop without notifying protocols.
     /// This is used when protocols have already been notified during accept().
-    pub(crate) async fn start_event_loop(mut self) -> crate::Result<()> {
+    pub(crate) async fn start(mut self) -> crate::Result<()> {
         loop {
             tokio::select! {
                 substream = self.connection.next() => {

src/transport/tcp/mod.rs

@@ -389,7 +389,10 @@ impl Transport for TcpTransport {
         )
     }
 
-    fn accept(&mut self, connection_id: ConnectionId) -> crate::Result<BoxFuture<'static, crate::Result<()>>> {
+    fn accept(
+        &mut self,
+        connection_id: ConnectionId,
+    ) -> crate::Result<BoxFuture<'static, crate::Result<()>>> {
         let context = self
             .pending_open
             .remove(&connection_id)
@@ -411,15 +414,13 @@ impl Transport for TcpTransport {
         Ok(Box::pin(async move {
             // First, notify all protocols about the connection establishment
             // This ensures that when the accept() future completes, protocols are ready
-            protocol_set
-                .report_connection_established(peer, endpoint)
-                .await?;
+            protocol_set.report_connection_established(peer, endpoint).await?;
 
             // After protocols are notified, spawn the connection event loop
             executor.run(Box::pin(async move {
                 if let Err(error) =
                     TcpConnection::new(context, protocol_set, bandwidth_sink, next_substream_id)
-                        .start_event_loop()
+                        .start()
                         .await
                 {
                     tracing::debug!(

src/transport/websocket/connection.rs

@@ -453,18 +453,9 @@ impl WebSocketConnection {
         })
     }
 
-    /// Start connection event loop.
-    pub(crate) async fn start(mut self) -> crate::Result<()> {
-        self.protocol_set
-            .report_connection_established(self.peer, self.endpoint.clone())
-            .await?;
-
-        self.start_event_loop().await
-    }
-
     /// Start the connection event loop without notifying protocols.
     /// This is used when protocols have already been notified during accept().
-    pub(crate) async fn start_event_loop(mut self) -> crate::Result<()> {
+    pub(crate) async fn start(mut self) -> crate::Result<()> {
         loop {
             tokio::select! {
                 substream = self.connection.next() => match substream {

src/transport/websocket/mod.rs

@@ -398,7 +398,10 @@ impl Transport for WebSocketTransport {
         Ok(())
     }
 
-    fn accept(&mut self, connection_id: ConnectionId) -> crate::Result<BoxFuture<'static, crate::Result<()>>> {
+    fn accept(
+        &mut self,
+        connection_id: ConnectionId,
+    ) -> crate::Result<BoxFuture<'static, crate::Result<()>>> {
         let context = self
             .pending_open
             .remove(&connection_id)
@@ -419,9 +422,7 @@ impl Transport for WebSocketTransport {
 
         Ok(Box::pin(async move {
             // First, notify all protocols about the connection establishment
-            protocol_set
-                .report_connection_established(peer, endpoint)
-                .await?;
+            protocol_set.report_connection_established(peer, endpoint).await?;
 
             // After protocols are notified, spawn the connection event loop
             executor.run(Box::pin(async move {
@@ -431,7 +432,7 @@ impl Transport for WebSocketTransport {
                     bandwidth_sink,
                     substream_open_timeout,
                 )
-                .start_event_loop()
+                .start()
                 .await
                 {
                     tracing::debug!(