Is it possible to not decode the request body?

[kschat]

I have a service that listens to N Kafka topics. We'd like to put this service behind a gRPC "gateway", where callers send us the message to write to Kafka and a header for which topic to write to. This service handles a lot of load, so we need to make this gateway as efficient as possible. Given that, we don't want to waste resources decoding a message that is just going to be re-encoded right away. Is it possible to use codegen and receive a Request<Bytes> in the handler?

I tried writing a custom codec that effectively passes through the bytes. But it seems the input type generation is based on the gRPC file.

Codec:

#[derive(Debug)]
pub struct PassthroughEncoder<T>(PhantomData<T>);

impl<T: Message> Encoder for PassthroughEncoder<T> {
    type Item = T;
    type Error = Status;

    fn encode(&mut self, item: Self::Item, buf: &mut EncodeBuf<'_>) -> Result<(), Self::Error> {
        item.encode(buf)
            .map_err(|error| Status::internal(error.to_string()))
    }
}

#[derive(Debug)]
pub struct PassthroughDecoder;

impl Decoder for PassthroughDecoder {
    type Item = Bytes;
    type Error = Status;

    fn decode(&mut self, src: &mut DecodeBuf<'_>) -> Result<Option<Self::Item>, Self::Error> {
        if !src.has_remaining() {
            return Ok(None);
        }

        let mut buf = vec![];
        src.reader()
            .read_to_end(&mut buf)
            .map_err(|error| Status::internal(error.to_string()))?;

        Ok(Some(buf.into()))
    }
}

#[derive(Debug, Clone)]
pub struct PassthroughCodec<T: Message>(PhantomData<T>);

impl<T> Codec for PassthroughCodec<T>
where
    T: Message + 'static,
{
    type Encode = T;
    type Decode = Bytes;
    type Encoder = PassthroughEncoder<T>;
    type Decoder = PassthroughDecoder;

    fn encoder(&mut self) -> Self::Encoder {
        PassthroughEncoder(PhantomData)
    }

    fn decoder(&mut self) -> Self::Decoder {
        PassthroughDecoder
    }
}

proto file:

// Redacted since it's not important
message ImportNotificationEvent {}

message ImportNotificationResponse {
  required string id = 1;
}

service NotificationImporterService {
  rpc ImportNotification(ImportNotificationEvent) returns (ImportNotificationResponse)
}

build.rs:

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let importer_gateway = tonic_build::manual::Service::builder()
        .name("NotificationImporterService")
        .package("notification_center.events.v1")
        .method(
            tonic_build::manual::Method::builder()
                .name("import_notification")
                .route_name("ImportNotification")
                .input_type("bytes::Bytes")
                .output_type(
                    "notification_processor_schemas::provider::v1::ImportNotificationResponse",
                )
                .codec_path("notification_provider::grpc::PassthroughCodec")
                .build(),
        )
        .build();

    tonic_build::manual::Builder::new().compile(&[importer_gateway]);
}

And here is what I was hoping the trait would look like:

#[tonic::async_trait]
impl NotificationImporterService for NotificationProtoService {
    async fn import_notification(
        &self,
        request: Request<Bytes>,
    ) -> Result<tonic::Response<ImportNotificationResponse>, tonic::Status>;
}

but the request type is Request<ImportNotificationEvent>. Is there anyway to do this save writing a Service manually?