1 use bytes::Bytes; 2 use http::header::CONTENT_TYPE; 3 use http::{Request, Response, Version}; 4 use http_body::Body; 5 use pin_project::pin_project; 6 use std::error::Error; 7 use std::future::Future; 8 use std::pin::Pin; 9 use std::task::{ready, Context, Poll}; 10 use tower_layer::Layer; 11 use tower_service::Service; 12 use tracing::debug; 13 14 use crate::call::content_types::GRPC_WEB; 15 use crate::call::GrpcWebCall; 16 17 /// Layer implementing the grpc-web protocol for clients. 18 #[derive(Debug, Clone)] 19 pub struct GrpcWebClientLayer { 20 _priv: (), 21 } 22 23 impl GrpcWebClientLayer { 24 /// Create a new grpc-web for clients layer. 25 pub fn new() -> GrpcWebClientLayer { 26 Self { _priv: () } 27 } 28 } 29 30 impl Default for GrpcWebClientLayer { 31 fn default() -> Self { 32 Self::new() 33 } 34 } 35 36 impl<S> Layer<S> for GrpcWebClientLayer { 37 type Service = GrpcWebClientService<S>; 38 39 fn layer(&self, inner: S) -> Self::Service { 40 GrpcWebClientService::new(inner) 41 } 42 } 43 44 /// A [`Service`] that wraps some inner http service that will 45 /// coerce requests coming from [`tonic::client::Grpc`] into proper 46 /// `grpc-web` requests. 47 #[derive(Debug, Clone)] 48 pub struct GrpcWebClientService<S> { 49 inner: S, 50 } 51 52 impl<S> GrpcWebClientService<S> { 53 /// Create a new grpc-web for clients service. 54 pub fn new(inner: S) -> Self { 55 Self { inner } 56 } 57 } 58 59 impl<S, B1, B2> Service<Request<B1>> for GrpcWebClientService<S> 60 where 61 S: Service<Request<GrpcWebCall<B1>>, Response = Response<B2>>, 62 B1: Body, 63 B2: Body<Data = Bytes>, 64 B2::Error: Error, 65 { 66 type Response = Response<GrpcWebCall<B2>>; 67 type Error = S::Error; 68 type Future = ResponseFuture<S::Future>; 69 70 fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { 71 self.inner.poll_ready(cx) 72 } 73 74 fn call(&mut self, mut req: Request<B1>) -> Self::Future { 75 if req.version() == Version::HTTP_2 { 76 debug!("coercing HTTP2 request to HTTP1.1"); 77 78 *req.version_mut() = Version::HTTP_11; 79 } 80 81 req.headers_mut() 82 .insert(CONTENT_TYPE, GRPC_WEB.try_into().unwrap()); 83 84 let req = req.map(GrpcWebCall::client_request); 85 86 let fut = self.inner.call(req); 87 88 ResponseFuture { inner: fut } 89 } 90 } 91 92 /// Response future for the [`GrpcWebService`]. 93 #[allow(missing_debug_implementations)] 94 #[pin_project] 95 #[must_use = "futures do nothing unless polled"] 96 pub struct ResponseFuture<F> { 97 #[pin] 98 inner: F, 99 } 100 101 impl<F, B, E> Future for ResponseFuture<F> 102 where 103 B: Body<Data = Bytes>, 104 F: Future<Output = Result<Response<B>, E>>, 105 { 106 type Output = Result<Response<GrpcWebCall<B>>, E>; 107 108 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { 109 let res = ready!(self.project().inner.poll(cx)); 110 111 Poll::Ready(res.map(|r| r.map(GrpcWebCall::client_response))) 112 } 113 } 114