moq_native/
tcp.rs

1//! Plain-TCP qmux transport, reachable via the `tcp://` URL scheme.
2//!
3//! Runs the QMux wire format directly over TCP with no TLS or WebSocket
4//! framing. There is no transport encryption and no authentication, so only
5//! use this on a trusted network (loopback, a private VPC interface, etc.).
6//!
7//! TCP has no TLS handshake, so the application protocol (the moq ALPN) is
8//! negotiated in-band: pass the offered/supported protocols and the resulting
9//! `qmux::Session::protocol()` is populated before connect/accept returns.
10
11use std::net;
12use url::Url;
13
14/// The QMux wire-format version both ends speak over a raw stream. Fixed (not
15/// negotiated) since there's no TLS ALPN to carry it.
16const WIRE_VERSION: qmux::Version = qmux::Version::QMux01;
17
18/// Errors specific to the plain-TCP qmux transport.
19#[derive(Debug, thiserror::Error)]
20#[non_exhaustive]
21pub enum Error {
22	/// The TCP socket failed to bind, accept, or connect.
23	#[error(transparent)]
24	Io(#[from] std::io::Error),
25
26	/// The `tcp://` URL had no host.
27	#[error("missing hostname")]
28	MissingHostname,
29
30	/// The `tcp://` URL had no port. Unlike `https`, there is no default.
31	#[error("missing port")]
32	MissingPort,
33
34	/// The qmux handshake failed while dialing.
35	#[error("qmux connect failed")]
36	Connect(#[source] qmux::Error),
37
38	/// The qmux handshake failed while accepting.
39	#[error("qmux accept failed")]
40	Accept(#[source] qmux::Error),
41}
42
43type Result<T> = std::result::Result<T, Error>;
44
45/// Dial a `tcp://host:port` URL, advertising `protocols` for in-band ALPN
46/// negotiation. Returns a qmux session over plain TCP.
47///
48/// The port is required; there is no default for the `tcp` scheme.
49pub(crate) async fn connect(url: Url, protocols: &[&str]) -> Result<qmux::Session> {
50	let host = url.host_str().ok_or(Error::MissingHostname)?;
51	let port = url.port().ok_or(Error::MissingPort)?;
52
53	tracing::debug!(%url, "connecting via TCP");
54	qmux::tcp::Config::new(WIRE_VERSION)
55		.protocols(protocols.iter().copied())
56		.connect((host, port))
57		.await
58		.map_err(Error::Connect)
59}
60
61/// Listens for incoming plain-TCP qmux connections on a TCP port.
62pub struct Listener {
63	listener: tokio::net::TcpListener,
64	protocols: Vec<String>,
65}
66
67impl Listener {
68	/// Bind a TCP listener to the given address.
69	pub async fn bind(addr: net::SocketAddr) -> Result<Self> {
70		let listener = tokio::net::TcpListener::bind(addr).await?;
71		Ok(Self {
72			listener,
73			protocols: Vec::new(),
74		})
75	}
76
77	/// Advertise these application protocols (moq ALPNs) for in-band negotiation,
78	/// in preference order. The first server entry the client also offers wins.
79	pub fn with_protocols<I, S>(mut self, protocols: I) -> Self
80	where
81		I: IntoIterator<Item = S>,
82		S: Into<String>,
83	{
84		self.protocols = protocols.into_iter().map(Into::into).collect();
85		self
86	}
87
88	/// The local address the listener is bound to.
89	pub fn local_addr(&self) -> Result<net::SocketAddr> {
90		Ok(self.listener.local_addr()?)
91	}
92
93	/// Accept the next connection, performing the qmux handshake over plain TCP.
94	///
95	/// Returns `None` only if the listener itself is gone; a per-connection
96	/// failure is yielded as `Some(Err(..))` so the accept loop keeps running.
97	pub async fn accept(&self) -> Option<Result<qmux::Session>> {
98		match self.listener.accept().await {
99			Ok((stream, addr)) => {
100				tracing::debug!(%addr, "accepted TCP connection");
101				let session = qmux::tcp::Config::new(WIRE_VERSION)
102					.protocols(self.protocols.iter().map(String::as_str))
103					.accept(stream)
104					.await
105					.map_err(Error::Accept);
106				Some(session)
107			}
108			Err(e) => Some(Err(e.into())),
109		}
110	}
111}
moq_native/tcp.rs

moq_native/
tcp.rs
