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

This change introduces data objects for the internal generic API.
We introduce separate objects for Unary and Streaming (e.g., `UnaryRequest` vs `StreamingRequest`, `UnaryResponseWriter` vs `StreamingResponseWriter`) as unifying reference and owned APIs proved difficult.

The input side (`UnaryRequest`, `StreamingRequest`) is straightforward, providing access to metadata and the message/stream.

The output side (`StreamingResponseWriter`) is slightly nuanced, offering a "staged builder" like API:
1.  `send_initial_metadata(metadata)` transitions to a `BodyWriter`.
2.  The `BodyWriter` allows writing the actual stream and sending trailers.

A unit test (`test_interceptor_composition`) is added to demonstrate how a writer can be intercepted, which is more nuanced than intercepting the read input stream.

UnaryMethod ByteStream Handler

Unary stuff


Refactor message traits

call changes

Change hander call options(might probably be deleted) since call options shouldn't contain metadata.

Author: sauravzg

grpc/Cargo.toml

@@ -17,6 +17,7 @@ allowed_external_types = [
 default = ["dns", "_runtime-tokio", "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.
@@ -60,6 +61,7 @@ tower = { version = "0.5.2", features = [
 tower-service = "0.3.3"
 url = "2.5.0"
 trait-variant = "0.1"
+send-future = "0.1"
 
 [dev-dependencies]
 async-stream = "0.3.6"

grpc/src/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::{StreamingResponseBodyWriter, StreamingResponseWriter};
+
+#[cfg(test)]
+pub(crate) mod test_util;

grpc/src/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/call/lazy.rs

@@ -0,0 +1,146 @@
+// ... (imports remain)
+use crate::message::AsMut;
+use crate::Status;
+use send_future::SendFuture;
+
+#[trait_variant::make(Send)]
+pub trait Lazy<Req>: Send
+where
+    Req: AsMut,
+{
+    async fn resolve(self, target: <Req as AsMut>::Mut<'_>) -> Result<(), Status>;
+}
+
+#[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 ---
+
+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).send().await?;
+        self.mapper.map(target).send().await
+    }
+}
+
+// --- Sync Map Combinator (Sugar) ---
+
+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).send().await?;
+        (self.f)(target)
+    }
+}
+
+pub trait LazyExt<Req>: Lazy<Req> + Sized
+where
+    Req: AsMut,
+{
+    fn then<M>(self, mapper: M) -> MapLazy<Self, M> {
+        MapLazy {
+            inner: self,
+            mapper,
+        }
+    }
+
+    // Sugar for synchronous closures
+    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/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/call/metadata.rs

@@ -0,0 +1,90 @@
+use http::HeaderMap;
+
+#[derive(Debug, Default, Clone, PartialEq)]
+pub struct Metadata {
+    pub(crate) inner: HeaderMap,
+}
+
+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/call/metadata_writer.rs

@@ -0,0 +1,18 @@
+use std::marker::Send;
+
+use crate::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>;
+}