refactor: remove ConnectStreamingRequest && update document

Author: washanhanzi

connectrpc-axum/src/handler.rs

@@ -12,7 +12,7 @@ use crate::{
         validate_unary_content_type,
     },
     error::ConnectError,
-    message::{ConnectRequest, ConnectResponse, ConnectStreamingRequest, StreamBody, Streaming},
+    message::{ConnectRequest, ConnectResponse, StreamBody, Streaming},
 };
 use futures::Stream;
 use prost::Message;
@@ -619,10 +619,10 @@ pub struct ConnectClientStreamHandlerWrapper<F>(pub F);
 /// Type alias for compatibility with generated code
 pub type ConnectClientStreamHandler<F> = ConnectClientStreamHandlerWrapper<F>;
 
-impl<F, Fut, Req, Resp> Handler<(ConnectStreamingRequest<Req>,), ()>
+impl<F, Fut, Req, Resp> Handler<(ConnectRequest<Streaming<Req>>,), ()>
     for ConnectClientStreamHandlerWrapper<F>
 where
-    F: Fn(ConnectStreamingRequest<Req>) -> Fut + Clone + Send + Sync + 'static,
+    F: Fn(ConnectRequest<Streaming<Req>>) -> Fut + Clone + Send + Sync + 'static,
     Fut: Future<Output = Result<ConnectResponse<Resp>, ConnectError>> + Send + 'static,
     Req: Message + DeserializeOwned + Default + Send + 'static,
     Resp: Message + serde::Serialize + Send + Clone + Sync + 'static,
@@ -644,10 +644,11 @@ where
             }
 
             // Extract the streaming request
-            let streaming_req = match ConnectStreamingRequest::<Req>::from_request(req, &()).await {
-                Ok(value) => value,
-                Err(err) => return err.into_response(),
-            };
+            let streaming_req =
+                match ConnectRequest::<Streaming<Req>>::from_request(req, &()).await {
+                    Ok(value) => value,
+                    Err(err) => return err.into_response(),
+                };
 
             // Call the handler function
             // Note: Timeout is enforced by ConnectLayer, not here
@@ -672,7 +673,7 @@ where
 /// Client streaming: client sends a stream of messages, server responds with one message.
 pub fn post_client_stream<F, Req, Resp, Fut>(f: F) -> MethodRouter<()>
 where
-    F: Fn(ConnectStreamingRequest<Req>) -> Fut + Clone + Send + Sync + 'static,
+    F: Fn(ConnectRequest<Streaming<Req>>) -> Fut + Clone + Send + Sync + 'static,
     Fut: Future<Output = Result<ConnectResponse<Resp>, ConnectError>> + Send + 'static,
     Req: Message + DeserializeOwned + Default + Send + 'static,
     Resp: Message + serde::Serialize + Send + Clone + Sync + 'static,
@@ -694,10 +695,10 @@ pub struct ConnectBidiStreamHandlerWrapper<F>(pub F);
 /// Type alias for compatibility with generated code
 pub type ConnectBidiStreamHandler<F> = ConnectBidiStreamHandlerWrapper<F>;
 
-impl<F, Fut, Req, Resp, St> Handler<(ConnectStreamingRequest<Req>,), ()>
+impl<F, Fut, Req, Resp, St> Handler<(ConnectRequest<Streaming<Req>>,), ()>
     for ConnectBidiStreamHandlerWrapper<F>
 where
-    F: Fn(ConnectStreamingRequest<Req>) -> Fut + Clone + Send + Sync + 'static,
+    F: Fn(ConnectRequest<Streaming<Req>>) -> Fut + Clone + Send + Sync + 'static,
     Fut: Future<Output = Result<ConnectResponse<StreamBody<St>>, ConnectError>> + Send + 'static,
     St: Stream<Item = Result<Resp, ConnectError>> + Send + 'static,
     Req: Message + DeserializeOwned + Default + Send + 'static,
@@ -720,10 +721,11 @@ where
             }
 
             // Extract the streaming request
-            let streaming_req = match ConnectStreamingRequest::<Req>::from_request(req, &()).await {
-                Ok(value) => value,
-                Err(err) => return err.into_response(),
-            };
+            let streaming_req =
+                match ConnectRequest::<Streaming<Req>>::from_request(req, &()).await {
+                    Ok(value) => value,
+                    Err(err) => return err.into_response(),
+                };
 
             // Call the handler function
             // Note: Timeout is enforced by ConnectLayer, not here
@@ -748,7 +750,7 @@ where
 /// Requires HTTP/2 for full-duplex communication.
 pub fn post_bidi_stream<F, Req, Resp, St, Fut>(f: F) -> MethodRouter<()>
 where
-    F: Fn(ConnectStreamingRequest<Req>) -> Fut + Clone + Send + Sync + 'static,
+    F: Fn(ConnectRequest<Streaming<Req>>) -> Fut + Clone + Send + Sync + 'static,
     Fut: Future<Output = Result<ConnectResponse<StreamBody<St>>, ConnectError>> + Send + 'static,
     St: Stream<Item = Result<Resp, ConnectError>> + Send + 'static,
     Req: Message + DeserializeOwned + Default + Send + 'static,

