refactor: clean up comments and naming

Author: DaleSeo

crates/rmcp/src/transport/streamable_http_client.rs

@@ -282,9 +282,7 @@ impl<C: StreamableHttpClient> StreamableHttpClientWorker<C> {
 
 impl<C: StreamableHttpClient> StreamableHttpClientWorker<C> {
     /// Convert a raw SSE stream into a JSON-RPC message stream without
-    /// reconnection logic.  Used for per-request POST SSE responses where
-    /// we close the stream after the first response and want the underlying
-    /// HTTP connection to be returned to the pool promptly.
+    /// reconnection logic.
     fn raw_sse_to_jsonrpc(
         stream: BoxedSseStream,
     ) -> impl Stream<Item = Result<ServerJsonRpcMessage, StreamableHttpError<C::Error>>> + Send + 'static
@@ -347,11 +345,8 @@ impl<C: StreamableHttpClient> StreamableHttpClientWorker<C> {
             }
             if close_on_response && is_response {
                 tracing::debug!("got response, draining sse stream for connection reuse");
-                // Drain remaining stream bytes so the HTTP/1.1 connection can
-                // be returned to the pool instead of being discarded.  The
-                // server closes the channel shortly after sending the response,
-                // so this normally completes in microseconds on localhost.  The
-                // timeout guards against servers that keep the stream open.
+                // Consume the remaining stream so the HTTP/1.1 connection
+                // returns to the pool cleanly.
                 let _ = tokio::time::timeout(std::time::Duration::from_millis(50), async {
                     while sse_stream.next().await.is_some() {}
                 })
@@ -788,10 +783,6 @@ impl<C: StreamableHttpClient> Worker for StreamableHttpClientWorker<C> {
                             Ok(())
                         }
                         Ok(StreamableHttpPostResponse::Sse(stream, ..)) => {
-                            // Per-request POST SSE streams use a thin
-                            // adapter instead of SseAutoReconnectStream so
-                            // the stream ends immediately when the server
-                            // closes the channel, enabling connection reuse.
                             streams.spawn(Self::execute_sse_stream(
                                 Self::raw_sse_to_jsonrpc(stream),
                                 sse_worker_tx.clone(),

crates/rmcp/tests/test_streamable_http_connection_reuse.rs

@@ -24,11 +24,11 @@ struct SumRequest {
 }
 
 #[derive(Debug, Clone)]
-struct EchoServer {
+struct SumServer {
     tool_router: ToolRouter<Self>,
 }
 
-impl EchoServer {
+impl SumServer {
     fn new() -> Self {
         Self {
             tool_router: Self::tool_router(),
@@ -37,15 +37,15 @@ impl EchoServer {
 }
 
 #[tool_router]
-impl EchoServer {
+impl SumServer {
     #[tool(description = "Sum two numbers")]
     fn sum(&self, Parameters(SumRequest { a, b }): Parameters<SumRequest>) -> String {
         (a + b).to_string()
     }
 }
 
 #[tool_handler(router = self.tool_router)]
-impl ServerHandler for EchoServer {
+impl ServerHandler for SumServer {
     fn get_info(&self) -> ServerInfo {
         ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
     }
@@ -59,14 +59,13 @@ impl ServerHandler for EchoServer {
 async fn test_subsequent_tool_calls_reuse_connections() -> anyhow::Result<()> {
     let ct = CancellationToken::new();
 
-    let service: StreamableHttpService<EchoServer, LocalSessionManager> =
-        StreamableHttpService::new(
-            || Ok(EchoServer::new()),
-            Default::default(),
-            StreamableHttpServerConfig::default()
-                .with_sse_keep_alive(None)
-                .with_cancellation_token(ct.child_token()),
-        );
+    let service: StreamableHttpService<SumServer, LocalSessionManager> = StreamableHttpService::new(
+        || Ok(SumServer::new()),
+        Default::default(),
+        StreamableHttpServerConfig::default()
+            .with_sse_keep_alive(None)
+            .with_cancellation_token(ct.child_token()),
+    );
 
     let router = axum::Router::new().nest_service("/mcp", service);
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;