wtx 0.28.0

A collection of different transport implementations and related tools focused primarily on web technologies.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

use crate::{
  data_transformation::{
    dnsn::{De, DecodeWrapper},
    format::{VerbatimRequest, VerbatimResponse},
  },
  grpc::serialize,
  http::{
    Header, Headers, HttpClient, KnownHeaderName, Method, ReqResBuffer, Response, WTX_USER_AGENT,
  },
  http2::{Http2, Http2Buffer, Http2Data},
  misc::{Decode, Encode, LeaseMut, Lock, RefCounter, SingleTypeStorage, StreamWriter, UriRef},
};

/// Performs requests to gRPC servers.
#[derive(Debug)]
pub struct GrpcClient<C, DRSR> {
  client: C,
  drsr: DRSR,
}

impl<C, DRSR, HD, SW> GrpcClient<C, DRSR>
where
  C: LeaseMut<Http2<HD, true>> + SingleTypeStorage<Item = HD>,
  HD: RefCounter,
  HD::Item: Lock<Resource = Http2Data<Http2Buffer, SW, true>>,
  SW: StreamWriter,
{
  /// Constructor
  #[inline]
  pub fn new(client: C, drsr: DRSR) -> Self {
    Self { client, drsr }
  }

  /// Deserialize From Response Bytes
  #[inline]
  pub fn des_from_res_bytes<'de, T>(&mut self, bytes: &mut &'de [u8]) -> crate::Result<T>
  where
    VerbatimResponse<T>: Decode<'de, De<DRSR>>,
  {
    let elem = if let [_, _, _, _, _, elem @ ..] = bytes { elem } else { &[] };
    Ok(VerbatimResponse::decode(&mut self.drsr, &mut DecodeWrapper::new(elem))?.data)
  }

  /// Send Unary Request
  ///
  /// Builds a valid unary gRPC request and awaits for a raw response.
  ///
  /// It is necessary to call [`Self::des_from_res_bytes`] to create the corresponding decoded element.
  #[inline]
  pub async fn send_unary_req<T>(
    &mut self,
    data: T,
    mut rrb: ReqResBuffer,
    uri: &UriRef<'_>,
  ) -> crate::Result<Response<ReqResBuffer>>
  where
    VerbatimRequest<T>: Encode<De<DRSR>>,
  {
    rrb.clear();
    serialize(&mut rrb.body, VerbatimRequest { data }, &mut self.drsr)?;
    Self::push_headers(&mut rrb.headers)?;
    let res = self.client.lease_mut().send_recv_single(Method::Post, rrb, uri).await?;
    Ok(Response::http2(res.rrd, res.status_code))
  }

  #[inline]
  fn push_headers(headers: &mut Headers) -> crate::Result<()> {
    headers.push_from_iter_many([
      Header::from_name_and_value(
        KnownHeaderName::ContentType.into(),
        ["application/grpc".as_bytes()].into_iter(),
      ),
      Header::from_name_and_value(KnownHeaderName::Te.into(), ["trailers".as_bytes()].into_iter()),
      Header::from_name_and_value(
        KnownHeaderName::UserAgent.into(),
        [WTX_USER_AGENT.as_bytes()].into_iter(),
      ),
    ])?;
    Ok(())
  }
}