feat: add grpc support

2026-03-25 21:23:57 +03:00
parent ada2436e54
commit 8005510ad2
33 changed files with 1725 additions and 32 deletions
@@ -5,10 +5,25 @@ license.workspace = true
 rust-version.workspace = true
 version.workspace = true

+[features]
+test-support = []
+
 [dependencies]
+base64.workspace = true
 mcpaas-core = { path = "../mcpaas-core" }
 mcpaas-proto = { path = "../mcpaas-proto" }
+prost.workspace = true
+prost-reflect.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true
+tonic.workspace = true
+tonic-prost.workspace = true
+tokio.workspace = true

+[dev-dependencies]
+prost-types.workspace = true
+
+[build-dependencies]
+protoc-bin-vendored.workspace = true
+tonic-prost-build.workspace = true
@@ -0,0 +1,17 @@
+use std::{env, path::PathBuf};
+
+fn main() {
+    let protoc = protoc_bin_vendored::protoc_bin_path().expect("vendored protoc must exist");
+    unsafe {
+        env::set_var("PROTOC", protoc);
+    }
+    let out_dir = PathBuf::from(env::var("OUT_DIR").expect("OUT_DIR must exist"));
+    let descriptor_path = out_dir.join("echo_descriptor.bin");
+
+    tonic_prost_build::configure()
+        .build_server(true)
+        .build_client(true)
+        .file_descriptor_set_path(descriptor_path)
+        .compile_protos(&["proto/echo.proto"], &["proto"])
+        .expect("echo proto must compile");
+}
@@ -0,0 +1,15 @@
+syntax = "proto3";
+
+package echo;
+
+service EchoService {
+  rpc UnaryEcho(EchoRequest) returns (EchoResponse);
+}
+
+message EchoRequest {
+  string message = 1;
+}
+
+message EchoResponse {
+  string message = 1;
+}
@@ -0,0 +1,168 @@
+use std::{collections::BTreeMap, str::FromStr, time::Duration};
+
+use base64::{Engine as _, engine::general_purpose::STANDARD};
+use mcpaas_core::GrpcTarget;
+use prost::Message;
+use prost_reflect::{DescriptorPool, MethodDescriptor, prost_types::FileDescriptorSet};
+use tonic::{
+    Request,
+    client::Grpc,
+    metadata::{KeyAndValueRef, MetadataKey, MetadataValue},
+    transport::Endpoint,
+};
+
+use crate::{GrpcAdapterError, GrpcRequest, GrpcResponse, codec::JsonCodec};
+
+#[derive(Clone, Debug, Default)]
+pub struct GrpcAdapter;
+
+impl GrpcAdapter {
+    pub fn new() -> Self {
+        Self
+    }
+
+    pub async fn execute(
+        &self,
+        target: &GrpcTarget,
+        request: &GrpcRequest,
+    ) -> Result<GrpcResponse, GrpcAdapterError> {
+        let method = resolve_method(target)?;
+        if method.is_client_streaming() || method.is_server_streaming() {
+            return Err(GrpcAdapterError::UnsupportedMethodKind {
+                service: format!("{}.{}", target.package, target.service),
+                method: target.method.clone(),
+            });
+        }
+
+        let endpoint = Endpoint::from_shared(target.server_addr.clone())?
+            .timeout(Duration::from_millis(request.timeout_ms));
+        let channel = endpoint.connect().await?;
+        let mut grpc = Grpc::new(channel);
+        grpc.ready()
+            .await
+            .map_err(|error| GrpcAdapterError::Status {
+                code: tonic::Code::Unavailable,
+                message: error.to_string(),
+            })?;
+
+        let service_name = method.parent_service().full_name().to_owned();
+        let method_name = method.name().to_owned();
+        let path = tonic::codegen::http::uri::PathAndQuery::from_str(&format!(
+            "/{service_name}/{method_name}"
+        ))
+        .map_err(|_| GrpcAdapterError::InvalidMethodPath {
+            service: service_name,
+            method: method_name,
+        })?;
+        let codec = JsonCodec::new(method.input(), method.output());
+        let mut tonic_request = Request::new(request.body.clone());
+
+        apply_headers(&mut tonic_request, &request.headers)?;
+
+        let response = grpc.unary(tonic_request, path, codec).await?;
+        let headers = normalize_headers(response.metadata());
+        let body = response.into_inner();
+
+        Ok(GrpcResponse {
+            status_code: 200,
+            headers,
+            body,
+        })
+    }
+}
+
+fn resolve_method(target: &GrpcTarget) -> Result<MethodDescriptor, GrpcAdapterError> {
+    let bytes = STANDARD
+        .decode(&target.descriptor_set_b64)
+        .map_err(|_| GrpcAdapterError::InvalidDescriptorEncoding)?;
+    let descriptor_set = FileDescriptorSet::decode(bytes.as_slice())
+        .map_err(|_| GrpcAdapterError::InvalidDescriptorSet)?;
+    let pool = DescriptorPool::from_file_descriptor_set(descriptor_set)
+        .map_err(|_| GrpcAdapterError::InvalidDescriptorPool)?;
+    let full_service_name = if target.package.is_empty() {
+        target.service.clone()
+    } else {
+        format!("{}.{}", target.package, target.service)
+    };
+    let service = pool
+        .get_service_by_name(&full_service_name)
+        .ok_or_else(|| GrpcAdapterError::ServiceNotFound {
+            service: full_service_name.clone(),
+        })?;
+
+    service
+        .methods()
+        .find(|method| method.name() == target.method)
+        .ok_or_else(|| GrpcAdapterError::MethodNotFound {
+            service: full_service_name,
+            method: target.method.clone(),
+        })
+}
+
+fn apply_headers(
+    request: &mut Request<serde_json::Value>,
+    headers: &BTreeMap<String, String>,
+) -> Result<(), GrpcAdapterError> {
+    for (key, value) in headers {
+        let metadata_key =
+            MetadataKey::from_str(key).map_err(|source| GrpcAdapterError::InvalidMetadataKey {
+                key: key.clone(),
+                source,
+            })?;
+        let metadata_value = MetadataValue::from_str(value).map_err(|source| {
+            GrpcAdapterError::InvalidMetadataValue {
+                key: key.clone(),
+                source,
+            }
+        })?;
+
+        request.metadata_mut().insert(metadata_key, metadata_value);
+    }
+
+    Ok(())
+}
+
+fn normalize_headers(metadata: &tonic::metadata::MetadataMap) -> BTreeMap<String, String> {
+    metadata
+        .iter()
+        .filter_map(|entry| match entry {
+            KeyAndValueRef::Ascii(key, value) => {
+                Some((key.as_str().to_owned(), value.to_str().ok()?.to_owned()))
+            }
+            KeyAndValueRef::Binary(_, _) => None,
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeMap;
+
+    use mcpaas_core::{DescriptorId, GrpcTarget};
+    use serde_json::json;
+
+    use crate::{GrpcAdapter, GrpcRequest, test_support};
+
+    #[tokio::test]
+    async fn executes_unary_grpc_request() {
+        let server_addr = test_support::spawn_unary_echo_server().await;
+        let adapter = GrpcAdapter::new();
+        let target = GrpcTarget {
+            server_addr,
+            package: "echo".to_owned(),
+            service: "EchoService".to_owned(),
+            method: "UnaryEcho".to_owned(),
+            descriptor_ref: DescriptorId::new("desc_echo"),
+            descriptor_set_b64: test_support::descriptor_set_b64(),
+        };
+        let request = GrpcRequest {
+            headers: BTreeMap::new(),
+            body: json!({ "message": "hello" }),
+            timeout_ms: 1_000,
+        };
+
+        let response = adapter.execute(&target, &request).await.unwrap();
+
+        assert_eq!(response.body, json!({ "message": "hello" }));
+    }
+}
@@ -0,0 +1,72 @@
+use prost::Message;
+use prost_reflect::{DynamicMessage, MessageDescriptor};
+use tonic::{
+    Status,
+    codec::{Codec, DecodeBuf, Decoder, EncodeBuf, Encoder},
+};
+
+pub struct JsonCodec {
+    request_descriptor: MessageDescriptor,
+    response_descriptor: MessageDescriptor,
+}
+
+impl JsonCodec {
+    pub fn new(
+        request_descriptor: MessageDescriptor,
+        response_descriptor: MessageDescriptor,
+    ) -> Self {
+        Self {
+            request_descriptor,
+            response_descriptor,
+        }
+    }
+}
+
+impl Codec for JsonCodec {
+    type Encode = serde_json::Value;
+    type Decode = serde_json::Value;
+
+    type Encoder = JsonEncoder;
+    type Decoder = JsonDecoder;
+
+    fn encoder(&mut self) -> Self::Encoder {
+        JsonEncoder(self.request_descriptor.clone())
+    }
+
+    fn decoder(&mut self) -> Self::Decoder {
+        JsonDecoder(self.response_descriptor.clone())
+    }
+}
+
+pub struct JsonEncoder(MessageDescriptor);
+
+impl Encoder for JsonEncoder {
+    type Item = serde_json::Value;
+    type Error = Status;
+
+    fn encode(&mut self, item: Self::Item, dst: &mut EncodeBuf<'_>) -> Result<(), Self::Error> {
+        let message = DynamicMessage::deserialize(self.0.clone(), item)
+            .map_err(|error| Status::invalid_argument(error.to_string()))?;
+
+        message.encode_raw(dst);
+        Ok(())
+    }
+}
+
+pub struct JsonDecoder(MessageDescriptor);
+
+impl Decoder for JsonDecoder {
+    type Item = serde_json::Value;
+    type Error = Status;
+
+    fn decode(&mut self, src: &mut DecodeBuf<'_>) -> Result<Option<Self::Item>, Self::Error> {
+        let mut message = DynamicMessage::new(self.0.clone());
+        message
+            .merge(src)
+            .map_err(|error| Status::internal(error.to_string()))?;
+
+        serde_json::to_value(&message)
+            .map(Some)
+            .map_err(|error| Status::internal(error.to_string()))
+    }
+}
@@ -0,0 +1,48 @@
+use thiserror::Error;
+use tonic::{
+    Code, Status,
+    metadata::errors::{InvalidMetadataKey, InvalidMetadataValue},
+};
+
+#[derive(Debug, Error)]
+pub enum GrpcAdapterError {
+    #[error("invalid descriptor set encoding")]
+    InvalidDescriptorEncoding,
+    #[error("invalid descriptor set")]
+    InvalidDescriptorSet,
+    #[error("invalid descriptor pool")]
+    InvalidDescriptorPool,
+    #[error("service {service} was not found in descriptor set")]
+    ServiceNotFound { service: String },
+    #[error("method {method} was not found in service {service}")]
+    MethodNotFound { service: String, method: String },
+    #[error("grpc method path is invalid for service {service} and method {method}")]
+    InvalidMethodPath { service: String, method: String },
+    #[error("grpc method {service}/{method} is not unary")]
+    UnsupportedMethodKind { service: String, method: String },
+    #[error("invalid metadata key {key}")]
+    InvalidMetadataKey {
+        key: String,
+        #[source]
+        source: InvalidMetadataKey,
+    },
+    #[error("invalid metadata value for key {key}")]
+    InvalidMetadataValue {
+        key: String,
+        #[source]
+        source: InvalidMetadataValue,
+    },
+    #[error("transport endpoint is invalid")]
+    InvalidEndpoint(#[from] tonic::transport::Error),
+    #[error("grpc status {code}: {message}")]
+    Status { code: Code, message: String },
+}
+
+impl From<Status> for GrpcAdapterError {
+    fn from(value: Status) -> Self {
+        Self::Status {
+            code: value.code(),
+            message: value.message().to_owned(),
+        }
+    }
+}
@@ -1,3 +1,11 @@
-pub fn crate_name() -> &'static str {
-    "mcpaas-adapter-grpc"
-}
+mod client;
+mod codec;
+mod error;
+mod model;
+
+pub use client::GrpcAdapter;
+pub use error::GrpcAdapterError;
+pub use model::{GrpcRequest, GrpcResponse};
+
+#[cfg(any(test, feature = "test-support"))]
+pub mod test_support;
@@ -0,0 +1,20 @@
+use std::collections::BTreeMap;
+
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default)]
+pub struct GrpcRequest {
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    pub headers: BTreeMap<String, String>,
+    pub body: Value,
+    pub timeout_ms: u64,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct GrpcResponse {
+    pub status_code: u16,
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    pub headers: BTreeMap<String, String>,
+    pub body: Value,
+}
@@ -0,0 +1,45 @@
+use base64::{Engine as _, engine::general_purpose::STANDARD};
+use tokio::net::TcpListener;
+use tonic::{Request, Response, Status, transport::Server};
+
+pub mod echo {
+    tonic::include_proto!("echo");
+    pub const FILE_DESCRIPTOR_SET: &[u8] = tonic::include_file_descriptor_set!("echo_descriptor");
+}
+
+#[derive(Default)]
+pub struct EchoServiceImpl;
+
+#[tonic::async_trait]
+impl echo::echo_service_server::EchoService for EchoServiceImpl {
+    async fn unary_echo(
+        &self,
+        request: Request<echo::EchoRequest>,
+    ) -> Result<Response<echo::EchoResponse>, Status> {
+        Ok(Response::new(echo::EchoResponse {
+            message: request.into_inner().message,
+        }))
+    }
+}
+
+pub async fn spawn_unary_echo_server() -> String {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    let incoming = tonic::transport::server::TcpIncoming::from(listener);
+
+    tokio::spawn(async move {
+        Server::builder()
+            .add_service(echo::echo_service_server::EchoServiceServer::new(
+                EchoServiceImpl,
+            ))
+            .serve_with_incoming(incoming)
+            .await
+            .unwrap();
+    });
+
+    format!("http://{}", address)
+}
+
+pub fn descriptor_set_b64() -> String {
+    STANDARD.encode(echo::FILE_DESCRIPTOR_SET)
+}