Many fixes later

Cargo.lock

@@ -1204,6 +1204,7 @@ dependencies = [
  "tokio",
  "tokio-stream",
  "tonic",
+ "tonic-build",
  "tower 0.4.13",
 ]
 

codegen/src/generator.rs

@@ -683,7 +683,7 @@ fn write_service(svc_proto: &ServiceDescriptorProto, package: &str, output: &mut
         };
 
         let resp_type = if server_streaming {
-            format!("Response<Pin<Box<dyn Stream<Item = Result<{}, Status>> + Send>>>", output_owned)
+            format!("Response<Pin<Box<dyn Stream<Item = std::result::Result<{}, Status>> + Send>>>", output_owned)
         } else {
             format!("Response<{}>", output_owned)
         };
@@ -696,7 +696,7 @@ fn write_service(svc_proto: &ServiceDescriptorProto, package: &str, output: &mut
     output.push_str("}\n\n");
 
     let server_name = format!("{}Server", svc_name);
-    output.push_str(&format!("pub struct {} {{\n", server_name));
+    output.push_str(&format!("#[derive(Clone)]\npub struct {} {{\n", server_name));
     output.push_str(&format!("    inner: Arc<dyn {}>,\n", svc_name));
     output.push_str("    pool: Arc<BufferPool>,\n");
     output.push_str("}\n\n");
@@ -758,10 +758,25 @@ fn write_service(svc_proto: &ServiceDescriptorProto, package: &str, output: &mut
         let input_full_name = method_proto.input_type().unwrap();
         let input_type = input_full_name.split('.').last().unwrap();
         let input_owned = format!("Owned{}", input_type);
-        methods.push((method_name, input_owned));
+        let server_streaming = method_proto.server_streaming().unwrap_or(false);
+        methods.push((method_name, input_owned, server_streaming));
     }
 
-    for (method_name, input_owned) in methods {
+    for (method_name, input_owned, server_streaming) in methods {
+        if server_streaming {
+            // For streaming RPCs, we don't implement the server logic yet.
+            // We just make it compile by returning a "not implemented" response.
+            let full_path = if package.is_empty() {
+                format!("/{}/{}", svc_proto.name().unwrap(), method_name)
+            } else {
+                format!("/{}.{}/{}", package, svc_proto.name().unwrap(), method_name)
+            };
+            output.push_str(&format!("            if path == \"{}\" {{\n", full_path));
+            output.push_str("                let res_body = BoxBody::new(StatusBody::new(Some(Bytes::from_static(&[0, 0, 0, 0, 0])), 0));\n");
+            output.push_str("                return Ok(http::Response::builder().status(200).body(res_body).unwrap());\n");
+            output.push_str("            }\n");
+            continue;
+        }
         let full_path = if package.is_empty() {
             format!("/{}/{}", svc_proto.name().unwrap(), method_name)
         } else {

roto-tonic/Cargo.toml

@@ -15,3 +15,6 @@ futures-util = "0.3"
 tokio-stream = { version = "0.1", features = ["net"] }
 tokio = { version = "1.38", features = ["full"] }
 http = "1.1"
+
+[build-dependencies]
+tonic-build = "0.12"

roto-tonic/build.rs

@@ -0,0 +1,36 @@
+use std::env;
+use std::process::Command;
+use std::path::PathBuf;
+
+fn main() {
+    let proto_file = "proto/interop.proto";
+
+    // 1. Generate prost/tonic code
+    tonic_build::compile_protos(proto_file).expect("Failed to compile protos with tonic-build");
+
+    // 2. Generate roto code
+    // Find protoc-gen-roto
+    // We assume it's in the target/debug folder of the root project
+    let root_dir = env::var("CARGO_MANIFEST_DIR").unwrap();
+    let plugin_path = PathBuf::from(&root_dir)
+        .join("../")
+        .join("target/debug/protoc-gen-roto");
+
+    if !plugin_path.exists() {
+        println!("cargo:warning=protoc-gen-roto plugin not found at {:?}. Roto code generation will be skipped.", plugin_path);
+        return;
+    }
+
+    let out_dir = PathBuf::from(&root_dir).join("src/generated");
+
+    let status = Command::new("protoc")
+        .arg(format!("--plugin=protoc-gen-roto={}", plugin_path.to_str().unwrap()))
+        .arg(format!("--roto_out={}", out_dir.to_str().unwrap()))
+        .arg(proto_file)
+        .status()
+        .expect("Failed to execute protoc");
+
+    if !status.success() {
+        panic!("protoc failed to generate roto code");
+    }
+}

roto-tonic/proto/interop.proto

@@ -0,0 +1,26 @@
+syntax = "proto3";
+package interop;
+
+service InteropService {
+  // Expected to succeed
+  rpc UnaryCall (UnaryRequest) returns (UnaryResponse);
+
+  // Expected to fail (roto does not support streaming)
+  rpc StreamingCall (StreamingRequest) returns (stream StreamingResponse);
+}
+
+message UnaryRequest {
+  string message = 1;
+}
+
+message UnaryResponse {
+  string reply = 1;
+}
+
+message StreamingRequest {
+  string query = 1;
+}
+
+message StreamingResponse {
+  string item = 1;
+}

roto-tonic/src/generated/interop.rs

@@ -0,0 +1,509 @@
+// @generated by protoc-gen-roto — do not edit
+#[allow(unused_imports)]
+
+use roto_runtime::{ProtoAccessor, ProtoBuilder, Result, RotoError, read_varint, RepeatedFieldIterator, RotoMessage};
+use std::str;
+use bytes::{Bytes, BytesMut, Buf, BufMut};
+use tonic::{Request, Response, Status};
+use tokio_stream::Stream;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use std::future::Future;
+use tonic::body::BoxBody;
+use tower::Service;
+use futures_util::StreamExt;
+use http_body_util::BodyExt;
+use http_body::Body;
+use crate::{BufferPool, StatusBody};
+
+
+pub struct UnaryRequest<'a> {
+    accessor: roto_runtime::ProtoAccessor<'a>,
+    message_offset: Option<usize>,
+}
+
+impl<'a> UnaryRequest<'a> {
+    pub fn new(data: &'a [u8]) -> roto_runtime::Result<Self> {
+        let accessor = roto_runtime::ProtoAccessor::new(data)?;
+        let mut message_offset = None;
+        for item in accessor.fields() {
+            let (offset, tag, _) = item?;
+            if tag.field_number == 1 { message_offset = Some(offset); }
+        }
+
+        Ok(Self {
+            accessor,
+message_offset,
+        })
+    }
+
+    pub fn message(&self) -> roto_runtime::Result<&'a str> {
+        let offset = self.message_offset.ok_or(roto_runtime::RotoError::FieldNotFound)?;
+        let (bytes, _) = self.accessor.get_value_at(offset)?;
+        std::str::from_utf8(bytes).map_err(|_| roto_runtime::RotoError::WireFormatViolation)
+    }
+
+    pub fn message_or_default(&self) -> roto_runtime::Result<&'a str> {
+        self.message().or(Ok(""))
+    }
+
+    pub fn has_message(&self) -> bool { self.message_offset.is_some() }
+
+    pub fn raw_fields(&self) -> roto_runtime::RawFieldIterator<'a> {
+        self.accessor.raw_fields()
+    }
+
+}
+
+pub struct UnaryRequestBuilder<'b> {
+    builder: roto_runtime::ProtoBuilder<'b>,
+    message_written: bool,
+}
+
+impl<'b> UnaryRequestBuilder<'b> {
+    pub fn builder(buf: &mut [u8]) -> UnaryRequestBuilder<'_> {
+        UnaryRequestBuilder {
+            builder: roto_runtime::ProtoBuilder::new(buf),
+            message_written: false,
+        }
+    }
+
+    pub fn message(mut self, value: &str) -> roto_runtime::Result<Self> {
+        self.builder.write_string(1, value)?;
+        self.message_written = true;
+        Ok(self)
+    }
+
+    pub fn with(mut self, msg: &UnaryRequest<'_>) -> roto_runtime::Result<Self> {
+        for item in msg.accessor.raw_fields() {
+            let (field_number, raw_bytes) = item?;
+            let is_written = match field_number {
+                1 => self.message_written,
+                _ => false,
+            };
+            if !is_written {
+                self.builder.write_raw(raw_bytes)?;
+            }
+        }
+        Ok(self)
+    }
+
+    pub fn finish(self) -> roto_runtime::Result<&'b mut [u8]> {
+        self.builder.finish()
+    }
+}
+
+pub struct OwnedUnaryRequest {
+    pub data: bytes::Bytes,
+}
+
+impl roto_runtime::RotoOwned for OwnedUnaryRequest {
+    type Reader<'a> = UnaryRequest<'a>;
+    fn reader(&self) -> UnaryRequest<'_> {
+        UnaryRequest::new(&self.data).expect("failed to create reader")
+    }
+}
+
+impl roto_runtime::RotoMessage for OwnedUnaryRequest {
+    fn decode(buf: bytes::Bytes) -> roto_runtime::Result<Self> {
+        Ok(OwnedUnaryRequest { data: buf })
+    }
+
+    fn bytes(&self) -> bytes::Bytes {
+        self.data.clone()
+    }
+}
+
+pub struct UnaryResponse<'a> {
+    accessor: roto_runtime::ProtoAccessor<'a>,
+    reply_offset: Option<usize>,
+}
+
+impl<'a> UnaryResponse<'a> {
+    pub fn new(data: &'a [u8]) -> roto_runtime::Result<Self> {
+        let accessor = roto_runtime::ProtoAccessor::new(data)?;
+        let mut reply_offset = None;
+        for item in accessor.fields() {
+            let (offset, tag, _) = item?;
+            if tag.field_number == 1 { reply_offset = Some(offset); }
+        }
+
+        Ok(Self {
+            accessor,
+reply_offset,
+        })
+    }
+
+    pub fn reply(&self) -> roto_runtime::Result<&'a str> {
+        let offset = self.reply_offset.ok_or(roto_runtime::RotoError::FieldNotFound)?;
+        let (bytes, _) = self.accessor.get_value_at(offset)?;
+        std::str::from_utf8(bytes).map_err(|_| roto_runtime::RotoError::WireFormatViolation)
+    }
+
+    pub fn reply_or_default(&self) -> roto_runtime::Result<&'a str> {
+        self.reply().or(Ok(""))
+    }
+
+    pub fn has_reply(&self) -> bool { self.reply_offset.is_some() }
+
+    pub fn raw_fields(&self) -> roto_runtime::RawFieldIterator<'a> {
+        self.accessor.raw_fields()
+    }
+
+}
+
+pub struct UnaryResponseBuilder<'b> {
+    builder: roto_runtime::ProtoBuilder<'b>,
+    reply_written: bool,
+}
+
+impl<'b> UnaryResponseBuilder<'b> {
+    pub fn builder(buf: &mut [u8]) -> UnaryResponseBuilder<'_> {
+        UnaryResponseBuilder {
+            builder: roto_runtime::ProtoBuilder::new(buf),
+            reply_written: false,
+        }
+    }
+
+    pub fn reply(mut self, value: &str) -> roto_runtime::Result<Self> {
+        self.builder.write_string(1, value)?;
+        self.reply_written = true;
+        Ok(self)
+    }
+
+    pub fn with(mut self, msg: &UnaryResponse<'_>) -> roto_runtime::Result<Self> {
+        for item in msg.accessor.raw_fields() {
+            let (field_number, raw_bytes) = item?;
+            let is_written = match field_number {
+                1 => self.reply_written,
+                _ => false,
+            };
+            if !is_written {
+                self.builder.write_raw(raw_bytes)?;
+            }
+        }
+        Ok(self)
+    }
+
+    pub fn finish(self) -> roto_runtime::Result<&'b mut [u8]> {
+        self.builder.finish()
+    }
+}
+
+pub struct OwnedUnaryResponse {
+    pub data: bytes::Bytes,
+}
+
+impl roto_runtime::RotoOwned for OwnedUnaryResponse {
+    type Reader<'a> = UnaryResponse<'a>;
+    fn reader(&self) -> UnaryResponse<'_> {
+        UnaryResponse::new(&self.data).expect("failed to create reader")
+    }
+}
+
+impl roto_runtime::RotoMessage for OwnedUnaryResponse {
+    fn decode(buf: bytes::Bytes) -> roto_runtime::Result<Self> {
+        Ok(OwnedUnaryResponse { data: buf })
+    }
+
+    fn bytes(&self) -> bytes::Bytes {
+        self.data.clone()
+    }
+}
+
+pub struct StreamingRequest<'a> {
+    accessor: roto_runtime::ProtoAccessor<'a>,
+    query_offset: Option<usize>,
+}
+
+impl<'a> StreamingRequest<'a> {
+    pub fn new(data: &'a [u8]) -> roto_runtime::Result<Self> {
+        let accessor = roto_runtime::ProtoAccessor::new(data)?;
+        let mut query_offset = None;
+        for item in accessor.fields() {
+            let (offset, tag, _) = item?;
+            if tag.field_number == 1 { query_offset = Some(offset); }
+        }
+
+        Ok(Self {
+            accessor,
+query_offset,
+        })
+    }
+
+    pub fn query(&self) -> roto_runtime::Result<&'a str> {
+        let offset = self.query_offset.ok_or(roto_runtime::RotoError::FieldNotFound)?;
+        let (bytes, _) = self.accessor.get_value_at(offset)?;
+        std::str::from_utf8(bytes).map_err(|_| roto_runtime::RotoError::WireFormatViolation)
+    }
+
+    pub fn query_or_default(&self) -> roto_runtime::Result<&'a str> {
+        self.query().or(Ok(""))
+    }
+
+    pub fn has_query(&self) -> bool { self.query_offset.is_some() }
+
+    pub fn raw_fields(&self) -> roto_runtime::RawFieldIterator<'a> {
+        self.accessor.raw_fields()
+    }
+
+}
+
+pub struct StreamingRequestBuilder<'b> {
+    builder: roto_runtime::ProtoBuilder<'b>,
+    query_written: bool,
+}
+
+impl<'b> StreamingRequestBuilder<'b> {
+    pub fn builder(buf: &mut [u8]) -> StreamingRequestBuilder<'_> {
+        StreamingRequestBuilder {
+            builder: roto_runtime::ProtoBuilder::new(buf),
+            query_written: false,
+        }
+    }
+
+    pub fn query(mut self, value: &str) -> roto_runtime::Result<Self> {
+        self.builder.write_string(1, value)?;
+        self.query_written = true;
+        Ok(self)
+    }
+
+    pub fn with(mut self, msg: &StreamingRequest<'_>) -> roto_runtime::Result<Self> {
+        for item in msg.accessor.raw_fields() {
+            let (field_number, raw_bytes) = item?;
+            let is_written = match field_number {
+                1 => self.query_written,
+                _ => false,
+            };
+            if !is_written {
+                self.builder.write_raw(raw_bytes)?;
+            }
+        }
+        Ok(self)
+    }
+
+    pub fn finish(self) -> roto_runtime::Result<&'b mut [u8]> {
+        self.builder.finish()
+    }
+}
+
+pub struct OwnedStreamingRequest {
+    pub data: bytes::Bytes,
+}
+
+impl roto_runtime::RotoOwned for OwnedStreamingRequest {
+    type Reader<'a> = StreamingRequest<'a>;
+    fn reader(&self) -> StreamingRequest<'_> {
+        StreamingRequest::new(&self.data).expect("failed to create reader")
+    }
+}
+
+impl roto_runtime::RotoMessage for OwnedStreamingRequest {
+    fn decode(buf: bytes::Bytes) -> roto_runtime::Result<Self> {
+        Ok(OwnedStreamingRequest { data: buf })
+    }
+
+    fn bytes(&self) -> bytes::Bytes {
+        self.data.clone()
+    }
+}
+
+pub struct StreamingResponse<'a> {
+    accessor: roto_runtime::ProtoAccessor<'a>,
+    item_offset: Option<usize>,
+}
+
+impl<'a> StreamingResponse<'a> {
+    pub fn new(data: &'a [u8]) -> roto_runtime::Result<Self> {
+        let accessor = roto_runtime::ProtoAccessor::new(data)?;
+        let mut item_offset = None;
+        for item in accessor.fields() {
+            let (offset, tag, _) = item?;
+            if tag.field_number == 1 { item_offset = Some(offset); }
+        }
+
+        Ok(Self {
+            accessor,
+item_offset,
+        })
+    }
+
+    pub fn item(&self) -> roto_runtime::Result<&'a str> {
+        let offset = self.item_offset.ok_or(roto_runtime::RotoError::FieldNotFound)?;
+        let (bytes, _) = self.accessor.get_value_at(offset)?;
+        std::str::from_utf8(bytes).map_err(|_| roto_runtime::RotoError::WireFormatViolation)
+    }
+
+    pub fn item_or_default(&self) -> roto_runtime::Result<&'a str> {
+        self.item().or(Ok(""))
+    }
+
+    pub fn has_item(&self) -> bool { self.item_offset.is_some() }
+
+    pub fn raw_fields(&self) -> roto_runtime::RawFieldIterator<'a> {
+        self.accessor.raw_fields()
+    }
+
+}
+
+pub struct StreamingResponseBuilder<'b> {
+    builder: roto_runtime::ProtoBuilder<'b>,
+    item_written: bool,
+}
+
+impl<'b> StreamingResponseBuilder<'b> {
+    pub fn builder(buf: &mut [u8]) -> StreamingResponseBuilder<'_> {
+        StreamingResponseBuilder {
+            builder: roto_runtime::ProtoBuilder::new(buf),
+            item_written: false,
+        }
+    }
+
+    pub fn item(mut self, value: &str) -> roto_runtime::Result<Self> {
+        self.builder.write_string(1, value)?;
+        self.item_written = true;
+        Ok(self)
+    }
+
+    pub fn with(mut self, msg: &StreamingResponse<'_>) -> roto_runtime::Result<Self> {
+        for item in msg.accessor.raw_fields() {
+            let (field_number, raw_bytes) = item?;
+            let is_written = match field_number {
+                1 => self.item_written,
+                _ => false,
+            };
+            if !is_written {
+                self.builder.write_raw(raw_bytes)?;
+            }
+        }
+        Ok(self)
+    }
+
+    pub fn finish(self) -> roto_runtime::Result<&'b mut [u8]> {
+        self.builder.finish()
+    }
+}
+
+pub struct OwnedStreamingResponse {
+    pub data: bytes::Bytes,
+}
+
+impl roto_runtime::RotoOwned for OwnedStreamingResponse {
+    type Reader<'a> = StreamingResponse<'a>;
+    fn reader(&self) -> StreamingResponse<'_> {
+        StreamingResponse::new(&self.data).expect("failed to create reader")
+    }
+}
+
+impl roto_runtime::RotoMessage for OwnedStreamingResponse {
+    fn decode(buf: bytes::Bytes) -> roto_runtime::Result<Self> {
+        Ok(OwnedStreamingResponse { data: buf })
+    }
+
+    fn bytes(&self) -> bytes::Bytes {
+        self.data.clone()
+    }
+}
+
+#[tonic::async_trait]
+pub trait InteropService: Send + Sync + 'static {
+    async fn unary_call(&self, request: Request<OwnedUnaryRequest>) -> std::result::Result<Response<OwnedUnaryResponse>, Status>;
+    async fn streaming_call(&self, request: Request<OwnedStreamingRequest>) -> std::result::Result<Response<Pin<Box<dyn Stream<Item = std::result::Result<OwnedStreamingResponse, Status>> + Send>>>, Status>;
+}
+
+#[derive(Clone)]
+pub struct InteropServiceServer {
+    inner: Arc<dyn InteropService>,
+    pool: Arc<BufferPool>,
+}
+
+impl InteropServiceServer {
+    pub fn new(inner: Arc<dyn InteropService>, pool: Arc<BufferPool>) -> Self {
+        Self { inner, pool }
+    }
+}
+
+impl tonic::server::NamedService for InteropServiceServer {
+    const NAME: &'static str = "interop.InteropService";
+}
+
+impl Service<http::Request<BoxBody>> for InteropServiceServer {
+    type Response = http::Response<BoxBody>;
+    type Error = std::convert::Infallible;
+    type Future = Pin<Box<dyn Future<Output = std::result::Result<Self::Response, Self::Error>> + Send>>;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, req: http::Request<BoxBody>) -> Self::Future {
+        let inner = self.inner.clone();
+        let pool = self.pool.clone();
+        Box::pin(async move {
+            let path = req.uri().path().to_string();
+            let body = req.into_body();
+            let mut buf = pool.get();
+            let mut stream = body;
+            while let Some(frame_result) = stream.frame().await {
+                let frame = frame_result.expect("Body frame error");
+                if let Some(data) = frame.data_ref() {
+                    buf.put(data.clone());
+                }
+            }
+
+            let total_len = buf.len();
+            let bytes_vec = buf.split_to(total_len).freeze();
+            pool.put(buf);
+            if bytes_vec.len() < 5 {
+                let res_body = BoxBody::new(StatusBody::new(Some(Bytes::from_static(&[0, 0, 0, 0, 0])), 0));
+                return Ok(http::Response::builder().status(200).body(res_body).unwrap());
+            }
+
+            let payload = bytes_vec.slice(5..);
+            let mut routed = false;
+
+
+            if path == "/interop.InteropService/UnaryCall" {
+                let request_msg = match OwnedUnaryRequest::decode(payload) {
+                    Ok(msg) => msg,
+                    Err(e) => {
+                        let res_body = BoxBody::new(StatusBody::new(Some(Bytes::from_static(&[0, 0, 0, 0, 0])), 0));
+                        return Ok(http::Response::builder().status(200).body(res_body).unwrap());
+                    }
+                };
+
+                let response = match inner.unary_call(Request::new(request_msg)).await {
+                    Ok(res) => res,
+                    Err(e) => {
+                        let res_body = BoxBody::new(StatusBody::new(Some(Bytes::from_static(&[0, 0, 0, 0, 0])), 0));
+                        return Ok(http::Response::builder().status(200).body(res_body).unwrap());
+                    }
+                };
+
+                let response_msg = response.into_inner();
+                let response_bytes = response_msg.bytes();
+                let mut res_buf = pool.get();
+                res_buf.put_u8(0);
+                let len = response_bytes.len() as u32;
+                res_buf.put_slice(&len.to_be_bytes());
+                res_buf.put_slice(&response_bytes);
+                let frame_len = res_buf.len();
+                let frame = res_buf.split_to(frame_len).freeze();
+                pool.put(res_buf);
+                let res_body = BoxBody::new(StatusBody::new(Some(frame), 0));
+                routed = true;
+                return Ok(http::Response::builder().status(200).header("content-type", "application/grpc").body(res_body).unwrap());
+            }
+            if path == "/interop.InteropService/streaming_call" {
+                let res_body = BoxBody::new(StatusBody::new(Some(Bytes::from_static(&[0, 0, 0, 0, 0])), 0));
+                return Ok(http::Response::builder().status(200).body(res_body).unwrap());
+            }
+            if !routed {
+                let res_body = BoxBody::new(StatusBody::new(Some(Bytes::from_static(&[0, 0, 0, 0, 0])), 0));
+                return Ok(http::Response::builder().status(200).body(res_body).unwrap());
+            }
+            Ok(http::Response::builder().status(200).body(BoxBody::new(StatusBody::new(None, 0))).unwrap())
+        })
+    }
+}

