sova_grpc/
client.rs

1//! Outbound Connect-JSON client.
2
3use crate::error::GrpcError;
4use crate::fake::FakeGrpc;
5use crate::transport::GrpcTransport;
6use bytes::Bytes;
7use serde::{de::DeserializeOwned, Serialize};
8use sova_core::extend::BoxFuture;
9use std::sync::Arc;
10
11struct HttpTransport {
12    client: reqwest::Client,
13}
14
15impl GrpcTransport for HttpTransport {
16    fn call(&self, base: &str, method: &str, body: Bytes) -> BoxFuture<Result<Bytes, GrpcError>> {
17        let client = self.client.clone();
18        let url = format!(
19            "{}/{}",
20            base.trim_end_matches('/'),
21            method.trim_start_matches('/')
22        );
23        Box::pin(async move {
24            let res = client
25                .post(url)
26                .header(http::header::CONTENT_TYPE, "application/json")
27                .header("connect-protocol-version", "1")
28                .body(body)
29                .send()
30                .await
31                .map_err(|e| GrpcError::Transport(e.to_string()))?;
32            let status = res.status().as_u16();
33            let bytes = res
34                .bytes()
35                .await
36                .map_err(|e| GrpcError::Transport(e.to_string()))?;
37            if !(200..300).contains(&status) {
38                let body_str = String::from_utf8_lossy(&bytes).into_owned();
39                if let Some(parsed) = crate::error_envelope::parse_connect_error(&body_str) {
40                    return Err(GrpcError::Rpc {
41                        code: parsed.code,
42                        message: parsed.message,
43                    });
44                }
45                return Err(GrpcError::Http {
46                    status,
47                    body: body_str,
48                });
49            }
50            Ok(bytes)
51        })
52    }
53}
54
55/// Shared client in app state.
56#[derive(Clone)]
57pub struct GrpcClient {
58    base: String,
59    transport: Arc<dyn GrpcTransport>,
60    fake: Option<FakeGrpc>,
61}
62
63impl GrpcClient {
64    pub fn base(&self) -> &str {
65        &self.base
66    }
67
68    pub fn fake(&self) -> Option<&FakeGrpc> {
69        self.fake.as_ref()
70    }
71
72    pub(crate) fn http(base: impl Into<String>) -> Self {
73        Self {
74            base: base.into(),
75            transport: Arc::new(HttpTransport {
76                client: reqwest::Client::new(),
77            }),
78            fake: None,
79        }
80    }
81
82    pub(crate) fn with_fake(base: impl Into<String>, fake: FakeGrpc) -> Self {
83        Self {
84            base: base.into(),
85            transport: Arc::new(fake.clone()),
86            fake: Some(fake),
87        }
88    }
89
90    pub async fn call<Req, Res>(&self, method: &str, req: &Req) -> Result<Res, GrpcError>
91    where
92        Req: Serialize,
93        Res: DeserializeOwned,
94    {
95        let body =
96            Bytes::from(serde_json::to_vec(req).map_err(|e| GrpcError::Decode(e.to_string()))?);
97        let bytes = self.transport.call(&self.base, method, body).await?;
98        serde_json::from_slice(&bytes).map_err(|e| GrpcError::Decode(e.to_string()))
99    }
100
101    pub async fn call_raw(&self, method: &str, body: Bytes) -> Result<Bytes, GrpcError> {
102        let started = std::time::Instant::now();
103        let bytes_in = body.len() as u64;
104        let result = self.transport.call(&self.base, method, body).await;
105        crate::trace::emit_client(
106            method,
107            &self.base,
108            started.elapsed().as_secs_f64() * 1000.0,
109            &result,
110            bytes_in,
111        );
112        result
113    }
114}
sova_grpc/client.rs

sova_grpc/
client.rs
