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