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 = Box::pin(self.connect_inner(url)).await?;
326 tracing::info!(version = %pair.0.version(), "connected");
327 Ok(crate::spawn_session(pair))
328 }
329
330 #[cfg(any(feature = "tcp", feature = "uds"))]
332 fn moq_with_path(&self, path: Option<String>) -> moq_net::Client {
333 match path {
334 Some(path) => self.moq.clone().with_path(path),
335 None => self.moq.clone(),
336 }
337 }
338
339 #[cfg(any(
340 feature = "noq",
341 feature = "quinn",
342 feature = "quiche",
343 feature = "iroh",
344 feature = "websocket",
345 feature = "tcp",
346 feature = "uds"
347 ))]
348 async fn connect_inner(&self, url: Url) -> crate::Result<(moq_net::Session, moq_net::Driver)> {
349 #[cfg(feature = "tcp")]
355 if url.scheme() == "tcp" {
356 let path = setup_path(&url, false);
357 let session = crate::tcp::connect(url, &self.versions.alpns()).await?;
358 return Ok(self.moq_with_path(path).connect(session).await?);
359 }
360
361 #[cfg(all(feature = "uds", unix))]
367 if url.scheme() == "unix" {
368 let path = setup_path(&url, true);
369 let session = crate::unix::connect(url, &self.versions.alpns()).await?;
370 return Ok(self.moq_with_path(path).connect(session).await?);
371 }
372
373 #[cfg(feature = "iroh")]
374 if url.scheme() == "iroh" {
375 let endpoint = self.iroh.as_ref().ok_or(Error::IrohDisabled)?;
376 let session = crate::iroh::connect(endpoint, url, self.iroh_addrs.iter().copied()).await?;
377 let session = self.moq.connect(session).await?;
378 return Ok(session);
379 }
380
381 #[cfg(feature = "noq")]
382 if let Some(noq) = self.noq.as_ref() {
383 let tls = self.tls.clone();
384 let quic_url = url.clone();
385 let quic_handle = async { noq.connect(&tls, quic_url, &self.versions).await.map_err(Error::from) };
386
387 #[cfg(feature = "websocket")]
388 {
389 return self.race_moq_connect(url, quic_handle).await;
390 }
391
392 #[cfg(not(feature = "websocket"))]
393 {
394 let session = quic_handle.await?;
395 return Ok(self.moq.connect(session).await?);
396 }
397 }
398
399 #[cfg(feature = "quinn")]
400 if let Some(quinn) = self.quinn.as_ref() {
401 let tls = self.tls.clone();
402 let quic_url = url.clone();
403 let quic_handle = async { quinn.connect(&tls, quic_url, &self.versions).await.map_err(Error::from) };
404
405 #[cfg(feature = "websocket")]
406 {
407 return self.race_moq_connect(url, quic_handle).await;
408 }
409
410 #[cfg(not(feature = "websocket"))]
411 {
412 let session = quic_handle.await?;
413 return Ok(self.moq.connect(session).await?);
414 }
415 }
416
417 #[cfg(feature = "quiche")]
418 if let Some(quiche) = self.quiche.as_ref() {
419 let quic_url = url.clone();
420 let quic_handle = async { quiche.connect(quic_url, &self.versions).await.map_err(Error::from) };
421
422 #[cfg(feature = "websocket")]
423 {
424 return self.race_moq_connect(url, quic_handle).await;
425 }
426
427 #[cfg(not(feature = "websocket"))]
428 {
429 let session = quic_handle.await?;
430 return Ok(self.moq.connect(session).await?);
431 }
432 }
433
434 #[cfg(feature = "websocket")]
435 {
436 let alpns = self.versions.alpns();
437 let session = crate::websocket::connect(&self.websocket, &self.tls, url, &alpns).await?;
438 return Ok(self.moq.connect(session).await?);
439 }
440
441 #[cfg(not(feature = "websocket"))]
442 return Err(Error::NoBackend("no QUIC backend matched; this should not happen"));
443 }
444
445 #[cfg(feature = "websocket")]
446 async fn race_moq_connect<Q, S>(&self, url: Url, quic: Q) -> crate::Result<(moq_net::Session, moq_net::Driver)>
447 where
448 Q: Future<Output = crate::Result<S>>,
449 S: web_transport_trait::Session,
450 {
451 let alpns = self.versions.alpns();
452 let ws_config = self.websocket.clone();
453 let ws_tls = self.tls.clone();
454 let websocket = async move {
455 crate::websocket::race_handle(&ws_config, &ws_tls, url, &alpns)
456 .await
457 .map(|res| res.map_err(Error::from))
458 };
459
460 match race_transport_connect(quic, websocket).await? {
461 TransportRace::Quic(quic) => Ok(self.moq.connect(quic).await?),
462 TransportRace::WebSocket(websocket) => Ok(self.moq.connect(websocket).await?),
463 }
464 }
465}
466
467#[cfg(any(feature = "tcp", feature = "uds"))]
472fn setup_path(url: &Url, path_is_address: bool) -> Option<String> {
473 let path = if path_is_address {
474 url.query_pairs()
475 .find(|(k, _)| k == "path")
476 .map(|(_, v)| v.into_owned())
477 } else {
478 Some(url.path().to_string())
479 };
480
481 path.filter(|path| !path.is_empty())
484}
485
486#[cfg(feature = "websocket")]
487#[derive(Debug, PartialEq, Eq)]
488enum TransportRace<Q, W> {
489 Quic(Q),
490 WebSocket(W),
491}
492
493#[cfg(feature = "websocket")]
494async fn race_transport_connect<Q, W, QT, WT>(quic: Q, websocket: W) -> crate::Result<TransportRace<QT, WT>>
495where
496 Q: Future<Output = crate::Result<QT>>,
497 W: Future<Output = Option<crate::Result<WT>>>,
498{
499 tokio::pin!(quic);
500 tokio::pin!(websocket);
501
502 let mut quic_err = None;
503 let mut websocket_err = None;
504 let mut quic_done = false;
505 let mut websocket_done = false;
506
507 loop {
508 tokio::select! {
509 res = &mut quic, if !quic_done => {
510 match res {
511 Ok(session) => return Ok(TransportRace::Quic(session)),
512 Err(err) if err.is_auth() => return Err(err),
513 Err(err) => {
514 tracing::warn!(%err, "QUIC connection failed");
515 quic_err = Some(err);
516 quic_done = true;
517 }
518 }
519 }
520 res = &mut websocket, if !websocket_done => {
521 match res {
522 Some(Ok(session)) => return Ok(TransportRace::WebSocket(session)),
523 Some(Err(err)) if err.is_auth() => return Err(err),
524 Some(Err(err)) => {
525 tracing::warn!(%err, "WebSocket connection failed");
526 websocket_err = Some(err);
527 websocket_done = true;
528 }
529 None => {
530 websocket_done = true;
531 }
532 }
533 }
534 else => break,
535 }
536
537 if quic_done && websocket_done {
538 break;
539 }
540 }
541
542 match (quic_err, websocket_err) {
543 (Some(quic), Some(websocket)) => Err(Error::TransportRace {
544 quic: std::sync::Arc::new(quic),
545 websocket: std::sync::Arc::new(websocket),
546 }),
547 (Some(err), None) | (None, Some(err)) => Err(err),
548 (None, None) => Err(Error::ConnectFailed),
549 }
550}
551
552#[cfg(test)]
553mod tests {
554 use super::*;
555 use clap::Parser;
556
557 #[cfg(any(feature = "tcp", feature = "uds"))]
558 #[test]
559 fn setup_path_omits_an_empty_path() {
560 let cases = [
563 ("unix:///run/moq.sock?path=/room", true, Some("/room")),
564 ("unix:///run/moq.sock?path=", true, None),
565 ("unix:///run/moq.sock", true, None),
566 ("tcp://localhost:4443/room", false, Some("/room")),
567 ("tcp://localhost:4443", false, None),
568 ];
569
570 for (url, path_is_address, want) in cases {
571 let url = Url::parse(url).unwrap();
572 let got = setup_path(&url, path_is_address);
573 assert_eq!(got.as_deref(), want, "{url}");
574 }
575 }
576
577 #[test]
578 fn test_toml_disable_verify_survives_update_from() {
579 let toml = r#"
580 tls.disable_verify = true
581 "#;
582
583 let mut config: ClientConfig = toml::from_str(toml).unwrap();
584 assert_eq!(config.tls.disable_verify, Some(true));
585
586 config.update_from(["test"]);
588 assert_eq!(config.tls.disable_verify, Some(true));
589 }
590
591 #[test]
592 fn test_cli_disable_verify_flag() {
593 let config = ClientConfig::parse_from(["test", "--client-tls-disable-verify"]);
594 assert_eq!(config.tls.disable_verify, Some(true));
595 }
596
597 #[test]
598 fn test_cli_disable_verify_explicit_false() {
599 let config = ClientConfig::parse_from(["test", "--client-tls-disable-verify=false"]);
600 assert_eq!(config.tls.disable_verify, Some(false));
601 }
602
603 #[test]
604 fn test_cli_disable_verify_explicit_true() {
605 let config = ClientConfig::parse_from(["test", "--client-tls-disable-verify=true"]);
606 assert_eq!(config.tls.disable_verify, Some(true));
607 }
608
609 #[test]
610 fn test_cli_deprecated_tls_flags_fold_into_canonical() {
611 let config = ClientConfig::parse_from(["test", "--tls-disable-verify=true", "--tls-fingerprint", "abcd1234"]);
615 assert_eq!(
616 config.tls.disable_verify, None,
617 "deprecated flag must not set the canonical field"
618 );
619 assert_eq!(config.tls.effective_disable_verify(), Some(true));
620 assert_eq!(config.tls.effective_fingerprint(), vec!["abcd1234"]);
621 }
622
623 #[test]
624 fn test_canonical_tls_flag_wins_over_deprecated() {
625 let config = ClientConfig::parse_from([
627 "test",
628 "--client-tls-disable-verify=false",
629 "--tls-disable-verify=true",
630 "--client-tls-fingerprint",
631 "aaaa",
632 "--tls-fingerprint",
633 "bbbb",
634 ]);
635 assert_eq!(config.tls.effective_disable_verify(), Some(false));
636 assert_eq!(config.tls.effective_fingerprint(), vec!["aaaa", "bbbb"]);
637 }
638
639 #[test]
640 fn test_cli_no_disable_verify() {
641 let config = ClientConfig::parse_from(["test"]);
642 assert_eq!(config.tls.disable_verify, None);
643 }
644
645 #[test]
646 fn test_toml_fingerprint_survives_update_from() {
647 let toml = r#"
648 tls.fingerprint = ["abcd1234", "ef567890"]
649 "#;
650
651 let mut config: ClientConfig = toml::from_str(toml).unwrap();
652 assert_eq!(config.tls.fingerprint, vec!["abcd1234", "ef567890"]);
653
654 config.update_from(["test"]);
656 assert_eq!(config.tls.fingerprint, vec!["abcd1234", "ef567890"]);
657 }
658
659 #[test]
660 fn test_toml_fingerprint_accepts_single_string() {
661 let toml = r#"
662 tls.fingerprint = "abcd1234"
663 "#;
664
665 let config: ClientConfig = toml::from_str(toml).unwrap();
666 assert_eq!(config.tls.fingerprint, vec!["abcd1234"]);
667 }
668
669 #[test]
670 fn test_cli_fingerprint() {
671 let config = ClientConfig::parse_from(["test", "--client-tls-fingerprint", "abcd1234"]);
672 assert_eq!(config.tls.fingerprint, vec!["abcd1234"]);
673 }
674
675 #[test]
676 fn test_toml_version_survives_update_from() {
677 let toml = r#"
678 version = ["moq-lite-02"]
679 "#;
680
681 let mut config: ClientConfig = toml::from_str(toml).unwrap();
682 assert_eq!(config.version, vec!["moq-lite-02".parse::<moq_net::Version>().unwrap()]);
683
684 config.update_from(["test"]);
686 assert_eq!(config.version, vec!["moq-lite-02".parse::<moq_net::Version>().unwrap()]);
687 }
688
689 #[test]
690 fn test_cli_version() {
691 let config = ClientConfig::parse_from(["test", "--client-version", "moq-lite-03"]);
692 assert_eq!(config.version, vec!["moq-lite-03".parse::<moq_net::Version>().unwrap()]);
693 }
694
695 #[test]
696 fn test_toml_connect_survives_update_from() {
697 let toml = r#"
698 connect = "https://relay.example.com/anon"
699 "#;
700
701 let mut config: ClientConfig = toml::from_str(toml).unwrap();
702 assert_eq!(
703 config.connect.as_ref().unwrap().as_str(),
704 "https://relay.example.com/anon"
705 );
706
707 config.update_from(["test"]);
709 assert_eq!(
710 config.connect.as_ref().unwrap().as_str(),
711 "https://relay.example.com/anon"
712 );
713 }
714
715 #[test]
716 fn test_cli_connect() {
717 let config = ClientConfig::parse_from(["test", "--client-connect", "https://relay.example.com/anon"]);
718 assert_eq!(
719 config.connect.as_ref().unwrap().as_str(),
720 "https://relay.example.com/anon"
721 );
722 }
723
724 #[test]
725 fn test_toml_host_name_survives_update_from() {
726 let toml = r#"
727 tls.host_name = "example.host"
728 "#;
729
730 let mut config: ClientConfig = toml::from_str(toml).unwrap();
731 assert_eq!(config.tls.host_name.as_deref(), Some("example.host"));
732
733 config.update_from(["test"]);
735 assert_eq!(config.tls.host_name.as_deref(), Some("example.host"));
736 }
737
738 #[test]
739 fn test_cli_host_name() {
740 let config = ClientConfig::parse_from(["test", "--client-tls-host-name", "override.example"]);
741 assert_eq!(config.tls.host_name.as_deref(), Some("override.example"));
742 }
743
744 #[test]
745 fn test_cli_no_version_defaults_to_all() {
746 let config = ClientConfig::parse_from(["test"]);
747 assert!(config.version.is_empty());
748 assert_eq!(config.versions().alpns().len(), moq_net::ALPNS.len());
750 }
751
752 #[cfg(feature = "websocket")]
753 #[tokio::test]
754 async fn race_transport_connect_stops_on_quic_auth_error() {
755 let quic = async { Err::<usize, _>(crate::ConnectError::Unauthorized.into()) };
756 let websocket = async {
757 tokio::task::yield_now().await;
759 Some(Ok(1usize))
760 };
761
762 let err = super::race_transport_connect(quic, websocket).await.unwrap_err();
763 assert_eq!(err.connect_error(), Some(crate::ConnectError::Unauthorized));
764 }
765
766 #[cfg(feature = "websocket")]
767 #[tokio::test]
768 async fn race_transport_connect_keeps_websocket_after_quic_non_auth_error() {
769 let quic = async { Err::<usize, _>(Error::ConnectFailed) };
770 let websocket = async { Some(Ok(7usize)) };
771
772 let value = super::race_transport_connect(quic, websocket).await.unwrap();
773 assert_eq!(value, super::TransportRace::WebSocket(7));
774 }
775
776 #[cfg(feature = "websocket")]
777 #[tokio::test]
778 async fn race_transport_connect_returns_when_quic_transport_connects() {
779 let quic = async { Ok("quic") };
780 let websocket = std::future::pending::<Option<crate::Result<&str>>>();
781
782 let value = tokio::time::timeout(
783 std::time::Duration::from_secs(1),
784 super::race_transport_connect(quic, websocket),
785 )
786 .await
787 .expect("race waited for WebSocket after QUIC transport connected")
788 .unwrap();
789 assert_eq!(value, super::TransportRace::Quic("quic"));
790 }
791}