1use crate::{Backoff, Error, QuicBackend, Reconnect};
2#[cfg(feature = "websocket")]
3use std::future::Future;
4use std::net;
5use url::Url;
6
7#[derive(Clone, Debug, clap::Parser, serde::Serialize, serde::Deserialize)]
9#[serde(deny_unknown_fields, default)]
10#[non_exhaustive]
11pub struct ClientConfig {
12 #[serde(skip_serializing_if = "Option::is_none")]
20 #[arg(id = "client-connect", long = "client-connect", env = "MOQ_CLIENT_CONNECT")]
21 pub connect: Option<Url>,
22
23 #[arg(
25 id = "client-bind",
26 long = "client-bind",
27 default_value = "[::]:0",
28 env = "MOQ_CLIENT_BIND"
29 )]
30 pub bind: net::SocketAddr,
31
32 #[arg(id = "client-backend", long = "client-backend", env = "MOQ_CLIENT_BACKEND")]
35 pub backend: Option<QuicBackend>,
36
37 #[command(flatten)]
39 #[serde(default)]
40 pub quic: crate::quic::Client,
41
42 #[serde(default, skip_serializing_if = "Vec::is_empty")]
50 #[arg(id = "client-version", long = "client-version", env = "MOQ_CLIENT_VERSION")]
51 pub version: Vec<moq_net::Version>,
52
53 #[command(flatten)]
55 #[serde(default)]
56 pub tls: crate::tls::Client,
57
58 #[command(flatten)]
60 #[serde(default)]
61 pub backoff: Backoff,
62
63 #[cfg(feature = "websocket")]
66 #[command(flatten)]
67 #[serde(default)]
68 pub websocket: crate::websocket::Client,
69}
70
71impl ClientConfig {
72 pub fn init(self) -> crate::Result<Client> {
74 Client::new(self)
75 }
76
77 pub fn versions(&self) -> moq_net::Versions {
79 if self.version.is_empty() {
80 moq_net::Versions::all()
81 } else {
82 moq_net::Versions::from(self.version.clone())
83 }
84 }
85}
86
87impl Default for ClientConfig {
88 fn default() -> Self {
89 Self {
90 connect: None,
91 bind: "[::]:0".parse().unwrap(),
92 backend: None,
93 quic: crate::quic::Client::default(),
94 version: Vec::new(),
95 tls: crate::tls::Client::default(),
96 backoff: Backoff::default(),
97 #[cfg(feature = "websocket")]
98 websocket: crate::websocket::Client::default(),
99 }
100 }
101}
102
103#[derive(Clone)]
107pub struct Client {
108 moq: moq_net::Client,
109 versions: moq_net::Versions,
114 connect: Option<Url>,
116 backoff: Backoff,
117 #[cfg(feature = "websocket")]
118 websocket: crate::websocket::Client,
119 tls: rustls::ClientConfig,
120 #[cfg(feature = "noq")]
121 noq: Option<crate::noq::NoqClient>,
122 #[cfg(feature = "quinn")]
123 quinn: Option<crate::quinn::QuinnClient>,
124 #[cfg(feature = "quiche")]
125 quiche: Option<crate::quiche::QuicheClient>,
126 #[cfg(feature = "iroh")]
127 iroh: Option<crate::iroh::Endpoint>,
128 #[cfg(feature = "iroh")]
129 iroh_addrs: Vec<std::net::SocketAddr>,
130}
131
132impl Client {
133 #[cfg(not(any(
137 feature = "noq",
138 feature = "quinn",
139 feature = "quiche",
140 feature = "websocket",
141 feature = "tcp",
142 feature = "uds"
143 )))]
144 pub fn new(_config: ClientConfig) -> crate::Result<Self> {
145 Err(Error::NoBackend(
146 "no QUIC or WebSocket backend compiled; enable noq, quinn, quiche, websocket, tcp, or uds feature",
147 ))
148 }
149
150 #[cfg(any(
152 feature = "noq",
153 feature = "quinn",
154 feature = "quiche",
155 feature = "websocket",
156 feature = "tcp",
157 feature = "uds"
158 ))]
159 pub fn new(config: ClientConfig) -> crate::Result<Self> {
160 #[cfg(any(feature = "noq", feature = "quinn", feature = "quiche"))]
161 let backend = config.backend.clone().unwrap_or_else(crate::default_quic_backend);
162
163 let tls = config.tls.build()?;
164
165 #[cfg(feature = "noq")]
166 #[allow(unreachable_patterns)]
167 let noq = match backend {
168 QuicBackend::Noq => Some(crate::noq::NoqClient::new(&config)?),
169 _ => None,
170 };
171
172 #[cfg(feature = "quinn")]
173 #[allow(unreachable_patterns)]
174 let quinn = match backend {
175 QuicBackend::Quinn => Some(crate::quinn::QuinnClient::new(&config)?),
176 _ => None,
177 };
178
179 #[cfg(feature = "quiche")]
180 let quiche = match backend {
181 QuicBackend::Quiche => Some(crate::quiche::QuicheClient::new(&config)?),
182 _ => None,
183 };
184
185 let versions = config.versions();
186 Ok(Self {
187 moq: moq_net::Client::new().with_versions(versions.clone()),
188 versions,
189 connect: config.connect,
190 backoff: config.backoff,
191 #[cfg(feature = "websocket")]
192 websocket: config.websocket,
193 tls,
194 #[cfg(feature = "noq")]
195 noq,
196 #[cfg(feature = "quinn")]
197 quinn,
198 #[cfg(feature = "quiche")]
199 quiche,
200 #[cfg(feature = "iroh")]
201 iroh: None,
202 #[cfg(feature = "iroh")]
203 iroh_addrs: Vec::new(),
204 })
205 }
206
207 #[cfg(feature = "iroh")]
212 pub fn with_iroh(mut self, iroh: crate::iroh::Endpoint) -> Self {
213 self.iroh = Some(iroh);
214 self
215 }
216
217 #[cfg(feature = "iroh")]
222 pub fn with_iroh_addrs(mut self, addrs: Vec<std::net::SocketAddr>) -> Self {
223 self.iroh_addrs = addrs;
224 self
225 }
226
227 pub fn with_publisher(mut self, publish: impl moq_net::Consume<moq_net::origin::Consumer>) -> Self {
229 self.moq = self.moq.with_publisher(publish);
230 self
231 }
232
233 pub fn with_subscriber(mut self, subscribe: moq_net::origin::Producer) -> Self {
235 self.moq = self.moq.with_subscriber(subscribe);
236 self
237 }
238
239 pub fn with_stats(mut self, stats: moq_net::stats::Session) -> Self {
242 self.moq = self.moq.with_stats(stats);
243 self
244 }
245
246 pub fn with_cost(mut self, cost: u64) -> Self {
248 self.moq = self.moq.with_cost(cost);
249 self
250 }
251
252 pub fn reconnect(&self, url: Url) -> Reconnect {
257 Reconnect::new(self.clone(), url, self.backoff.clone())
258 }
259
260 pub fn publish(self, origin: moq_net::origin::Consumer) -> Option<Reconnect> {
266 let url = self.connect.clone()?;
267 Some(self.with_publisher(origin).reconnect(url))
268 }
269
270 pub fn consume(self, origin: moq_net::origin::Producer) -> Option<Reconnect> {
276 let url = self.connect.clone()?;
277 Some(self.with_subscriber(origin).reconnect(url))
278 }
279
280 #[cfg(not(any(
284 feature = "noq",
285 feature = "quinn",
286 feature = "quiche",
287 feature = "iroh",
288 feature = "websocket",
289 feature = "tcp",
290 feature = "uds"
291 )))]
292 pub async fn connect(&self, _url: Url) -> crate::Result<moq_net::Session> {
293 Err(Error::NoBackend(
294 "no backend compiled; enable noq, quinn, quiche, iroh, websocket, tcp, or uds feature",
295 ))
296 }
297
298 #[cfg(any(
305 feature = "noq",
306 feature = "quinn",
307 feature = "quiche",
308 feature = "iroh",
309 feature = "websocket",
310 feature = "tcp",
311 feature = "uds"
312 ))]
313 pub async fn connect(&self, url: Url) -> crate::Result<moq_net::Session> {
314 let pair = self.connect_inner(url).await?;
315 tracing::info!(version = %pair.0.version(), "connected");
316 Ok(crate::spawn_session(pair))
317 }
318
319 #[cfg(any(feature = "tcp", feature = "uds"))]
321 fn moq_with_path(&self, path: Option<String>) -> moq_net::Client {
322 match path {
323 Some(path) => self.moq.clone().with_path(path),
324 None => self.moq.clone(),
325 }
326 }
327
328 #[cfg(any(
329 feature = "noq",
330 feature = "quinn",
331 feature = "quiche",
332 feature = "iroh",
333 feature = "websocket",
334 feature = "tcp",
335 feature = "uds"
336 ))]
337 async fn connect_inner(&self, url: Url) -> crate::Result<(moq_net::Session, moq_net::Driver)> {
338 #[cfg(feature = "tcp")]
344 if url.scheme() == "tcp" {
345 let path = setup_path(&url, false);
346 let session = crate::tcp::connect(url, &self.versions.alpns()).await?;
347 return Ok(self.moq_with_path(path).connect(session).await?);
348 }
349
350 #[cfg(all(feature = "uds", unix))]
356 if url.scheme() == "unix" {
357 let path = setup_path(&url, true);
358 let session = crate::unix::connect(url, &self.versions.alpns()).await?;
359 return Ok(self.moq_with_path(path).connect(session).await?);
360 }
361
362 #[cfg(feature = "iroh")]
363 if url.scheme() == "iroh" {
364 let endpoint = self.iroh.as_ref().ok_or(Error::IrohDisabled)?;
365 let session = crate::iroh::connect(endpoint, url, self.iroh_addrs.iter().copied()).await?;
366 let session = self.moq.connect(session).await?;
367 return Ok(session);
368 }
369
370 #[cfg(feature = "noq")]
371 if let Some(noq) = self.noq.as_ref() {
372 let tls = self.tls.clone();
373 let quic_url = url.clone();
374 let quic_handle = async { noq.connect(&tls, quic_url, &self.versions).await.map_err(Error::from) };
375
376 #[cfg(feature = "websocket")]
377 {
378 return self.race_moq_connect(url, quic_handle).await;
379 }
380
381 #[cfg(not(feature = "websocket"))]
382 {
383 let session = quic_handle.await?;
384 return Ok(self.moq.connect(session).await?);
385 }
386 }
387
388 #[cfg(feature = "quinn")]
389 if let Some(quinn) = self.quinn.as_ref() {
390 let tls = self.tls.clone();
391 let quic_url = url.clone();
392 let quic_handle = async { quinn.connect(&tls, quic_url, &self.versions).await.map_err(Error::from) };
393
394 #[cfg(feature = "websocket")]
395 {
396 return self.race_moq_connect(url, quic_handle).await;
397 }
398
399 #[cfg(not(feature = "websocket"))]
400 {
401 let session = quic_handle.await?;
402 return Ok(self.moq.connect(session).await?);
403 }
404 }
405
406 #[cfg(feature = "quiche")]
407 if let Some(quiche) = self.quiche.as_ref() {
408 let quic_url = url.clone();
409 let quic_handle = async { quiche.connect(quic_url, &self.versions).await.map_err(Error::from) };
410
411 #[cfg(feature = "websocket")]
412 {
413 return self.race_moq_connect(url, quic_handle).await;
414 }
415
416 #[cfg(not(feature = "websocket"))]
417 {
418 let session = quic_handle.await?;
419 return Ok(self.moq.connect(session).await?);
420 }
421 }
422
423 #[cfg(feature = "websocket")]
424 {
425 let alpns = self.versions.alpns();
426 let session = crate::websocket::connect(&self.websocket, &self.tls, url, &alpns).await?;
427 return Ok(self.moq.connect(session).await?);
428 }
429
430 #[cfg(not(feature = "websocket"))]
431 return Err(Error::NoBackend("no QUIC backend matched; this should not happen"));
432 }
433
434 #[cfg(feature = "websocket")]
435 async fn race_moq_connect<Q, S>(&self, url: Url, quic: Q) -> crate::Result<(moq_net::Session, moq_net::Driver)>
436 where
437 Q: Future<Output = crate::Result<S>>,
438 S: web_transport_trait::Session,
439 {
440 let alpns = self.versions.alpns();
441 let ws_config = self.websocket.clone();
442 let ws_tls = self.tls.clone();
443 let websocket = async move {
444 crate::websocket::race_handle(&ws_config, &ws_tls, url, &alpns)
445 .await
446 .map(|res| res.map_err(Error::from))
447 };
448
449 match race_transport_connect(quic, websocket).await? {
450 TransportRace::Quic(quic) => Ok(self.moq.connect(quic).await?),
451 TransportRace::WebSocket(websocket) => Ok(self.moq.connect(websocket).await?),
452 }
453 }
454}
455
456#[cfg(any(feature = "tcp", feature = "uds"))]
461fn setup_path(url: &Url, path_is_address: bool) -> Option<String> {
462 let path = if path_is_address {
463 url.query_pairs()
464 .find(|(k, _)| k == "path")
465 .map(|(_, v)| v.into_owned())
466 } else {
467 Some(url.path().to_string())
468 };
469
470 path.filter(|path| !path.is_empty())
473}
474
475#[cfg(feature = "websocket")]
476#[derive(Debug, PartialEq, Eq)]
477enum TransportRace<Q, W> {
478 Quic(Q),
479 WebSocket(W),
480}
481
482#[cfg(feature = "websocket")]
483async fn race_transport_connect<Q, W, QT, WT>(quic: Q, websocket: W) -> crate::Result<TransportRace<QT, WT>>
484where
485 Q: Future<Output = crate::Result<QT>>,
486 W: Future<Output = Option<crate::Result<WT>>>,
487{
488 tokio::pin!(quic);
489 tokio::pin!(websocket);
490
491 let mut quic_err = None;
492 let mut websocket_err = None;
493 let mut quic_done = false;
494 let mut websocket_done = false;
495
496 loop {
497 tokio::select! {
498 res = &mut quic, if !quic_done => {
499 match res {
500 Ok(session) => return Ok(TransportRace::Quic(session)),
501 Err(err) if err.is_auth() => return Err(err),
502 Err(err) => {
503 tracing::warn!(%err, "QUIC connection failed");
504 quic_err = Some(err);
505 quic_done = true;
506 }
507 }
508 }
509 res = &mut websocket, if !websocket_done => {
510 match res {
511 Some(Ok(session)) => return Ok(TransportRace::WebSocket(session)),
512 Some(Err(err)) if err.is_auth() => return Err(err),
513 Some(Err(err)) => {
514 tracing::warn!(%err, "WebSocket connection failed");
515 websocket_err = Some(err);
516 websocket_done = true;
517 }
518 None => {
519 websocket_done = true;
520 }
521 }
522 }
523 else => break,
524 }
525
526 if quic_done && websocket_done {
527 break;
528 }
529 }
530
531 match (quic_err, websocket_err) {
532 (Some(quic), Some(websocket)) => Err(Error::TransportRace {
533 quic: std::sync::Arc::new(quic),
534 websocket: std::sync::Arc::new(websocket),
535 }),
536 (Some(err), None) | (None, Some(err)) => Err(err),
537 (None, None) => Err(Error::ConnectFailed),
538 }
539}
540
541#[cfg(test)]
542mod tests {
543 use super::*;
544 use clap::Parser;
545
546 #[cfg(any(feature = "tcp", feature = "uds"))]
547 #[test]
548 fn setup_path_omits_an_empty_path() {
549 let cases = [
552 ("unix:///run/moq.sock?path=/room", true, Some("/room")),
553 ("unix:///run/moq.sock?path=", true, None),
554 ("unix:///run/moq.sock", true, None),
555 ("tcp://localhost:4443/room", false, Some("/room")),
556 ("tcp://localhost:4443", false, None),
557 ];
558
559 for (url, path_is_address, want) in cases {
560 let url = Url::parse(url).unwrap();
561 let got = setup_path(&url, path_is_address);
562 assert_eq!(got.as_deref(), want, "{url}");
563 }
564 }
565
566 #[test]
567 fn test_toml_disable_verify_survives_update_from() {
568 let toml = r#"
569 tls.disable_verify = true
570 "#;
571
572 let mut config: ClientConfig = toml::from_str(toml).unwrap();
573 assert_eq!(config.tls.disable_verify, Some(true));
574
575 config.update_from(["test"]);
577 assert_eq!(config.tls.disable_verify, Some(true));
578 }
579
580 #[test]
581 fn test_cli_disable_verify_flag() {
582 let config = ClientConfig::parse_from(["test", "--client-tls-disable-verify"]);
583 assert_eq!(config.tls.disable_verify, Some(true));
584 }
585
586 #[test]
587 fn test_cli_disable_verify_explicit_false() {
588 let config = ClientConfig::parse_from(["test", "--client-tls-disable-verify=false"]);
589 assert_eq!(config.tls.disable_verify, Some(false));
590 }
591
592 #[test]
593 fn test_cli_disable_verify_explicit_true() {
594 let config = ClientConfig::parse_from(["test", "--client-tls-disable-verify=true"]);
595 assert_eq!(config.tls.disable_verify, Some(true));
596 }
597
598 #[test]
599 fn test_cli_deprecated_tls_flags_fold_into_canonical() {
600 let config = ClientConfig::parse_from(["test", "--tls-disable-verify=true", "--tls-fingerprint", "abcd1234"]);
604 assert_eq!(
605 config.tls.disable_verify, None,
606 "deprecated flag must not set the canonical field"
607 );
608 assert_eq!(config.tls.effective_disable_verify(), Some(true));
609 assert_eq!(config.tls.effective_fingerprint(), vec!["abcd1234"]);
610 }
611
612 #[test]
613 fn test_canonical_tls_flag_wins_over_deprecated() {
614 let config = ClientConfig::parse_from([
616 "test",
617 "--client-tls-disable-verify=false",
618 "--tls-disable-verify=true",
619 "--client-tls-fingerprint",
620 "aaaa",
621 "--tls-fingerprint",
622 "bbbb",
623 ]);
624 assert_eq!(config.tls.effective_disable_verify(), Some(false));
625 assert_eq!(config.tls.effective_fingerprint(), vec!["aaaa", "bbbb"]);
626 }
627
628 #[test]
629 fn test_cli_no_disable_verify() {
630 let config = ClientConfig::parse_from(["test"]);
631 assert_eq!(config.tls.disable_verify, None);
632 }
633
634 #[test]
635 fn test_toml_fingerprint_survives_update_from() {
636 let toml = r#"
637 tls.fingerprint = ["abcd1234", "ef567890"]
638 "#;
639
640 let mut config: ClientConfig = toml::from_str(toml).unwrap();
641 assert_eq!(config.tls.fingerprint, vec!["abcd1234", "ef567890"]);
642
643 config.update_from(["test"]);
645 assert_eq!(config.tls.fingerprint, vec!["abcd1234", "ef567890"]);
646 }
647
648 #[test]
649 fn test_toml_fingerprint_accepts_single_string() {
650 let toml = r#"
651 tls.fingerprint = "abcd1234"
652 "#;
653
654 let config: ClientConfig = toml::from_str(toml).unwrap();
655 assert_eq!(config.tls.fingerprint, vec!["abcd1234"]);
656 }
657
658 #[test]
659 fn test_cli_fingerprint() {
660 let config = ClientConfig::parse_from(["test", "--client-tls-fingerprint", "abcd1234"]);
661 assert_eq!(config.tls.fingerprint, vec!["abcd1234"]);
662 }
663
664 #[test]
665 fn test_toml_version_survives_update_from() {
666 let toml = r#"
667 version = ["moq-lite-02"]
668 "#;
669
670 let mut config: ClientConfig = toml::from_str(toml).unwrap();
671 assert_eq!(config.version, vec!["moq-lite-02".parse::<moq_net::Version>().unwrap()]);
672
673 config.update_from(["test"]);
675 assert_eq!(config.version, vec!["moq-lite-02".parse::<moq_net::Version>().unwrap()]);
676 }
677
678 #[test]
679 fn test_cli_version() {
680 let config = ClientConfig::parse_from(["test", "--client-version", "moq-lite-03"]);
681 assert_eq!(config.version, vec!["moq-lite-03".parse::<moq_net::Version>().unwrap()]);
682 }
683
684 #[test]
685 fn test_toml_connect_survives_update_from() {
686 let toml = r#"
687 connect = "https://relay.example.com/anon"
688 "#;
689
690 let mut config: ClientConfig = toml::from_str(toml).unwrap();
691 assert_eq!(
692 config.connect.as_ref().unwrap().as_str(),
693 "https://relay.example.com/anon"
694 );
695
696 config.update_from(["test"]);
698 assert_eq!(
699 config.connect.as_ref().unwrap().as_str(),
700 "https://relay.example.com/anon"
701 );
702 }
703
704 #[test]
705 fn test_cli_connect() {
706 let config = ClientConfig::parse_from(["test", "--client-connect", "https://relay.example.com/anon"]);
707 assert_eq!(
708 config.connect.as_ref().unwrap().as_str(),
709 "https://relay.example.com/anon"
710 );
711 }
712
713 #[test]
714 fn test_toml_host_name_survives_update_from() {
715 let toml = r#"
716 tls.host_name = "example.host"
717 "#;
718
719 let mut config: ClientConfig = toml::from_str(toml).unwrap();
720 assert_eq!(config.tls.host_name.as_deref(), Some("example.host"));
721
722 config.update_from(["test"]);
724 assert_eq!(config.tls.host_name.as_deref(), Some("example.host"));
725 }
726
727 #[test]
728 fn test_cli_host_name() {
729 let config = ClientConfig::parse_from(["test", "--client-tls-host-name", "override.example"]);
730 assert_eq!(config.tls.host_name.as_deref(), Some("override.example"));
731 }
732
733 #[test]
734 fn test_cli_no_version_defaults_to_all() {
735 let config = ClientConfig::parse_from(["test"]);
736 assert!(config.version.is_empty());
737 assert_eq!(config.versions().alpns().len(), moq_net::ALPNS.len());
739 }
740
741 #[cfg(feature = "websocket")]
742 #[tokio::test]
743 async fn race_transport_connect_stops_on_quic_auth_error() {
744 let quic = async { Err::<usize, _>(crate::ConnectError::Unauthorized.into()) };
745 let websocket = async {
746 tokio::task::yield_now().await;
748 Some(Ok(1usize))
749 };
750
751 let err = super::race_transport_connect(quic, websocket).await.unwrap_err();
752 assert_eq!(err.connect_error(), Some(crate::ConnectError::Unauthorized));
753 }
754
755 #[cfg(feature = "websocket")]
756 #[tokio::test]
757 async fn race_transport_connect_keeps_websocket_after_quic_non_auth_error() {
758 let quic = async { Err::<usize, _>(Error::ConnectFailed) };
759 let websocket = async { Some(Ok(7usize)) };
760
761 let value = super::race_transport_connect(quic, websocket).await.unwrap();
762 assert_eq!(value, super::TransportRace::WebSocket(7));
763 }
764
765 #[cfg(feature = "websocket")]
766 #[tokio::test]
767 async fn race_transport_connect_returns_when_quic_transport_connects() {
768 let quic = async { Ok("quic") };
769 let websocket = std::future::pending::<Option<crate::Result<&str>>>();
770
771 let value = tokio::time::timeout(
772 std::time::Duration::from_secs(1),
773 super::race_transport_connect(quic, websocket),
774 )
775 .await
776 .expect("race waited for WebSocket after QUIC transport connected")
777 .unwrap();
778 assert_eq!(value, super::TransportRace::Quic("quic"));
779 }
780}