wtx 0.45.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

use crate::{
  http::OptionedServer,
  misc::FnFut,
  pool::{SimplePool, WebSocketRM},
  rng::Xorshift64,
  stream::Stream,
  web_socket::{Compression, WebSocket, WebSocketAcceptor, WebSocketBuffer},
};
use core::fmt::Debug;
use std::sync::OnceLock;
use tokio::net::{TcpListener, TcpStream};

static POOL: OnceLock<SimplePool<WebSocketRM>> = OnceLock::new();

impl OptionedServer {
  /// Optioned WebSocket server using tokio.
  #[inline]
  pub async fn web_socket_tokio<ACPT, C, E, H, N, S>(
    addr: &str,
    buffers_len_opt: Option<usize>,
    compression_cb: impl Clone + Fn() -> C + Send + 'static,
    err_cb: impl Clone + Fn(E) + Send + 'static,
    handle_cb: H,
    (acceptor_cb, net_cb): (
      impl FnOnce() -> crate::Result<ACPT> + Send + 'static,
      impl Clone + Fn(ACPT, TcpStream) -> N + Send + 'static,
    ),
  ) -> crate::Result<()>
  where
    ACPT: Clone + Send + 'static,
    C: Compression<false> + Send + 'static,
    C::NegotiatedCompression: Send,
    E: Debug + From<crate::Error> + Send + 'static,
    for<'wsb> H: Clone
      + FnFut<
        (WebSocket<C::NegotiatedCompression, Xorshift64, S, &'wsb mut WebSocketBuffer, false>,),
        Result = Result<(), E>,
      > + Send
      + 'static,
    N: Send + Future<Output = crate::Result<S>>,
    S: Stream<read(..): Send, write_all(..): Send> + Send,
    for<'wsb> <H as FnFut<(
      WebSocket<C::NegotiatedCompression, Xorshift64, S, &'wsb mut WebSocketBuffer, false>,
    )>>::Future: Send,
    for<'handle> &'handle H: Send,
  {
    let buffers_len = number_or_available_parallelism(buffers_len_opt)?;
    let listener = TcpListener::bind(addr).await?;
    let acceptor = acceptor_cb()?;
    loop {
      let conn_acceptor = acceptor.clone();
      let conn_compression_cb = compression_cb.clone();
      let conn_conn_err = err_cb.clone();
      let conn_handle_cb = handle_cb.clone();
      let conn_net_cb = net_cb.clone();
      let tcp_stream = listener.accept().await?.0;
      let mut conn_buffer = POOL
        .get_or_init(|| SimplePool::new(buffers_len, WebSocketRM::new(|| Ok(Default::default()))))
        .get_with_unit()
        .await?;
      let _jh = tokio::spawn(async move {
        let wsb = &mut ***conn_buffer;
        let fun = async move {
          let net = conn_net_cb(conn_acceptor, tcp_stream).await?;
          conn_handle_cb
            .call((WebSocketAcceptor::default()
              .compression(conn_compression_cb())
              .no_masking(true)
              .wsb(wsb)
              .accept(net)
              .await?,))
            .await?;
          Ok::<_, E>(())
        };
        if let Err(err) = fun.await {
          conn_conn_err(err);
        }
      });
    }
  }
}

pub(crate) fn number_or_available_parallelism(n: Option<usize>) -> crate::Result<usize> {
  Ok(if let Some(elem) = n { elem } else { usize::from(std::thread::available_parallelism()?) })
}