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 config.quic.validate()?;
164
165 let tls = config.tls.build()?;
166
167 #[cfg(feature = "noq")]
168 #[allow(unreachable_patterns)]
169 let noq = match backend {
170 QuicBackend::Noq => Some(crate::noq::NoqClient::new(&config)?),
171 _ => None,
172 };
173
174 #[cfg(feature = "quinn")]
175 #[allow(unreachable_patterns)]
176 let quinn = match backend {
177 QuicBackend::Quinn => Some(crate::quinn::QuinnClient::new(&config)?),
178 _ => None,
179 };
180
181 #[cfg(feature = "quiche")]
182 let quiche = match backend {
183 QuicBackend::Quiche => Some(crate::quiche::QuicheClient::new(&config)?),
184 _ => None,
185 };
186
187 let versions = config.versions();
188 Ok(Self {
189 moq: moq_net::Client::new().with_versions(versions.clone()),
190 versions,
191 connect: config.connect,
192 backoff: config.backoff,
193 #[cfg(feature = "websocket")]
194 websocket: config.websocket,
195 tls,
196 #[cfg(feature = "noq")]
197 noq,
198 #[cfg(feature = "quinn")]
199 quinn,
200 #[cfg(feature = "quiche")]
201 quiche,
202 #[cfg(feature = "iroh")]
203 iroh: None,
204 #[cfg(feature = "iroh")]
205 iroh_addrs: Vec::new(),
206 })
207 }
208
209 #[cfg(feature = "iroh")]
214 pub fn with_iroh(mut self, iroh: crate::iroh::Endpoint) -> Self {
215 self.iroh = Some(iroh);
216 self
217 }
218
219 #[cfg(feature = "iroh")]
224 pub fn with_iroh_addrs(mut self, addrs: Vec<std::net::SocketAddr>) -> Self {
225 self.iroh_addrs = addrs;
226 self
227 }
228
229 pub fn with_publisher(mut self, publish: impl moq_net::Consume<moq_net::origin::Consumer>) -> Self {
231 self.moq = self.moq.with_publisher(publish);
232 self
233 }
234
235 pub fn with_subscriber(mut self, subscribe: moq_net::origin::Producer) -> Self {
237 self.moq = self.moq.with_subscriber(subscribe);
238 self
239 }
240
241 pub fn with_stats(mut self, stats: moq_net::stats::Session) -> Self {
244 self.moq = self.moq.with_stats(stats);
245 self
246 }
247
248 pub fn with_cost(mut self, cost: u64) -> Self {
250 self.moq = self.moq.with_cost(cost);
251 self
252 }
253
254 pub fn reconnect(&self, url: Url) -> Reconnect {
259 Reconnect::new(self.clone(), url, self.backoff.clone())
260 }
261
262 pub fn publish(self, origin: moq_net::origin::Consumer) -> Option<Reconnect> {
268 let url = self.connect.clone()?;
269 Some(self.with_publisher(origin).reconnect(url))
270 }
271
272 pub fn consume(self, origin: moq_net::origin::Producer) -> Option<Reconnect> {
284 let url = self.connect.clone()?;
285 let origin = origin.with_linger(self.backoff.linger());
286 Some(self.with_subscriber(origin).reconnect(url))
287 }
288
289 #[cfg(not(any(
293 feature = "noq",
294 feature = "quinn",
295 feature = "quiche",
296 feature = "iroh",
297 feature = "websocket",
298 feature = "tcp",
299 feature = "uds"
300 )))]
301 pub async fn connect(&self, _url: Url) -> crate::Result<moq_net::Session> {
302 Err(Error::NoBackend(
303 "no backend compiled; enable noq, quinn, quiche, iroh, websocket, tcp, or uds feature",
304 ))
305 }
306
307 #[cfg(any(
314 feature = "noq",
315 feature = "quinn",
316 feature = "quiche",
317 feature = "iroh",
318 feature = "websocket",
319 feature = "tcp",
320 feature = "uds"
321 ))]
322 pub async fn connect(&self, url: Url) -> crate::Result<moq_net::Session> {
323 let pair = self.connect_inner(url).await?;
324 tracing::info!(version = %pair.0.version(), "connected");
325 Ok(crate::spawn_session(pair))
326 }
327
328 #[cfg(any(feature = "tcp", feature = "uds"))]
330 fn moq_with_path(&self, path: Option<String>) -> moq_net::Client {
331 match path {
332 Some(path) => self.moq.clone().with_path(path),
333 None => self.moq.clone(),
334 }
335 }
336
337 #[cfg(any(
338 feature = "noq",
339 feature = "quinn",
340 feature = "quiche",
341 feature = "iroh",
342 feature = "websocket",
343 feature = "tcp",
344 feature = "uds"
345 ))]
346 async fn connect_inner(&self, url: Url) -> crate::Result<(moq_net::Session, moq_net::Driver)> {
347 #[cfg(feature = "tcp")]
353 if url.scheme() == "tcp" {
354 let path = setup_path(&url, false);
355 let session = crate::tcp::connect(url, &self.versions.alpns()).await?;
356 return Ok(self.moq_with_path(path).connect(session).await?);
357 }
358
359 #[cfg(all(feature = "uds", unix))]
365 if url.scheme() == "unix" {
366 let path = setup_path(&url, true);
367 let session = crate::unix::connect(url, &self.versions.alpns()).await?;
368 return Ok(self.moq_with_path(path).connect(session).await?);
369 }
370
371 #[cfg(feature = "iroh")]
372 if url.scheme() == "iroh" {
373 let endpoint = self.iroh.as_ref().ok_or(Error::IrohDisabled)?;
374 let session = crate::iroh::connect(endpoint, url, self.iroh_addrs.iter().copied()).await?;
375 let session = self.moq.connect(session).await?;
376 return Ok(session);
377 }
378
379 #[cfg(feature = "noq")]
380 if let Some(noq) = self.noq.as_ref() {
381 let tls = self.tls.clone();
382 let quic_url = url.clone();
383 let quic_handle = async { noq.connect(&tls, quic_url, &self.versions).await.map_err(Error::from) };
384
385 #[cfg(feature = "websocket")]
386 {
387 return self.race_moq_connect(url, quic_handle).await;
388 }
389
390 #[cfg(not(feature = "websocket"))]
391 {
392 let session = quic_handle.await?;
393 return Ok(self.moq.connect(session).await?);
394 }
395 }
396
397 #[cfg(feature = "quinn")]
398 if let Some(quinn) = self.quinn.as_ref() {
399 let tls = self.tls.clone();
400 let quic_url = url.clone();
401 let quic_handle = async { quinn.connect(&tls, quic_url, &self.versions).await.map_err(Error::from) };
402
403 #[cfg(feature = "websocket")]
404 {
405 return self.race_moq_connect(url, quic_handle).await;
406 }
407
408 #[cfg(not(feature = "websocket"))]
409 {
410 let session = quic_handle.await?;
411 return Ok(self.moq.connect(session).await?);
412 }
413 }
414
415 #[cfg(feature = "quiche")]
416 if let Some(quiche) = self.quiche.as_ref() {
417 let quic_url = url.clone();
418 let quic_handle = async { quiche.connect(quic_url, &self.versions).await.map_err(Error::from) };
419
420 #[cfg(feature = "websocket")]
421 {
422 return self.race_moq_connect(url, quic_handle).await;
423 }
424
425 #[cfg(not(feature = "websocket"))]
426 {
427 let session = quic_handle.await?;
428 return Ok(self.moq.connect(session).await?);
429 }
430 }
431
432 #[cfg(feature = "websocket")]
433 {
434 let alpns = self.versions.alpns();
435 let session = crate::websocket::connect(&self.websocket, &self.tls, url, &alpns).await?;
436 return Ok(self.moq.connect(session).await?);
437 }
438
439 #[cfg(not(feature = "websocket"))]
440 return Err(Error::NoBackend("no QUIC backend matched; this should not happen"));
441 }
442
443 #[cfg(feature = "websocket")]
444 async fn race_moq_connect<Q, S>(&self, url: Url, quic: Q) -> crate::Result<(moq_net::Session, moq_net::Driver)>
445 where
446 Q: Future<Output = crate::Result<S>>,
447 S: web_transport_trait::Session,
448 {
449 let alpns = self.versions.alpns();
450 let ws_config = self.websocket.clone();
451 let ws_tls = self.tls.clone();
452 let websocket = async move {
453 crate::websocket::race_handle(&ws_config, &ws_tls, url, &alpns)
454 .await
455 .map(|res| res.map_err(Error::from))
456 };
457
458 match race_transport_connect(quic, websocket).await? {
459 TransportRace::Quic(quic) => Ok(self.moq.connect(quic).await?),
460 TransportRace::WebSocket(websocket) => Ok(self.moq.connect(websocket).await?),
461 }
462 }
463}
464
465#[cfg(any(feature = "tcp", feature = "uds"))]
470fn setup_path(url: &Url, path_is_address: bool) -> Option<String> {
471 let path = if path_is_address {
472 url.query_pairs()
473 .find(|(k, _)| k == "path")
474 .map(|(_, v)| v.into_owned())
475 } else {
476 Some(url.path().to_string())
477 };
478
479 path.filter(|path| !path.is_empty())
482}
483
484#[cfg(feature = "websocket")]
485#[derive(Debug, PartialEq, Eq)]
486enum TransportRace<Q, W> {
487 Quic(Q),
488 WebSocket(W),
489}
490
491#[cfg(feature = "websocket")]
492async fn race_transport_connect<Q, W, QT, WT>(quic: Q, websocket: W) -> crate::Result<TransportRace<QT, WT>>
493where
494 Q: Future<Output = crate::Result<QT>>,
495 W: Future<Output = Option<crate::Result<WT>>>,
496{
497 tokio::pin!(quic);
498 tokio::pin!(websocket);
499
500 let mut quic_err = None;
501 let mut websocket_err = None;
502 let mut quic_done = false;
503 let mut websocket_done = false;
504
505 loop {
506 tokio::select! {
507 res = &mut quic, if !quic_done => {
508 match res {
509 Ok(session) => return Ok(TransportRace::Quic(session)),
510 Err(err) if err.is_auth() => return Err(err),
511 Err(err) => {
512 tracing::warn!(%err, "QUIC connection failed");
513 quic_err = Some(err);
514 quic_done = true;
515 }
516 }
517 }
518 res = &mut websocket, if !websocket_done => {
519 match res {
520 Some(Ok(session)) => return Ok(TransportRace::WebSocket(session)),
521 Some(Err(err)) if err.is_auth() => return Err(err),
522 Some(Err(err)) => {
523 tracing::warn!(%err, "WebSocket connection failed");
524 websocket_err = Some(err);
525 websocket_done = true;
526 }
527 None => {
528 websocket_done = true;
529 }
530 }
531 }
532 else => break,
533 }
534
535 if quic_done && websocket_done {
536 break;
537 }
538 }
539
540 match (quic_err, websocket_err) {
541 (Some(quic), Some(websocket)) => Err(Error::TransportRace {
542 quic: std::sync::Arc::new(quic),
543 websocket: std::sync::Arc::new(websocket),
544 }),
545 (Some(err), None) | (None, Some(err)) => Err(err),
546 (None, None) => Err(Error::ConnectFailed),
547 }
548}
549
550#[cfg(test)]
551mod tests {
552 use super::*;
553 use clap::Parser;
554
555 #[cfg(any(feature = "tcp", feature = "uds"))]
556 #[test]
557 fn setup_path_omits_an_empty_path() {
558 let cases = [
561 ("unix:///run/moq.sock?path=/room", true, Some("/room")),
562 ("unix:///run/moq.sock?path=", true, None),
563 ("unix:///run/moq.sock", true, None),
564 ("tcp://localhost:4443/room", false, Some("/room")),
565 ("tcp://localhost:4443", false, None),
566 ];
567
568 for (url, path_is_address, want) in cases {
569 let url = Url::parse(url).unwrap();
570 let got = setup_path(&url, path_is_address);
571 assert_eq!(got.as_deref(), want, "{url}");
572 }
573 }
574
575 #[test]
576 fn test_toml_disable_verify_survives_update_from() {
577 let toml = r#"
578 tls.disable_verify = true
579 "#;
580
581 let mut config: ClientConfig = toml::from_str(toml).unwrap();
582 assert_eq!(config.tls.disable_verify, Some(true));
583
584 config.update_from(["test"]);
586 assert_eq!(config.tls.disable_verify, Some(true));
587 }
588
589 #[test]
590 fn test_cli_disable_verify_flag() {
591 let config = ClientConfig::parse_from(["test", "--client-tls-disable-verify"]);
592 assert_eq!(config.tls.disable_verify, Some(true));
593 }
594
595 #[test]
596 fn test_cli_disable_verify_explicit_false() {
597 let config = ClientConfig::parse_from(["test", "--client-tls-disable-verify=false"]);
598 assert_eq!(config.tls.disable_verify, Some(false));
599 }
600
601 #[test]
602 fn test_cli_disable_verify_explicit_true() {
603 let config = ClientConfig::parse_from(["test", "--client-tls-disable-verify=true"]);
604 assert_eq!(config.tls.disable_verify, Some(true));
605 }
606
607 #[test]
608 fn test_cli_deprecated_tls_flags_fold_into_canonical() {
609 let config = ClientConfig::parse_from(["test", "--tls-disable-verify=true", "--tls-fingerprint", "abcd1234"]);
613 assert_eq!(
614 config.tls.disable_verify, None,
615 "deprecated flag must not set the canonical field"
616 );
617 assert_eq!(config.tls.effective_disable_verify(), Some(true));
618 assert_eq!(config.tls.effective_fingerprint(), vec!["abcd1234"]);
619 }
620
621 #[test]
622 fn test_canonical_tls_flag_wins_over_deprecated() {
623 let config = ClientConfig::parse_from([
625 "test",
626 "--client-tls-disable-verify=false",
627 "--tls-disable-verify=true",
628 "--client-tls-fingerprint",
629 "aaaa",
630 "--tls-fingerprint",
631 "bbbb",
632 ]);
633 assert_eq!(config.tls.effective_disable_verify(), Some(false));
634 assert_eq!(config.tls.effective_fingerprint(), vec!["aaaa", "bbbb"]);
635 }
636
637 #[test]
638 fn test_cli_no_disable_verify() {
639 let config = ClientConfig::parse_from(["test"]);
640 assert_eq!(config.tls.disable_verify, None);
641 }
642
643 #[test]
644 fn test_toml_fingerprint_survives_update_from() {
645 let toml = r#"
646 tls.fingerprint = ["abcd1234", "ef567890"]
647 "#;
648
649 let mut config: ClientConfig = toml::from_str(toml).unwrap();
650 assert_eq!(config.tls.fingerprint, vec!["abcd1234", "ef567890"]);
651
652 config.update_from(["test"]);
654 assert_eq!(config.tls.fingerprint, vec!["abcd1234", "ef567890"]);
655 }
656
657 #[test]
658 fn test_toml_fingerprint_accepts_single_string() {
659 let toml = r#"
660 tls.fingerprint = "abcd1234"
661 "#;
662
663 let config: ClientConfig = toml::from_str(toml).unwrap();
664 assert_eq!(config.tls.fingerprint, vec!["abcd1234"]);
665 }
666
667 #[test]
668 fn test_cli_fingerprint() {
669 let config = ClientConfig::parse_from(["test", "--client-tls-fingerprint", "abcd1234"]);
670 assert_eq!(config.tls.fingerprint, vec!["abcd1234"]);
671 }
672
673 #[test]
674 fn test_toml_version_survives_update_from() {
675 let toml = r#"
676 version = ["moq-lite-02"]
677 "#;
678
679 let mut config: ClientConfig = toml::from_str(toml).unwrap();
680 assert_eq!(config.version, vec!["moq-lite-02".parse::<moq_net::Version>().unwrap()]);
681
682 config.update_from(["test"]);
684 assert_eq!(config.version, vec!["moq-lite-02".parse::<moq_net::Version>().unwrap()]);
685 }
686
687 #[test]
688 fn test_cli_version() {
689 let config = ClientConfig::parse_from(["test", "--client-version", "moq-lite-03"]);
690 assert_eq!(config.version, vec!["moq-lite-03".parse::<moq_net::Version>().unwrap()]);
691 }
692
693 #[test]
694 fn test_toml_connect_survives_update_from() {
695 let toml = r#"
696 connect = "https://relay.example.com/anon"
697 "#;
698
699 let mut config: ClientConfig = toml::from_str(toml).unwrap();
700 assert_eq!(
701 config.connect.as_ref().unwrap().as_str(),
702 "https://relay.example.com/anon"
703 );
704
705 config.update_from(["test"]);
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_cli_connect() {
715 let config = ClientConfig::parse_from(["test", "--client-connect", "https://relay.example.com/anon"]);
716 assert_eq!(
717 config.connect.as_ref().unwrap().as_str(),
718 "https://relay.example.com/anon"
719 );
720 }
721
722 #[test]
723 fn test_toml_host_name_survives_update_from() {
724 let toml = r#"
725 tls.host_name = "example.host"
726 "#;
727
728 let mut config: ClientConfig = toml::from_str(toml).unwrap();
729 assert_eq!(config.tls.host_name.as_deref(), Some("example.host"));
730
731 config.update_from(["test"]);
733 assert_eq!(config.tls.host_name.as_deref(), Some("example.host"));
734 }
735
736 #[test]
737 fn test_cli_host_name() {
738 let config = ClientConfig::parse_from(["test", "--client-tls-host-name", "override.example"]);
739 assert_eq!(config.tls.host_name.as_deref(), Some("override.example"));
740 }
741
742 #[test]
743 fn test_cli_no_version_defaults_to_all() {
744 let config = ClientConfig::parse_from(["test"]);
745 assert!(config.version.is_empty());
746 assert_eq!(config.versions().alpns().len(), moq_net::ALPNS.len());
748 }
749
750 #[cfg(feature = "websocket")]
751 #[tokio::test]
752 async fn race_transport_connect_stops_on_quic_auth_error() {
753 let quic = async { Err::<usize, _>(crate::ConnectError::Unauthorized.into()) };
754 let websocket = async {
755 tokio::task::yield_now().await;
757 Some(Ok(1usize))
758 };
759
760 let err = super::race_transport_connect(quic, websocket).await.unwrap_err();
761 assert_eq!(err.connect_error(), Some(crate::ConnectError::Unauthorized));
762 }
763
764 #[cfg(feature = "websocket")]
765 #[tokio::test]
766 async fn race_transport_connect_keeps_websocket_after_quic_non_auth_error() {
767 let quic = async { Err::<usize, _>(Error::ConnectFailed) };
768 let websocket = async { Some(Ok(7usize)) };
769
770 let value = super::race_transport_connect(quic, websocket).await.unwrap();
771 assert_eq!(value, super::TransportRace::WebSocket(7));
772 }
773
774 #[cfg(feature = "websocket")]
775 #[tokio::test]
776 async fn race_transport_connect_returns_when_quic_transport_connects() {
777 let quic = async { Ok("quic") };
778 let websocket = std::future::pending::<Option<crate::Result<&str>>>();
779
780 let value = tokio::time::timeout(
781 std::time::Duration::from_secs(1),
782 super::race_transport_connect(quic, websocket),
783 )
784 .await
785 .expect("race waited for WebSocket after QUIC transport connected")
786 .unwrap();
787 assert_eq!(value, super::TransportRace::Quic("quic"));
788 }
789}