connectrpc-axum/src/lib.rs

@@ -80,8 +80,6 @@ pub mod prelude {
         ConnectStreamHandlerWrapper,
     };
     pub use crate::layer::{ConnectLayer, ConnectService};
-    pub use crate::message::{
-        ConnectRequest, ConnectResponse, ConnectStreamingRequest, StreamBody, Streaming,
-    };
+    pub use crate::message::{ConnectRequest, ConnectResponse, StreamBody, Streaming};
     pub use crate::service_builder::MakeServiceBuilder;
 }

connectrpc-axum/src/message.rs

@@ -3,5 +3,5 @@
 mod request;
 mod response;
 
-pub use request::{ConnectRequest, ConnectStreamingRequest, Streaming};
+pub use request::{ConnectRequest, Streaming};
 pub use response::{ConnectResponse, StreamBody};

connectrpc-axum/src/message/request.rs

@@ -256,48 +256,6 @@ where
     Ok(ConnectRequest(message))
 }
 
-/// Streaming request extractor for client streaming and bidi streaming RPCs.
-///
-/// Parses multiple frames from the request body into a `Stream`.
-/// Each frame follows the Connect protocol format: `[flags:1][length:4][payload]`
-///
-/// This extractor is designed for generated code and does not support
-/// additional Axum extractors like `ConnectRequest` does.
-pub struct ConnectStreamingRequest<T> {
-    /// Stream of decoded messages from the request body.
-    pub stream: Pin<Box<dyn Stream<Item = Result<T, ConnectError>> + Send>>,
-}
-
-impl<S, T> FromRequest<S> for ConnectStreamingRequest<T>
-where
-    S: Send + Sync,
-    T: Message + DeserializeOwned + Default + Send + 'static,
-{
-    type Rejection = ConnectError;
-
-    async fn from_request(req: Request, _state: &S) -> Result<Self, Self::Rejection> {
-        // Only POST is supported for streaming requests
-        if *req.method() != Method::POST {
-            return Err(ConnectError::new(
-                Code::Unimplemented,
-                "streaming requests only support POST method",
-            ));
-        }
-
-        // Get context (with fallback to default if layer is missing)
-        let ctx = get_context_or_default(&req);
-
-        let use_proto = ctx.protocol.is_proto();
-        let request_encoding = ctx.compression.request_encoding;
-        let body = req.into_body();
-
-        let stream = create_frame_stream::<T>(body, use_proto, ctx.limits, request_encoding);
-        Ok(ConnectStreamingRequest {
-            stream: Box::pin(stream),
-        })
-    }
-}
-
 /// `FromRequest` implementation for streaming requests using the unified `ConnectRequest<Streaming<T>>` pattern.
 ///
 /// This enables handlers to use the same `ConnectRequest` wrapper for both unary and streaming:

docs/guide/architecture.md

@@ -63,6 +63,11 @@ The `pipeline.rs` module provides the low-level functions used by extractors and
 - `decompress_bytes` - Decompress based on encoding
 - `decode_proto` / `decode_json` - Decode message from bytes
 - `unwrap_envelope` - Unwrap Connect streaming frame
+- `get_context_or_default` - Get `ConnectContext` from request extensions (with fallback)
+
+**`RequestPipeline` methods:**
+- `decode` - Decode from HTTP request (read body, decompress, decode)
+- `decode_bytes` - Decode from raw bytes (decompress, check size, decode)
 - `decode_enveloped_bytes` - Decode from enveloped bytes (for `application/connect+json` or `application/connect+proto`)
 
 **Response side:**
@@ -91,9 +96,8 @@ These are the types you interact with when building services:
 | Type | Purpose |
 |------|---------|
 | `ConnectRequest<T>` | Axum extractor - deserializes protobuf/JSON from request body |
-| `ConnectRequest<Streaming<T>>` | Extractor for client/bidi streaming requests (unified pattern) |
+| `ConnectRequest<Streaming<T>>` | Extractor for client/bidi streaming requests |
 | `Streaming<T>` | Stream of messages from client (similar to Tonic's `Streaming<T>`) |
-| `ConnectStreamingRequest<T>` | Extractor for client streaming requests (legacy) |
 | `ConnectResponse<T>` | Response wrapper - encodes per detected protocol |
 | `ConnectResponse<StreamBody<S>>` | Server streaming response wrapper |
 | `StreamBody<S>` | Marker for streaming response bodies |
@@ -272,12 +276,14 @@ Built-in implementations:
 
 The `default_codec()` function returns the appropriate codec for a `CompressionEncoding`. Custom codecs (zstd, brotli, etc.) can implement the `Codec` trait.
 
+Response compression negotiation follows RFC 7231: `negotiate_response_encoding()` parses `Accept-Encoding` headers respecting client preference order and `q=0` (not acceptable) values.
+
 ### message/ module
 
 | Module | Purpose |
 |--------|---------|
 | `request.rs` | `ConnectRequest<T>` and `Streaming<T>` extractors |
-| `response.rs` | `ConnectResponse<T>` and `ConnectResponse<StreamBody<S>>` encoding |
+| `response.rs` | `ConnectResponse<T>` and streaming response encoding |
 
 ## Code Generation