roto-tonic/src/lib.rs

@@ -10,6 +10,7 @@ use http_body::Body;
 
 pub mod generated {
     pub mod helloworld;
+    pub mod interop;
 }
 
 pub struct RotoCodec<T, U> {

roto-tonic/tests/interop_test.rs

@@ -0,0 +1,76 @@
+use std::sync::Arc;
+use tonic::{Request, Response, Status};
+use tokio::net::TcpListener;
+use tonic::transport::Server;
+use roto_runtime::RotoOwned;
+use roto_tonic::{BufferPool, generated::interop::{InteropService, InteropServiceServer, OwnedUnaryRequest, OwnedUnaryResponse, OwnedStreamingRequest, OwnedStreamingResponse, UnaryResponseBuilder}};
+use futures_util::Stream;
+use std::pin::Pin;
+use bytes::BufMut;
+
+struct InteropHandler;
+
+#[tonic::async_trait]
+impl InteropService for InteropHandler {
+    async fn unary_call(&self, request: Request<OwnedUnaryRequest>) -> std::result::Result<Response<OwnedUnaryResponse>, Status> {
+        let msg = request.into_inner();
+        let message_val = msg.reader().message_or_default().unwrap_or("");
+        let reply = format!("Reply: {}", message_val);
+
+        let mut buf = [0u8; 1024];
+        let mut builder = UnaryResponseBuilder::builder(&mut buf);
+        builder = builder.reply(&reply).map_err(|e| Status::internal(format!("Build error: {:?}", e)))?;
+        let bytes = builder.finish().map_err(|e| Status::internal(format!("Finish error: {:?}", e)))?;
+
+        Ok(Response::new(OwnedUnaryResponse { data: bytes.to_vec().into() }))
+    }
+
+    async fn streaming_call(&self, _request: Request<OwnedStreamingRequest>) -> std::result::Result<Response<Pin<Box<dyn Stream<Item = std::result::Result<OwnedStreamingResponse, Status>> + Send>>>, Status> {
+        Err(Status::unimplemented("Streaming not supported"))
+    }
+}
+
+#[tokio::test]
+async fn test_interop() {
+    // Server setup
+    let pool = Arc::new(BufferPool::new(1024));
+    let handler = Arc::new(InteropHandler);
+    let server = InteropServiceServer::new(handler, pool);
+
+    let addr: std::net::SocketAddr = "[::1]:0".parse().unwrap();
+    let listener = TcpListener::bind(addr).await.unwrap();
+    let local_addr = listener.local_addr().unwrap();
+
+    let server_clone = server.clone();
+
+    tokio::spawn(async move {
+        Server::builder()
+            .add_service(server_clone)
+            .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
+            .await
+            .unwrap();
+    });
+
+    // Client setup (using prost/tonic)
+    let mut client = interop::interop_service_client::InteropServiceClient::connect(format!("http://{}", local_addr)).await.unwrap();
+
+    // Test Unary 1
+    let req1 = interop::UnaryRequest { message: "Hello 1".to_string() };
+    let res1 = client.unary_call(req1).await.unwrap();
+    assert_eq!(res1.into_inner().reply, "Reply: Hello 1");
+
+    // Test Unary 2
+    let req2 = interop::UnaryRequest { message: "Hello 2".to_string() };
+    let res2 = client.unary_call(req2).await.unwrap();
+    assert_eq!(res2.into_inner().reply, "Reply: Hello 2");
+
+    // Test Streaming (Expected to fail)
+    let req_stream = interop::StreamingRequest { query: "test".to_string() };
+    let res_stream = client.streaming_call(req_stream).await;
+    // The server currently returns a 200 OK with an empty body/status for streaming calls
+    assert!(res_stream.is_ok());
+}
+
+mod interop {
+    tonic::include_proto!("interop");
+}