moqtap_client/transport/
mod.rs

1//! Transport abstraction for QUIC and WebTransport.
2//!
3//! Uses enum dispatch (not trait objects) since the transport set is closed.
4//! WebTransport support is behind the `webtransport` feature flag.
5
6pub mod quic;
7#[cfg(feature = "webtransport")]
8pub mod webtransport;
9
10use bytes::Bytes;
11
12/// Errors from the transport layer.
13#[derive(Debug, thiserror::Error)]
14pub enum TransportError {
15    /// Connection-level error (e.g., peer closed, timeout).
16    #[error("connection error: {0}")]
17    Connection(String),
18    /// Error writing to a stream.
19    #[error("write error: {0}")]
20    Write(String),
21    /// Error reading from a stream.
22    #[error("read error: {0}")]
23    Read(String),
24    /// Stream was closed.
25    #[error("stream closed")]
26    StreamClosed,
27    /// Error sending a datagram.
28    #[error("send datagram error: {0}")]
29    SendDatagram(String),
30    /// Connection was lost.
31    #[error("connection lost")]
32    ConnectionLost,
33    /// Error during connection establishment.
34    #[error("connect error: {0}")]
35    Connect(String),
36}
37
38/// A transport-agnostic connection (QUIC or WebTransport).
39pub enum Transport {
40    /// Raw QUIC via quinn.
41    Quic(quic::QuicTransport),
42    /// WebTransport via h3 + h3-quinn.
43    #[cfg(feature = "webtransport")]
44    WebTransport(webtransport::WebTransportTransport),
45}
46
47impl Transport {
48    /// Open a bidirectional stream.
49    pub async fn open_bi(&self) -> Result<(SendStream, RecvStream), TransportError> {
50        match self {
51            Transport::Quic(t) => t.open_bi().await,
52            #[cfg(feature = "webtransport")]
53            Transport::WebTransport(t) => t.open_bi().await,
54        }
55    }
56
57    /// Accept an incoming bidirectional stream.
58    pub async fn accept_bi(&self) -> Result<(SendStream, RecvStream), TransportError> {
59        match self {
60            Transport::Quic(t) => t.accept_bi().await,
61            #[cfg(feature = "webtransport")]
62            Transport::WebTransport(t) => t.accept_bi().await,
63        }
64    }
65
66    /// Open a unidirectional send stream.
67    pub async fn open_uni(&self) -> Result<SendStream, TransportError> {
68        match self {
69            Transport::Quic(t) => t.open_uni().await,
70            #[cfg(feature = "webtransport")]
71            Transport::WebTransport(t) => t.open_uni().await,
72        }
73    }
74
75    /// Accept an incoming unidirectional stream.
76    pub async fn accept_uni(&self) -> Result<RecvStream, TransportError> {
77        match self {
78            Transport::Quic(t) => t.accept_uni().await,
79            #[cfg(feature = "webtransport")]
80            Transport::WebTransport(t) => t.accept_uni().await,
81        }
82    }
83
84    /// Send a datagram.
85    pub fn send_datagram(&self, data: Bytes) -> Result<(), TransportError> {
86        match self {
87            Transport::Quic(t) => t.send_datagram(data),
88            #[cfg(feature = "webtransport")]
89            Transport::WebTransport(t) => t.send_datagram(data),
90        }
91    }
92
93    /// Receive a datagram.
94    pub async fn recv_datagram(&self) -> Result<Bytes, TransportError> {
95        match self {
96            Transport::Quic(t) => t.recv_datagram().await,
97            #[cfg(feature = "webtransport")]
98            Transport::WebTransport(t) => t.recv_datagram().await,
99        }
100    }
101
102    /// Close the connection.
103    pub fn close(&self, code: u32, reason: &[u8]) {
104        match self {
105            Transport::Quic(t) => t.close(code, reason),
106            #[cfg(feature = "webtransport")]
107            Transport::WebTransport(t) => t.close(code, reason),
108        }
109    }
110}
111
112/// A transport-agnostic send stream.
113pub enum SendStream {
114    /// Raw QUIC send stream.
115    Quic(quinn::SendStream),
116    /// WebTransport send stream.
117    #[cfg(feature = "webtransport")]
118    WebTransport(webtransport::WtSendStream),
119}
120
121impl SendStream {
122    /// Get the QUIC stream ID (transport-level identifier).
123    pub fn stream_id(&self) -> u64 {
124        match self {
125            SendStream::Quic(s) => s.id().index(),
126            #[cfg(feature = "webtransport")]
127            SendStream::WebTransport(_) => 0, // WebTransport doesn't expose stream IDs
128        }
129    }
130
131    /// Write all bytes to the stream.
132    pub async fn write_all(&mut self, buf: &[u8]) -> Result<(), TransportError> {
133        match self {
134            SendStream::Quic(s) => {
135                s.write_all(buf).await.map_err(|e| TransportError::Write(e.to_string()))
136            }
137            #[cfg(feature = "webtransport")]
138            SendStream::WebTransport(s) => s.write_all(buf).await,
139        }
140    }
141
142    /// Finish the stream (send FIN).
143    pub fn finish(&mut self) -> Result<(), TransportError> {
144        match self {
145            SendStream::Quic(s) => {
146                s.finish().map_err(|_| TransportError::StreamClosed)?;
147                Ok(())
148            }
149            #[cfg(feature = "webtransport")]
150            SendStream::WebTransport(s) => s.finish(),
151        }
152    }
153}
154
155/// A transport-agnostic receive stream.
156pub enum RecvStream {
157    /// Raw QUIC receive stream.
158    Quic(quinn::RecvStream),
159    /// WebTransport receive stream.
160    #[cfg(feature = "webtransport")]
161    WebTransport(webtransport::WtRecvStream),
162}
163
164impl RecvStream {
165    /// Get the QUIC stream ID (transport-level identifier).
166    pub fn stream_id(&self) -> u64 {
167        match self {
168            RecvStream::Quic(s) => s.id().index(),
169            #[cfg(feature = "webtransport")]
170            RecvStream::WebTransport(_) => 0,
171        }
172    }
173
174    /// Read data into the buffer. Returns `Ok(Some(n))` with bytes read,
175    /// `Ok(None)` on stream end, or `Err` on failure.
176    pub async fn read(&mut self, buf: &mut [u8]) -> Result<Option<usize>, TransportError> {
177        match self {
178            RecvStream::Quic(s) => {
179                s.read(buf).await.map_err(|e| TransportError::Read(e.to_string()))
180            }
181            #[cfg(feature = "webtransport")]
182            RecvStream::WebTransport(s) => s.read(buf).await,
183        }
184    }
185}
moqtap_client/transport/mod.rs

moqtap_client/transport/
mod.rs
