wrpc_webtransport/
lib.rs

1//! wRPC WebTransport transport
2
3use core::ops::{Deref, DerefMut};
4
5use anyhow::Context as _;
6use bytes::Bytes;
7use quinn::VarInt;
8use tokio::io::BufReader;
9use tracing::{debug, error, trace, warn};
10use wrpc_transport::Invoke;
11use wrpc_transport::frame::{Incoming, InvokeBuilder, Outgoing};
12use wtransport::{Connection, RecvStream, SendStream};
13
14/// WebTransport server with graceful stream shutdown handling
15pub type Server = wrpc_transport::Server<(), BufReader<RecvStream>, SendStream, ConnHandler>;
16
17/// WebTransport wRPC client
18#[derive(Clone, Debug)]
19pub struct Client(Connection);
20
21impl From<Connection> for Client {
22    fn from(session: Connection) -> Self {
23        Self(session)
24    }
25}
26
27impl Deref for Client {
28    type Target = Connection;
29
30    fn deref(&self) -> &Self::Target {
31        &self.0
32    }
33}
34
35impl DerefMut for Client {
36    fn deref_mut(&mut self) -> &mut Self::Target {
37        &mut self.0
38    }
39}
40
41/// Graceful stream shutdown handler
42pub struct ConnHandler;
43
44impl wrpc_transport::frame::ConnHandler<BufReader<RecvStream>, SendStream> for ConnHandler {
45    const DRAIN: bool = false;
46
47    async fn on_ingress(mut rx: BufReader<RecvStream>, res: std::io::Result<()>) {
48        if let Err(err) = res {
49            error!(?err, "ingress failed");
50        } else {
51            debug!("ingress successfully complete");
52        }
53        if let Ok(code) = VarInt::from_u64(0x52e4a40fa8db)
54            && let Err(err) = rx.get_mut().quic_stream_mut().stop(code)
55        {
56            debug!(?err, "failed to close incoming stream");
57        }
58    }
59
60    async fn on_egress(mut tx: SendStream, res: std::io::Result<()>) {
61        if let Err(err) = res {
62            error!(?err, "egress failed");
63        } else {
64            debug!("egress successfully complete");
65        }
66        match tx.quic_stream_mut().stopped().await {
67            Ok(None) => {
68                trace!("stream successfully closed");
69            }
70            Ok(Some(code)) => {
71                if u64::from(code) == 0x52e4a40fa8db {
72                    trace!("stream successfully closed");
73                } else {
74                    warn!(?code, "stream closed with code");
75                }
76            }
77            Err(err) => {
78                error!(?err, "failed to await stream close");
79            }
80        }
81    }
82}
83
84impl Invoke for &Client {
85    type Context = ();
86
87    async fn invoke<P>(
88        &self,
89        (): Self::Context,
90        instance: &str,
91        func: &str,
92        params: Bytes,
93        paths: impl AsRef<[P]> + Send,
94    ) -> anyhow::Result<(Outgoing, Incoming)>
95    where
96        P: AsRef<[Option<usize>]> + Send + Sync,
97    {
98        let stream = self
99            .0
100            .open_bi()
101            .await
102            .context("failed to initialize parameter stream")?;
103        let (tx, rx) = stream.await.context("failed to open parameter stream")?;
104        InvokeBuilder::<ConnHandler>::default()
105            .invoke(tx, BufReader::new(rx), instance, func, params, paths)
106            .await
107    }
108}
109
110impl Invoke for Client {
111    type Context = ();
112
113    async fn invoke<P>(
114        &self,
115        (): Self::Context,
116        instance: &str,
117        func: &str,
118        params: Bytes,
119        paths: impl AsRef<[P]> + Send,
120    ) -> anyhow::Result<(Outgoing, Incoming)>
121    where
122        P: AsRef<[Option<usize>]> + Send + Sync,
123    {
124        (&self).invoke((), instance, func, params, paths).await
125    }
126}
wrpc_webtransport/lib.rs

wrpc_webtransport/
lib.rs
