grpc_webnext_client/
typed.rs

1//! Typed helpers over `prost::Message`.
2//!
3//! Thin by design: encoding is `prost`'s job and framing is `codec`'s, so this is
4//! only the two conversions plus turning a decode failure into a gRPC status rather
5//! than a panic — a server that sends something unparseable is INTERNAL, not a
6//! crash in the tab.
7
8use futures::stream::{Stream, StreamExt};
9use prost::Message;
10
11use crate::client::{CallOptions, Client, UnaryResponse};
12use crate::metadata::Metadata;
13use crate::status::{Code, Status};
14
15/// A decoded response plus both metadata blocks.
16#[derive(Debug, Clone)]
17pub struct TypedResponse<M> {
18    pub message: M,
19    pub headers: Metadata,
20    pub trailers: Metadata,
21}
22
23impl<M> TypedResponse<M> {
24    /// Discard the metadata when only the message matters.
25    pub fn into_inner(self) -> M {
26        self.message
27    }
28}
29
30/// Typed calls over a [`Client`].
31pub trait TypedClient {
32    /// Unary with prost encode/decode.
33    fn unary_typed<Req, Res>(
34        &self,
35        path: &str,
36        request: Req,
37        options: CallOptions,
38    ) -> impl std::future::Future<Output = Result<TypedResponse<Res>, Status>>
39    where
40        Req: Message,
41        Res: Message + Default;
42
43    /// Client streaming with prost encode/decode.
44    fn client_streaming_typed<Req, Res, S>(
45        &self,
46        path: &str,
47        requests: S,
48        options: CallOptions,
49    ) -> impl std::future::Future<Output = Result<TypedResponse<Res>, Status>>
50    where
51        Req: Message + 'static,
52        Res: Message + Default,
53        S: Stream<Item = Req> + 'static;
54}
55
56impl TypedClient for Client {
57    async fn unary_typed<Req, Res>(
58        &self,
59        path: &str,
60        request: Req,
61        options: CallOptions,
62    ) -> Result<TypedResponse<Res>, Status>
63    where
64        Req: Message,
65        Res: Message + Default,
66    {
67        let response = self.unary(path, request.encode_to_vec(), options).await?;
68        decode(response)
69    }
70
71    async fn client_streaming_typed<Req, Res, S>(
72        &self,
73        path: &str,
74        requests: S,
75        options: CallOptions,
76    ) -> Result<TypedResponse<Res>, Status>
77    where
78        Req: Message + 'static,
79        Res: Message + Default,
80        S: Stream<Item = Req> + 'static,
81    {
82        let response = self
83            .client_streaming(path, requests.map(|m| m.encode_to_vec()), options)
84            .await?;
85        decode(response)
86    }
87}
88
89fn decode<Res: Message + Default>(response: UnaryResponse) -> Result<TypedResponse<Res>, Status> {
90    let message = Res::decode(response.message.as_slice()).map_err(|e| {
91        Status::new(Code::Internal, format!("failed to decode response message: {e}"))
92    })?;
93    Ok(TypedResponse { message, headers: response.headers, trailers: response.trailers })
94}
grpc_webnext_client/typed.rs

grpc_webnext_client/
typed.rs
