wtx 0.50.0

A collection of different transport implementations and related tools focused primarily on web technologies.
Documentation 
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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

use crate::{
  executor::Executor,
  http::{
    HttpRecvParams,
    http2_client_pool::{Http2ClientPoolResource, Http2Resource},
  },
  http2::{Http2, Http2Buffer},
  misc::{TcpParams, UriRef},
  pool::ResourceManager,
  rng::ChaCha20,
  stream::{Stream, StreamReader, StreamWriter},
  sync::{AsyncMutex, AtomicCell},
  tls::{TlsConfig, TlsConnectorBuilder, TlsMode, TlsStream},
};
use core::fmt::Debug;

/// Resource manager for `ClientPool`.
#[derive(Debug)]
pub struct Http2RM<EX, TM> {
  pub(crate) disable_auto_sni: bool,
  pub(crate) executor: EX,
  pub(crate) hrp: HttpRecvParams,
  pub(crate) rng: AtomicCell<ChaCha20>,
  pub(crate) tcp_params: TcpParams,
  pub(crate) tls_config: AsyncMutex<TlsConfig<TM>>,
}

impl<EX, TM> Http2RM<EX, TM>
where
  EX: Executor,
  TM: TlsMode,
{
  async fn tls_stream(&self, aux: &str) -> crate::Result<TlsStream<EX::TcpStream, TM, true>> {
    let uri = UriRef::new(aux);
    let mut tls_config = self.tls_config.lock().await;
    if !self.disable_auto_sni {
      push_server_name(&mut tls_config, &uri)?;
    }
    Ok(
      TlsConnectorBuilder::new(EX::default(), uri)
        .set_tcp_params(self.tcp_params)
        .build(&*tls_config, &self.rng)
        .await?
        .connect()
        .await?
        .tls_stream,
    )
  }
}

impl<EX, TM> ResourceManager for Http2RM<EX, TM>
where
  EX: Executor,
  EX::TcpStream: 'static,
  TM: Clone + Send + TlsMode + 'static,
  <EX::TcpStream as Stream>::ReadHalfOwned: Send,
  <EX::TcpStream as Stream>::WriteHalfOwned: Send,
  <<EX::TcpStream as Stream>::ReadHalfOwned as StreamReader>::read(..): Send,
  <<EX::TcpStream as Stream>::WriteHalfOwned as StreamWriter>::write_all(..): Send,
  <<EX::TcpStream as Stream>::WriteHalfOwned as StreamWriter>::write_all_vectored(..): Send,
{
  type CreateAux = str;
  type Error = crate::Error;
  type RecycleAux = str;
  type Resource = Http2Resource<<EX::TcpStream as Stream>::WriteHalfOwned, TM>;

  #[inline]
  async fn create(&self, aux: &Self::CreateAux) -> Result<Self::Resource, Self::Error> {
    let tls_stream = self.tls_stream(aux).await?;
    let tuple = Http2::connect(Http2Buffer::default(), self.hrp, tls_stream.into_split()?).await?;
    let _jh = self.executor.spawn(tuple.0);
    Ok(Http2ClientPoolResource { client: tuple.1 })
  }

  #[inline]
  fn is_invalid(&self, resource: &Self::Resource) -> bool {
    resource.client.connection_state().is_closed()
  }

  #[inline]
  async fn recycle(
    &self,
    aux: &Self::RecycleAux,
    resource: &mut Self::Resource,
  ) -> Result<(), Self::Error> {
    let tls_stream = self.tls_stream(aux).await?;
    let mut hb = Http2Buffer::default();
    resource.client.swap_buffers(&mut hb).await;
    let (frame_reader, http2) = Http2::connect(hb, self.hrp, tls_stream.into_split()?).await?;
    let _jh = self.executor.spawn(frame_reader);
    resource.client = http2;
    Ok(())
  }
}

fn push_server_name<S, TM>(tc: &mut TlsConfig<TM>, uri: &crate::misc::Uri<S>) -> crate::Result<()>
where
  S: crate::misc::Lease<str>,
{
  if tc.server_name_mut().is_some() {
    return Ok(());
  }
  tc.server_name_mut()
    .get_or_insert_default()
    .server_name_list
    .push(crate::tls::ServerName::from_name(uri.hostname().try_into()?))?;
  Ok(())
}