Dyn ByteStreamHandler

Author: sauravzg

grpc/Cargo.toml

@@ -68,6 +68,7 @@ flate2 = { version = "1.0", optional = true }
 zstd = { version = "0.13", optional = true }
 trait-variant = "0.1"
 send-future = "0.1"
+dynosaur = "0.3.0"
 
 [dev-dependencies]
 async-stream = "0.3.6"

grpc/src/server/method_handler.rs

@@ -24,3 +24,11 @@ pub use codec_unary_method_handler::CodecUnaryMethodHandler;
 
 mod message_allocator;
 pub use message_allocator::{HeapMessageAllocator, MessageAllocator, MessageHolder};
+
+mod byte_stream_method_handler;
+pub use byte_stream_method_handler::{
+    ByteStreamMethodHandler, DynByteStreamMethodHandler, GenericByteStreamAdapter,
+};
+
+/// The default response body type produced by standard Codecs.
+pub type CodecRespB = bytes::Bytes;

grpc/src/server/method_handler/byte_stream_method_handler.rs

@@ -0,0 +1,48 @@
+use crate::call::{HandlerCallOptions, Incoming, StreamingRequest, StreamingResponseWriter};
+use crate::server::method_handler::GenericByteStreamMethodHandler;
+use crate::stream::PushStreamProducer;
+use crate::Status;
+use bytes::Buf;
+
+/// A method handler that processes raw bytes.
+#[trait_variant::make(Send)]
+#[dynosaur::dynosaur(pub DynByteStreamMethodHandler = dyn(box) ByteStreamMethodHandler)]
+pub trait ByteStreamMethodHandler<ReqB, W, P>: Send + Sync {
+    type RespB: Buf + Send;
+
+    async fn call(
+        &self,
+        options: HandlerCallOptions,
+        req: StreamingRequest<Incoming<ReqB>, P>,
+        resp: W,
+    ) -> Result<(), Status>
+    where
+        ReqB: Buf + Send,
+        P: PushStreamProducer<Item = Incoming<ReqB>> + Send,
+        W: StreamingResponseWriter<Self::RespB>;
+}
+
+/// Adapter to bridge `GenericByteStreamMethodHandler` to `ByteStreamMethodHandler`.
+pub struct GenericByteStreamAdapter<T>(pub T);
+
+impl<T, ReqB, W, P> ByteStreamMethodHandler<ReqB, W, P> for GenericByteStreamAdapter<T>
+where
+    T: GenericByteStreamMethodHandler,
+    W: StreamingResponseWriter<T::RespB> + Send,
+{
+    type RespB = T::RespB;
+
+    async fn call(
+        &self,
+        options: HandlerCallOptions,
+        req: StreamingRequest<Incoming<ReqB>, P>,
+        resp: W,
+    ) -> Result<(), Status>
+    where
+        ReqB: Buf + Send,
+        W: StreamingResponseWriter<Self::RespB>,
+        P: PushStreamProducer<Item = Incoming<ReqB>> + Send,
+    {
+        <T as GenericByteStreamMethodHandler>::call(&self.0, options, req, resp).await
+    }
+}