use std::sync::Arc;
use tonic::client::{Grpc, GrpcService};
use tonic::codec::ProstCodec;
use tonic::codegen::{Body, Bytes, StdError};
use tonic::metadata::{Ascii, MetadataValue};
use crate::auth::{AuthorizationRequest, RequestAuthorizer};
use super::canonical::{GRPC_CONTENT_TYPE, grpc_canonical_request_string, grpc_method_path};
use super::error::GrpcClientError;
const SCHEMA_SHA_HEADER: &str = "x-cratestack-schema-sha";
#[derive(Clone)]
pub struct CratestackGrpcClient<T> {
inner: Grpc<T>,
package: &'static str,
request_authorizer: Option<Arc<dyn RequestAuthorizer>>,
schema_sha: Option<&'static str>,
}
impl<T> CratestackGrpcClient<T> {
pub fn new(inner: T, package: &'static str) -> Self {
Self {
inner: Grpc::new(inner),
package,
request_authorizer: None,
schema_sha: None,
}
}
pub fn with_request_authorizer(
mut self,
request_authorizer: Arc<dyn RequestAuthorizer>,
) -> Self {
self.request_authorizer = Some(request_authorizer);
self
}
pub fn with_schema_sha(mut self, schema_sha: &'static str) -> Self {
self.schema_sha = Some(schema_sha);
self
}
pub async fn unary<Req, Resp>(
&mut self,
method_name: &str,
message: Req,
) -> Result<Resp, GrpcClientError>
where
T: GrpcService<tonic::body::Body>,
T::Error: Into<StdError>,
T::ResponseBody: Body<Data = Bytes> + Send + 'static,
<T::ResponseBody as Body>::Error: Into<StdError> + Send,
Req: prost::Message + Clone + Default + 'static,
Resp: prost::Message + Default + 'static,
{
self.inner
.ready()
.await
.map_err(|error| GrpcClientError::Transport(error.into()))?;
let path = tonic::codegen::http::uri::PathAndQuery::try_from(grpc_method_path(
self.package,
method_name,
))
.map_err(|error| GrpcClientError::BadInput(error.to_string()))?;
let mut request = tonic::Request::new(message.clone());
if let Some(authorizer) = &self.request_authorizer {
let body = prost::Message::encode_to_vec(&message);
let canonical_request = grpc_canonical_request_string(self.package, method_name, &body);
let authorization_request = AuthorizationRequest {
method: "POST".to_owned(),
path: grpc_method_path(self.package, method_name),
canonical_query: None,
content_type: Some(GRPC_CONTENT_TYPE.to_owned()),
body,
canonical_request,
};
let headers = authorizer
.authorize(&authorization_request)
.map_err(|error| GrpcClientError::BadInput(error.to_string()))?;
let metadata = request.metadata_mut();
for (name, value) in headers {
let key = tonic::metadata::MetadataKey::<Ascii>::from_bytes(name.as_bytes())
.map_err(|error| {
GrpcClientError::BadInput(format!("invalid header name '{name}': {error}"))
})?;
let value = MetadataValue::try_from(value.as_str()).map_err(|error| {
GrpcClientError::BadInput(format!("invalid header value for '{name}': {error}"))
})?;
metadata.insert(key, value);
}
}
if let Some(schema_sha) = self.schema_sha {
let value = MetadataValue::try_from(schema_sha)
.map_err(|error| GrpcClientError::BadInput(error.to_string()))?;
request.metadata_mut().insert(SCHEMA_SHA_HEADER, value);
}
let codec = ProstCodec::default();
let response = self.inner.unary(request, path, codec).await?;
Ok(response.into_inner())
}
}
impl<T: std::fmt::Debug> std::fmt::Debug for CratestackGrpcClient<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("CratestackGrpcClient")
.field("inner", &self.inner)
.field("package", &self.package)
.field("request_authorizer", &self.request_authorizer.is_some())
.field("schema_sha", &self.schema_sha)
.finish()
}
}