address encoding errors

Author: arjan-bal

grpc/src/client/transport/tonic/mod.rs

@@ -95,19 +95,13 @@ fn create_error_response(status: Status) -> GrpcResponse {
     TonicResponse::new(Box::pin(stream))
 }
 
-fn convert_request(req: GrpcRequest) -> TonicRequest<Pin<Box<dyn Stream<Item = Bytes> + Send>>> {
+fn convert_request(req: GrpcRequest) -> TonicRequest<BoxStream<Bytes>> {
     let (metadata, extensions, stream) = req.into_parts();
 
-    let bytes_stream = Box::pin(stream.filter_map(|msg| {
+    let bytes_stream = Box::pin(stream.map(|msg| {
         let mut buf = BytesMut::with_capacity(msg.encoded_message_size_hint().unwrap_or(0));
-        if let Ok(()) = msg.encode(&mut buf) {
-            Some(buf.freeze())
-        } else {
-            // TODO: Handle encoding failures.
-            // If it fails, log the error and return None to filter it out.
-            eprintln!("A message could not be downcast to Bytes and was skipped.");
-            None
-        }
+        msg.encode(&mut buf).map_err(Status::internal)?;
+        Ok(buf.freeze())
     }));
 
     TonicRequest::from_parts(metadata, extensions, bytes_stream as _)

grpc/src/codec.rs

@@ -11,7 +11,7 @@ use tonic::{
 pub(crate) struct BytesCodec {}
 
 impl Codec for BytesCodec {
-    type Encode = Bytes;
+    type Encode = Result<Bytes, Status>;
     type Decode = Bytes;
     type Encoder = BytesEncoder;
     type Decoder = BytesDecoder;
@@ -28,11 +28,11 @@ impl Codec for BytesCodec {
 pub struct BytesEncoder {}
 
 impl Encoder for BytesEncoder {
-    type Item = Bytes;
+    type Item = Result<Bytes, Status>;
     type Error = Status;
 
     fn encode(&mut self, item: Self::Item, dst: &mut EncodeBuf<'_>) -> Result<(), Self::Error> {
-        dst.put_slice(&item);
+        dst.put_slice(&item?);
         Ok(())
     }
 }