feat(call): Add call request and response objects for internal generic API

grpc/Cargo.toml

@@ -18,6 +18,7 @@ allowed_external_types = [
 default = ["dns", "_runtime-tokio","tls-rustls", "protobuf"]
 protobuf = ["dep:protobuf"]
 prost = ["dep:prost"]
+unsafe-downcast = []
 dns = ["dep:hickory-resolver", "_runtime-tokio"]
 # The following feature is used to ensure all modules use the runtime
 # abstraction instead of using tokio directly.

grpc/src/server/call.rs

@@ -0,0 +1,17 @@
+pub mod handler_call_options;
+pub mod lazy;
+pub mod message_wrapper;
+pub mod metadata;
+pub mod metadata_writer;
+pub mod streaming_request;
+pub mod streaming_response_writer;
+pub mod streaming_response_writer_ext;
+pub use handler_call_options::HandlerCallOptions;
+pub use lazy::Lazy;
+pub use message_wrapper::{Incoming, Outgoing};
+pub use metadata::Metadata;
+pub use streaming_request::StreamingRequest;
+pub use streaming_response_writer::StreamingResponseWriter;
+
+#[cfg(test)]
+pub(crate) mod test_util;

grpc/src/server/call/handler_call_options.rs

@@ -0,0 +1,5 @@
+/// Options for method handlers.
+#[derive(Debug, Default, Clone)]
+pub struct HandlerCallOptions {
+    pub compression_encoding: Option<String>,
+}

grpc/src/server/call/lazy.rs

@@ -0,0 +1,168 @@
+// ... (imports remain)
+use crate::send_future::SendFuture;
+use crate::server::message::AsMut;
+use crate::Status;
+
+/// A trait for lazily resolving and mutating a request message.
+///
+/// This is a core component of Tonic's unified request architecture,
+/// allowing mutations to be deferred until the message is actually needed.
+/// This enables zero-copy patterns and flexible handling of request bodies.
+#[trait_variant::make(Send)]
+pub trait Lazy<Req>: Send
+where
+    Req: AsMut,
+{
+    async fn resolve(self, target: <Req as AsMut>::Mut<'_>) -> Result<(), Status>;
+}
+
+/// A trait for mapping or applying further mutations to a request message asynchronously.
+///
+/// This is typically used to chain operations without boxing or allocating closures.
+#[trait_variant::make(Send)]
+pub trait Mapper<Req>: Send
+where
+    Req: AsMut,
+{
+    async fn map(self, target: <Req as AsMut>::Mut<'_>) -> Result<(), Status>;
+}
+
+// --- Zero-Cost Map Combinator ---
+
+/// A zero-cost combinator that chains a `Lazy` implementation with an asynchronous `Mapper`.
+///
+/// This struct is created by the [`LazyExt::then`] method.
+pub struct MapLazy<L, M> {
+    pub inner: L,
+    pub mapper: M,
+}
+
+impl<Req, L, M> Lazy<Req> for MapLazy<L, M>
+where
+    Req: AsMut,
+    L: Lazy<Req>,
+    M: Mapper<Req>,
+{
+    async fn resolve(self, mut target: <Req as AsMut>::Mut<'_>) -> Result<(), Status> {
+        let inner_target = <Req as AsMut>::reborrow_view(&mut target);
+        self.inner.resolve(inner_target).make_send().await?;
+        self.mapper.map(target).make_send().await
+    }
+}
+
+// --- Sync Map Combinator (Sugar) ---
+
+/// A combinator that chains a `Lazy` implementation with a synchronous closure.
+///
+/// This struct is created by the [`LazyExt::map`] method.
+pub struct SyncMapLazy<L, F> {
+    pub inner: L,
+    pub f: F,
+}
+
+impl<Req, L, F> Lazy<Req> for SyncMapLazy<L, F>
+where
+    Req: AsMut,
+    L: Lazy<Req>,
+    F: FnOnce(<Req as AsMut>::Mut<'_>) -> Result<(), Status> + Send,
+{
+    async fn resolve(self, mut target: <Req as AsMut>::Mut<'_>) -> Result<(), Status> {
+        let inner_target = <Req as AsMut>::reborrow_view(&mut target);
+        self.inner.resolve(inner_target).make_send().await?;
+        (self.f)(target)
+    }
+}
+
+/// Extension trait providing combinators for `Lazy` implementations.
+pub trait LazyExt<Req>: Lazy<Req> + Sized
+where
+    Req: AsMut,
+{
+    /// Chains this lazy operation with an asynchronous `Mapper`.
+    ///
+    /// This produces a new `Lazy` that runs the initial resolution and then applies
+    /// the asynchronous mapper, avoiding allocations for the combined operation.
+    fn then<M>(self, mapper: M) -> MapLazy<Self, M> {
+        MapLazy {
+            inner: self,
+            mapper,
+        }
+    }
+
+    /// Chains this lazy operation with a synchronous closure.
+    ///
+    /// This is syntactic sugar to apply synchronous mutations after the initial
+    /// resolution completes.
+    fn map<F>(self, f: F) -> SyncMapLazy<Self, F>
+    where
+        F: FnOnce(<Req as AsMut>::Mut<'_>) -> Result<(), Status> + Send,
+    {
+        SyncMapLazy { inner: self, f }
+    }
+}
+
+impl<Req, L: Lazy<Req>> LazyExt<Req> for L where Req: AsMut {}
+
+// --- Test Usage ---
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use protobuf_well_known_types::Timestamp;
+
+    // A struct for our lazy logic (Zero Cost)
+    struct SetSeconds {
+        value: i64,
+    }
+
+    impl Lazy<Timestamp> for SetSeconds {
+        async fn resolve(self, mut target: <Timestamp as AsMut>::Mut<'_>) -> Result<(), Status> {
+            target.set_seconds(self.value);
+            Ok(())
+        }
+    }
+
+    // A struct for our mapping logic (Zero Cost)
+    struct AddFive;
+    impl Mapper<Timestamp> for AddFive {
+        async fn map(self, mut target: <Timestamp as AsMut>::Mut<'_>) -> Result<(), Status> {
+            let current = target.seconds();
+            target.set_seconds(current + 5);
+            Ok(())
+        }
+    }
+
+    #[tokio::test]
+    async fn test_zero_allocation_chain() {
+        // 1. Create the base lazy (Struct, not closure)
+        let lazy = SetSeconds { value: 10 };
+
+        // 2. Map it with a struct mapper
+        let chained = lazy.then(AddFive);
+
+        let mut msg = Timestamp::new();
+
+        chained.resolve(msg.as_mut()).await.unwrap();
+
+        assert_eq!(msg.seconds(), 15);
+    }
+
+    #[tokio::test]
+    async fn test_sync_closure_sugar() {
+        // 1. Create the base lazy
+        let lazy = SetSeconds { value: 10 };
+
+        // 2. Map it with a sync closure (Sugar)
+        let chained = lazy.map(|mut target| {
+            let current = target.seconds();
+            target.set_seconds(current + 20);
+            Ok(())
+        });
+
+        let mut msg = Timestamp::new();
+        // ZeroBox allocation! Sync closures are easy.
+        chained.resolve(msg.as_mut()).await.unwrap();
+
+        assert_eq!(msg.seconds(), 30);
+    }
+}

grpc/src/server/call/message_wrapper.rs

@@ -0,0 +1,51 @@
+/// A raw message containing bytes and optional read options.
+pub struct Incoming<B> {
+    pub message_bytes: B,
+    pub options: Option<MessageReadOptions>,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct MessageReadOptions {
+    pub compressed: bool,
+}
+
+/// A wrapped message with optional write options.
+pub struct Outgoing<T> {
+    pub message: T,
+    pub options: Option<MessageWriteOptions>,
+}
+
+impl<T> Outgoing<T> {
+    pub fn new(message: T) -> Self {
+        Self {
+            message,
+            options: None,
+        }
+    }
+
+    pub fn with_options(message: T, options: MessageWriteOptions) -> Self {
+        Self {
+            message,
+            options: Some(options),
+        }
+    }
+}
+
+impl<T> From<T> for Outgoing<T> {
+    fn from(message: T) -> Self {
+        Self::new(message)
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum CompressionEncoding {
+    #[default]
+    Inherit,
+    Enabled,
+    Disabled,
+}
+
+#[derive(Debug, Default, Clone, Copy)]
+pub struct MessageWriteOptions {
+    pub compression: CompressionEncoding,
+}

grpc/src/server/call/metadata.rs

@@ -0,0 +1,94 @@
+use http::HeaderMap;
+
+#[derive(Debug, Default, Clone, PartialEq)]
+pub struct Metadata {
+    pub(crate) inner: HeaderMap,
+}
+
+const GRPC_ENCODING: &str = "grpc-encoding";
+const GRPC_ACCEPT_ENCODING: &str = "grpc-accept-encoding";
+const PATH: &str = "path";
+
+impl Metadata {
+    pub fn new(inner: HeaderMap) -> Self {
+        Self { inner }
+    }
+
+    /// Returns the method name.
+    pub fn method_name(&self) -> Option<&str> {
+        self.inner.get(PATH).and_then(|v| v.to_str().ok())
+    }
+
+    /// Returns the `grpc-encoding` header value.
+    pub fn encoding(&self) -> Option<&str> {
+        self.inner
+            .get(GRPC_ENCODING)
+            .and_then(|v| v.to_str().ok())
+    }
+
+    /// Returns an iterator over the `grpc-accept-encoding` values.
+    /// Handles multiple headers and comma-separated values.
+    pub fn accept_encodings(&self) -> impl Iterator<Item = &str> + '_ {
+        self.inner
+            .get_all(GRPC_ACCEPT_ENCODING)
+            .iter()
+            .filter_map(|v| v.to_str().ok())
+            .flat_map(|s| s.split(','))
+            .map(|s| s.trim())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use http::HeaderValue;
+
+    #[test]
+    fn test_method_name() {
+        let mut map = HeaderMap::new();
+        map.insert("path", HeaderValue::from_static("/Service/Method"));
+        let metadata = Metadata::new(map);
+
+        assert_eq!(metadata.method_name(), Some("/Service/Method"));
+
+        let empty = Metadata::default();
+        assert_eq!(empty.method_name(), None);
+    }
+
+    #[test]
+    fn test_encoding() {
+        let mut map = HeaderMap::new();
+        map.insert("grpc-encoding", HeaderValue::from_static("gzip"));
+        let metadata = Metadata::new(map);
+
+        assert_eq!(metadata.encoding(), Some("gzip"));
+
+        let empty = Metadata::default();
+        assert_eq!(empty.encoding(), None);
+    }
+
+    #[test]
+    fn test_accept_encodings() {
+        let mut map = HeaderMap::new();
+        map.insert(
+            "grpc-accept-encoding",
+            HeaderValue::from_static("gzip,identity"),
+        );
+        let metadata = Metadata::new(map);
+
+        let encodings: Vec<_> = metadata.accept_encodings().collect();
+        assert_eq!(encodings, vec!["gzip", "identity"]);
+
+        // Test multiple headers
+        let mut map = HeaderMap::new();
+        map.append("grpc-accept-encoding", HeaderValue::from_static("gzip"));
+        map.append(
+            "grpc-accept-encoding",
+            HeaderValue::from_static("deflate, br"),
+        );
+        let metadata = Metadata::new(map);
+
+        let encodings: Vec<_> = metadata.accept_encodings().collect();
+        assert_eq!(encodings, vec!["gzip", "deflate", "br"]);
+    }
+}

grpc/src/server/call/metadata_writer.rs

@@ -0,0 +1,18 @@
+use std::marker::Send;
+
+use crate::server::call::Metadata;
+use crate::Status;
+
+/// A trait for writing initial metadata.
+#[trait_variant::make(Send)]
+pub trait InitialMetadataWriter: Send {
+    /// Sends initial metadata.
+    async fn send_initial_metadata(self, metadata: Metadata) -> Result<(), Status>;
+}
+
+/// A trait for writing trailing metadata.
+#[trait_variant::make(Send)]
+pub trait TrailingMetadataWriter: Send {
+    /// Sends trailing metadata.
+    async fn send_trailing_metadata(self, metadata: Metadata) -> Result<(), Status>;
+}