1use crate::{Backoff, Error, QuicBackend, Reconnect};
2#[cfg(all(feature = "websocket", any(feature = "noq", feature = "quinn", feature = "quiche")))]
3use std::future::Future;
4use std::net;
5use url::Url;
6
7const DEFAULT_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
8
9pub(crate) const DEFAULT_FAILOVER_DELAY: std::time::Duration = std::time::Duration::from_millis(250);
17
18pub(crate) const DEFAULT_RESOLUTION_DELAY: std::time::Duration = std::time::Duration::from_millis(50);
24
25#[derive(Clone, Debug, clap::Parser, serde::Serialize, serde::Deserialize)]
27#[serde(deny_unknown_fields, default)]
28#[non_exhaustive]
29pub struct ClientConfig {
30 #[serde(skip_serializing_if = "Option::is_none")]
38 #[arg(id = "client-connect", long = "client-connect", env = "MOQ_CLIENT_CONNECT")]
39 pub connect: Option<Url>,
40
41 #[arg(
43 id = "client-bind",
44 long = "client-bind",
45 default_value = "[::]:0",
46 env = "MOQ_CLIENT_BIND"
47 )]
48 pub bind: net::SocketAddr,
49
50 #[arg(id = "client-backend", long = "client-backend", env = "MOQ_CLIENT_BACKEND")]
53 pub backend: Option<QuicBackend>,
54
55 #[serde(default, skip_serializing_if = "Option::is_none", with = "humantime_serde::option")]
65 #[arg(
66 id = "client-failover-delay",
67 long = "client-failover-delay",
68 env = "MOQ_CLIENT_FAILOVER_DELAY",
69 value_parser = humantime::parse_duration,
70 )]
71 pub failover_delay: Option<std::time::Duration>,
72
73 #[serde(default, skip_serializing_if = "Option::is_none", with = "humantime_serde::option")]
81 #[arg(
82 id = "client-resolution-delay",
83 long = "client-resolution-delay",
84 env = "MOQ_CLIENT_RESOLUTION_DELAY",
85 value_parser = humantime::parse_duration,
86 )]
87 pub resolution_delay: Option<std::time::Duration>,
88
89 #[arg(
99 id = "client-connect-timeout",
100 long = "client-connect-timeout",
101 env = "MOQ_CLIENT_CONNECT_TIMEOUT",
102 value_parser = humantime::parse_duration,
103 )]
104 #[serde(default, skip_serializing_if = "Option::is_none", with = "humantime_serde::option")]
105 pub timeout: Option<std::time::Duration>,
106
107 #[command(flatten)]
109 #[serde(default)]
110 pub quic: crate::quic::Client,
111
112 #[serde(default, skip_serializing_if = "Vec::is_empty")]
118 #[arg(
119 id = "client-version",
120 long = "client-version",
121 env = "MOQ_CLIENT_VERSION",
122 value_parser = crate::version_parser(),
123 )]
124 pub version: Vec<moq_net::Version>,
125
126 #[command(flatten)]
128 #[serde(default)]
129 pub tls: crate::tls::Client,
130
131 #[command(flatten)]
133 #[serde(default)]
134 pub backoff: Backoff,
135
136 #[cfg(feature = "websocket")]
139 #[command(flatten)]
140 #[serde(default)]
141 pub websocket: crate::websocket::Client,
142}
143
144impl ClientConfig {
145 pub fn init(self) -> crate::Result<Client> {
147 Client::new(self)
148 }
149
150 pub fn versions(&self) -> moq_net::Versions {
152 if self.version.is_empty() {
153 moq_net::Versions::all()
154 } else {
155 moq_net::Versions::from(self.version.clone())
156 }
157 }
158
159 pub fn resolved_failover_delay(&self) -> std::time::Duration {
166 self.failover_delay.unwrap_or(DEFAULT_FAILOVER_DELAY)
167 }
168
169 pub fn resolved_resolution_delay(&self) -> std::time::Duration {
173 self.resolution_delay.unwrap_or(DEFAULT_RESOLUTION_DELAY)
174 }
175
176 pub fn resolved_connect_timeout(&self) -> std::time::Duration {
179 self.timeout.unwrap_or(DEFAULT_CONNECT_TIMEOUT)
180 }
181}
182
183impl Default for ClientConfig {
184 fn default() -> Self {
185 Self {
186 connect: None,
187 bind: "[::]:0".parse().unwrap(),
188 backend: None,
189 failover_delay: None,
190 resolution_delay: None,
191 timeout: None,
192 quic: crate::quic::Client::default(),
193 version: Vec::new(),
194 tls: crate::tls::Client::default(),
195 backoff: Backoff::default(),
196 #[cfg(feature = "websocket")]
197 websocket: crate::websocket::Client::default(),
198 }
199 }
200}
201
202#[derive(Clone)]
206pub struct Client {
207 moq: moq_net::Client,
208 #[cfg(any(
214 feature = "noq",
215 feature = "quinn",
216 feature = "quiche",
217 feature = "websocket",
218 feature = "tcp",
219 feature = "uds"
220 ))]
221 versions: moq_net::Versions,
222 connect: Option<Url>,
224 timeout: std::time::Duration,
226 backoff: Backoff,
227 #[cfg(feature = "tcp")]
230 failover_delay: std::time::Duration,
231 #[cfg(feature = "tcp")]
232 resolution_delay: std::time::Duration,
233 #[cfg(feature = "websocket")]
234 websocket: crate::websocket::Client,
235 #[cfg(feature = "websocket")]
238 tls_host_name: Option<String>,
239 #[cfg(any(feature = "noq", feature = "quinn", feature = "websocket"))]
242 tls: rustls::ClientConfig,
243 #[cfg(feature = "noq")]
244 noq: Option<crate::noq::NoqClient>,
245 #[cfg(feature = "quinn")]
246 quinn: Option<crate::quinn::QuinnClient>,
247 #[cfg(feature = "quiche")]
248 quiche: Option<crate::quiche::QuicheClient>,
249 #[cfg(feature = "iroh")]
250 iroh: Option<crate::iroh::Endpoint>,
251 #[cfg(feature = "iroh")]
252 iroh_addrs: Vec<std::net::SocketAddr>,
253}
254
255impl Client {
256 #[cfg(not(any(
260 feature = "noq",
261 feature = "quinn",
262 feature = "quiche",
263 feature = "iroh",
264 feature = "websocket",
265 feature = "tcp",
266 feature = "uds"
267 )))]
268 pub fn new(_config: ClientConfig) -> crate::Result<Self> {
269 Err(Error::NoBackend(
270 "no backend compiled; enable noq, quinn, quiche, iroh, websocket, tcp, or uds feature",
271 ))
272 }
273
274 #[cfg(any(
276 feature = "noq",
277 feature = "quinn",
278 feature = "quiche",
279 feature = "iroh",
280 feature = "websocket",
281 feature = "tcp",
282 feature = "uds"
283 ))]
284 pub fn new(config: ClientConfig) -> crate::Result<Self> {
285 #[cfg(any(feature = "noq", feature = "quinn", feature = "quiche"))]
286 let backend = config.backend.clone().unwrap_or_else(crate::default_quic_backend);
287
288 config.quic.validate()?;
289 config.backoff.validate()?;
290
291 #[cfg(any(feature = "noq", feature = "quinn", feature = "websocket"))]
294 let tls = config.tls.build()?;
295
296 #[cfg(feature = "noq")]
297 #[allow(unreachable_patterns)]
298 let noq = match backend {
299 QuicBackend::Noq => Some(crate::noq::NoqClient::new(&config)?),
300 _ => None,
301 };
302
303 #[cfg(feature = "quinn")]
304 #[allow(unreachable_patterns)]
305 let quinn = match backend {
306 QuicBackend::Quinn => Some(crate::quinn::QuinnClient::new(&config)?),
307 _ => None,
308 };
309
310 #[cfg(feature = "quiche")]
311 #[allow(unreachable_patterns)]
312 let quiche = match backend {
313 QuicBackend::Quiche => Some(crate::quiche::QuicheClient::new(&config)?),
314 _ => None,
315 };
316
317 let versions = config.versions();
318 #[cfg(feature = "tcp")]
320 let failover_delay = config.resolved_failover_delay();
321 #[cfg(feature = "tcp")]
322 let resolution_delay = config.resolved_resolution_delay();
323 #[cfg(feature = "websocket")]
324 let tls_host_name = config.tls.host_name.clone();
325 let timeout = config.resolved_connect_timeout();
326
327 Ok(Self {
328 moq: moq_net::Client::new().with_versions(versions.clone()),
329 #[cfg(any(
330 feature = "noq",
331 feature = "quinn",
332 feature = "quiche",
333 feature = "websocket",
334 feature = "tcp",
335 feature = "uds"
336 ))]
337 versions,
338 connect: config.connect,
339 timeout,
340 backoff: config.backoff,
341 #[cfg(feature = "tcp")]
342 failover_delay,
343 #[cfg(feature = "tcp")]
344 resolution_delay,
345 #[cfg(feature = "websocket")]
346 websocket: config.websocket,
347 #[cfg(feature = "websocket")]
348 tls_host_name,
349 #[cfg(any(feature = "noq", feature = "quinn", feature = "websocket"))]
350 tls,
351 #[cfg(feature = "noq")]
352 noq,
353 #[cfg(feature = "quinn")]
354 quinn,
355 #[cfg(feature = "quiche")]
356 quiche,
357 #[cfg(feature = "iroh")]
358 iroh: None,
359 #[cfg(feature = "iroh")]
360 iroh_addrs: Vec::new(),
361 })
362 }
363
364 #[cfg(feature = "iroh")]
369 pub fn with_iroh(mut self, iroh: crate::iroh::Endpoint) -> Self {
370 self.iroh = Some(iroh);
371 self
372 }
373
374 #[cfg(feature = "iroh")]
379 pub fn with_iroh_addrs(mut self, addrs: Vec<std::net::SocketAddr>) -> Self {
380 self.iroh_addrs = addrs;
381 self
382 }
383
384 pub fn with_publisher(mut self, publish: impl moq_net::Consume<moq_net::origin::Consumer>) -> Self {
386 self.moq = self.moq.with_publisher(publish);
387 self
388 }
389
390 pub fn with_subscriber(mut self, subscribe: moq_net::origin::Producer) -> Self {
392 self.moq = self.moq.with_subscriber(subscribe);
393 self
394 }
395
396 pub fn with_stats(mut self, stats: moq_net::stats::Session) -> Self {
399 self.moq = self.moq.with_stats(stats);
400 self
401 }
402
403 pub fn with_cost(mut self, cost: u64) -> Self {
405 self.moq = self.moq.with_cost(cost);
406 self
407 }
408
409 pub fn with_peer_origin(mut self, origin: moq_net::Origin) -> Self {
412 self.moq = self.moq.with_peer_origin(origin);
413 self
414 }
415
416 pub fn reconnect(&self, url: Url) -> Reconnect {
421 Reconnect::new(self.clone(), url, self.backoff.clone())
422 }
423
424 pub fn publish(self, origin: moq_net::origin::Consumer) -> Option<Reconnect> {
430 let url = self.connect.clone()?;
431 Some(self.with_publisher(origin).reconnect(url))
432 }
433
434 pub fn consume(self, origin: moq_net::origin::Producer) -> Option<Reconnect> {
446 let url = self.connect.clone()?;
447 let origin = origin.with_linger(self.backoff.linger());
448 Some(self.with_subscriber(origin).reconnect(url))
449 }
450
451 #[cfg(not(any(
455 feature = "noq",
456 feature = "quinn",
457 feature = "quiche",
458 feature = "iroh",
459 feature = "websocket",
460 feature = "tcp",
461 feature = "uds"
462 )))]
463 pub async fn connect(&self, _url: Url) -> crate::Result<moq_net::Session> {
464 Err(Error::NoBackend(
465 "no backend compiled; enable noq, quinn, quiche, iroh, websocket, tcp, or uds feature",
466 ))
467 }
468
469 #[cfg(any(
476 feature = "noq",
477 feature = "quinn",
478 feature = "quiche",
479 feature = "iroh",
480 feature = "websocket",
481 feature = "tcp",
482 feature = "uds"
483 ))]
484 pub async fn connect(&self, url: Url) -> crate::Result<moq_net::Session> {
485 let attempt = Box::pin(self.connect_inner(url));
488
489 let pair = match self.timeout.is_zero() {
493 true => attempt.await?,
494 false => match tokio::time::timeout(self.timeout, attempt).await {
495 Ok(res) => res?,
496 Err(_) => return Err(Error::ConnectTimeout(self.timeout)),
497 },
498 };
499
500 tracing::info!(version = %pair.0.version(), "connected");
501 Ok(crate::spawn_session(pair))
502 }
503
504 #[cfg(any(
506 feature = "noq",
507 feature = "quinn",
508 feature = "quiche",
509 feature = "iroh",
510 feature = "websocket",
511 feature = "tcp",
512 feature = "uds"
513 ))]
514 fn moq_with_path(&self, path: Option<String>) -> moq_net::Client {
515 match path {
516 Some(path) => self.moq.clone().with_path(path),
517 None => self.moq.clone(),
518 }
519 }
520
521 #[cfg(any(
522 feature = "noq",
523 feature = "quinn",
524 feature = "quiche",
525 feature = "iroh",
526 feature = "websocket",
527 feature = "tcp",
528 feature = "uds"
529 ))]
530 async fn connect_inner(&self, url: Url) -> crate::Result<(moq_net::Session, moq_net::Driver)> {
531 #[allow(unused_variables)]
536 let moq = self.moq_with_path(setup_path(&url));
537
538 #[cfg(feature = "tcp")]
541 if url.scheme() == "tcp" {
542 let session =
543 crate::tcp::connect(url, &self.versions.alpns(), self.failover_delay, self.resolution_delay).await?;
544 return Ok(moq.connect(session).await?);
545 }
546
547 #[cfg(all(feature = "uds", unix))]
550 if url.scheme() == "unix" {
551 let session = crate::unix::connect(url, &self.versions.alpns()).await?;
552 return Ok(moq.connect(session).await?);
553 }
554
555 #[cfg(feature = "iroh")]
560 if url.scheme() == "iroh" {
561 let endpoint = self.iroh.as_ref().ok_or(Error::IrohDisabled)?;
562 let target = request_target(&url);
563 let (session, binding) = crate::iroh::connect(endpoint, url, self.iroh_addrs.iter().copied()).await?;
564
565 let moq = match binding {
566 crate::iroh::Binding::Raw => self.moq_with_path(target),
567 crate::iroh::Binding::H3 => self.moq.clone(),
568 };
569
570 return Ok(moq.connect(session).await?);
571 }
572
573 #[cfg(feature = "noq")]
574 if let Some(noq) = self.noq.as_ref() {
575 let tls = self.tls.clone();
576 let quic_url = url.clone();
577 let quic_handle = async { noq.connect(&tls, quic_url, &self.versions).await.map_err(Error::from) };
578
579 #[cfg(feature = "websocket")]
580 {
581 return self.race_moq_connect(&moq, url, quic_handle).await;
582 }
583
584 #[cfg(not(feature = "websocket"))]
585 {
586 let session = quic_handle.await?;
587 return Ok(moq.connect(session).await?);
588 }
589 }
590
591 #[cfg(feature = "quinn")]
592 if let Some(quinn) = self.quinn.as_ref() {
593 let tls = self.tls.clone();
594 let quic_url = url.clone();
595 let quic_handle = async { quinn.connect(&tls, quic_url, &self.versions).await.map_err(Error::from) };
596
597 #[cfg(feature = "websocket")]
598 {
599 return self.race_moq_connect(&moq, url, quic_handle).await;
600 }
601
602 #[cfg(not(feature = "websocket"))]
603 {
604 let session = quic_handle.await?;
605 return Ok(moq.connect(session).await?);
606 }
607 }
608
609 #[cfg(feature = "quiche")]
610 if let Some(quiche) = self.quiche.as_ref() {
611 let quic_url = url.clone();
612 let quic_handle = async { quiche.connect(quic_url, &self.versions).await.map_err(Error::from) };
613
614 #[cfg(feature = "websocket")]
615 {
616 return self.race_moq_connect(&moq, url, quic_handle).await;
617 }
618
619 #[cfg(not(feature = "websocket"))]
620 {
621 let session = quic_handle.await?;
622 return Ok(moq.connect(session).await?);
623 }
624 }
625
626 #[cfg(feature = "websocket")]
627 {
628 let alpns = self.versions.alpns();
629 let session =
630 crate::websocket::connect(&self.websocket, &self.tls, self.tls_host_name.as_deref(), url, &alpns)
631 .await?;
632 return Ok(moq.connect(session).await?);
633 }
634
635 #[cfg(not(feature = "websocket"))]
636 return Err(Error::NoBackend("no QUIC backend matched; this should not happen"));
637 }
638
639 #[cfg(all(feature = "websocket", any(feature = "noq", feature = "quinn", feature = "quiche")))]
648 async fn race_moq_connect<Q, S>(
649 &self,
650 moq: &moq_net::Client,
651 url: Url,
652 quic: Q,
653 ) -> crate::Result<(moq_net::Session, moq_net::Driver)>
654 where
655 Q: Future<Output = crate::Result<S>>,
656 S: web_transport_trait::Session,
657 {
658 let alpns = self.versions.alpns();
659 let ws_config = self.websocket.clone();
660 let ws_tls = self.tls.clone();
661 let ws_tls_host_name = self.tls_host_name.clone();
662 let websocket = async move {
663 crate::websocket::race_handle(&ws_config, &ws_tls, ws_tls_host_name.as_deref(), url, &alpns)
664 .await
665 .map(|res| res.map_err(Error::from))
666 };
667
668 match race_transport_connect(quic, websocket).await? {
669 TransportRace::Quic(quic) => Ok(moq.connect(quic).await?),
670 TransportRace::WebSocket(websocket) => Ok(self.moq.connect(websocket).await?),
671 }
672 }
673}
674
675#[cfg(any(
682 feature = "noq",
683 feature = "quinn",
684 feature = "quiche",
685 feature = "iroh",
686 feature = "websocket",
687 feature = "tcp",
688 feature = "uds"
689))]
690fn request_target(url: &Url) -> Option<String> {
691 let target = match url.query().filter(|query| !query.is_empty()) {
695 Some(query) => format!("{}?{}", url.path(), query),
696 None => url.path().to_owned(),
697 };
698
699 (!target.is_empty()).then_some(target)
700}
701
702#[cfg(any(
710 feature = "noq",
711 feature = "quinn",
712 feature = "quiche",
713 feature = "iroh",
714 feature = "websocket",
715 feature = "tcp",
716 feature = "uds"
717))]
718fn setup_path(url: &Url) -> Option<String> {
719 match url.scheme() {
720 "unix" => url
724 .query_pairs()
725 .find(|(k, _)| k == "path")
726 .map(|(_, v)| v.into_owned())
727 .filter(|path| !path.is_empty()),
728 "moqt" | "moql" | "tcp" => request_target(url),
731 _ => None,
732 }
733}
734
735#[cfg(all(feature = "websocket", any(feature = "noq", feature = "quinn", feature = "quiche")))]
736#[derive(Debug, PartialEq, Eq)]
737enum TransportRace<Q, W> {
738 Quic(Q),
739 WebSocket(W),
740}
741
742#[cfg(all(feature = "websocket", any(feature = "noq", feature = "quinn", feature = "quiche")))]
743async fn race_transport_connect<Q, W, QT, WT>(quic: Q, websocket: W) -> crate::Result<TransportRace<QT, WT>>
744where
745 Q: Future<Output = crate::Result<QT>>,
746 W: Future<Output = Option<crate::Result<WT>>>,
747{
748 tokio::pin!(quic);
749 tokio::pin!(websocket);
750
751 let mut quic_err = None;
752 let mut websocket_err = None;
753 let mut quic_done = false;
754 let mut websocket_done = false;
755
756 loop {
757 tokio::select! {
758 res = &mut quic, if !quic_done => {
759 match res {
760 Ok(session) => return Ok(TransportRace::Quic(session)),
761 Err(err) if err.is_auth() => return Err(err),
762 Err(err) => {
763 tracing::warn!(%err, "QUIC connection failed");
764 quic_err = Some(err);
765 quic_done = true;
766 }
767 }
768 }
769 res = &mut websocket, if !websocket_done => {
770 match res {
771 Some(Ok(session)) => return Ok(TransportRace::WebSocket(session)),
772 Some(Err(err)) if err.is_auth() => return Err(err),
773 Some(Err(err)) => {
774 tracing::warn!(%err, "WebSocket connection failed");
775 websocket_err = Some(err);
776 websocket_done = true;
777 }
778 None => {
779 websocket_done = true;
780 }
781 }
782 }
783 else => break,
784 }
785
786 if quic_done && websocket_done {
787 break;
788 }
789 }
790
791 match (quic_err, websocket_err) {
792 (Some(quic), Some(websocket)) => Err(Error::TransportRace {
793 quic: std::sync::Arc::new(quic),
794 websocket: std::sync::Arc::new(websocket),
795 }),
796 (Some(err), None) | (None, Some(err)) => Err(err),
797 (None, None) => Err(Error::ConnectFailed),
798 }
799}
800
801#[cfg(test)]
802mod tests {
803 use super::*;
804 use clap::{CommandFactory, Parser};
805
806 #[cfg(any(
807 feature = "noq",
808 feature = "quinn",
809 feature = "quiche",
810 feature = "iroh",
811 feature = "websocket",
812 feature = "tcp",
813 feature = "uds"
814 ))]
815 #[test]
816 fn setup_path_covers_the_uri_less_transports() {
817 let cases = [
820 ("unix:///run/moq.sock?path=/room", Some("/room")),
821 ("unix:///run/moq.sock?path=/room%3Fjwt%3Dabc", Some("/room?jwt=abc")),
824 ("unix:///run/moq.sock?path=", None),
825 ("unix:///run/moq.sock", None),
826 ("tcp://localhost:4443/room", Some("/room")),
827 ("tcp://localhost:4443/room?jwt=abc", Some("/room?jwt=abc")),
828 ("tcp://localhost:4443", None),
829 ("moqt://relay.example.com/anon", Some("/anon")),
832 ("moqt://relay.example.com/anon?jwt=abc", Some("/anon?jwt=abc")),
833 ("moql://relay.example.com/anon?jwt=abc", Some("/anon?jwt=abc")),
834 ("moqt://relay.example.com", None),
835 ("moqt://relay.example.com/anon?jwt=abc#pos:12", Some("/anon?jwt=abc")),
837 ("moqt://relay.example.com/anon#pos:12", Some("/anon")),
838 ("moqt://relay.example.com/anon?", Some("/anon")),
840 ("moqt://relay.example.com?", None),
841 ("https://relay.example.com/anon?jwt=abc", None),
844 ("http://relay.example.com/anon", None),
845 ("wss://relay.example.com/anon?jwt=abc", None),
846 ("iroh://k5lnrlndqpqcgh4d5nhbnbnhcyrgvw6ttxwrsvsu4nlt6foorxaa/anon", None),
848 ];
849
850 for (url, want) in cases {
851 let url = Url::parse(url).unwrap();
852 let got = setup_path(&url);
853 assert_eq!(got.as_deref(), want, "{url}");
854 }
855 }
856
857 #[cfg(any(
860 feature = "noq",
861 feature = "quinn",
862 feature = "quiche",
863 feature = "iroh",
864 feature = "websocket",
865 feature = "tcp",
866 feature = "uds"
867 ))]
868 #[test]
869 fn request_target_joins_the_path_and_query() {
870 const PEER: &str = "k5lnrlndqpqcgh4d5nhbnbnhcyrgvw6ttxwrsvsu4nlt6foorxaa";
871
872 let cases = [
873 (format!("iroh://{PEER}/room?jwt=abc"), Some("/room?jwt=abc")),
874 (format!("iroh://{PEER}/room"), Some("/room")),
875 (format!("iroh://{PEER}"), None),
876 (format!("iroh://{PEER}/"), Some("/")),
877 ];
878
879 for (url, want) in cases {
880 let url = Url::parse(&url).unwrap();
881 let got = request_target(&url);
882 assert_eq!(got.as_deref(), want, "{url}");
883 }
884 }
885
886 #[test]
887 fn test_toml_disable_verify_survives_update_from() {
888 let toml = r#"
889 tls.disable_verify = true
890 "#;
891
892 let mut config: ClientConfig = toml::from_str(toml).unwrap();
893 assert_eq!(config.tls.disable_verify, Some(true));
894
895 config.update_from(["test"]);
897 assert_eq!(config.tls.disable_verify, Some(true));
898 }
899
900 #[test]
901 fn test_cli_disable_verify_flag() {
902 let config = ClientConfig::parse_from(["test", "--client-tls-disable-verify"]);
903 assert_eq!(config.tls.disable_verify, Some(true));
904 }
905
906 #[test]
907 fn test_cli_disable_verify_explicit_false() {
908 let config = ClientConfig::parse_from(["test", "--client-tls-disable-verify=false"]);
909 assert_eq!(config.tls.disable_verify, Some(false));
910 }
911
912 #[test]
913 fn test_cli_disable_verify_explicit_true() {
914 let config = ClientConfig::parse_from(["test", "--client-tls-disable-verify=true"]);
915 assert_eq!(config.tls.disable_verify, Some(true));
916 }
917
918 #[test]
919 fn test_cli_deprecated_tls_flags_fold_into_canonical() {
920 let config = ClientConfig::parse_from(["test", "--tls-disable-verify=true", "--tls-fingerprint", "abcd1234"]);
924 assert_eq!(
925 config.tls.disable_verify, None,
926 "deprecated flag must not set the canonical field"
927 );
928 assert_eq!(config.tls.effective_disable_verify(), Some(true));
929 assert_eq!(config.tls.effective_fingerprint(), vec!["abcd1234"]);
930 }
931
932 #[test]
933 fn test_canonical_tls_flag_wins_over_deprecated() {
934 let config = ClientConfig::parse_from([
936 "test",
937 "--client-tls-disable-verify=false",
938 "--tls-disable-verify=true",
939 "--client-tls-fingerprint",
940 "aaaa",
941 "--tls-fingerprint",
942 "bbbb",
943 ]);
944 assert_eq!(config.tls.effective_disable_verify(), Some(false));
945 assert_eq!(config.tls.effective_fingerprint(), vec!["aaaa", "bbbb"]);
946 }
947
948 #[test]
949 fn test_cli_no_disable_verify() {
950 let config = ClientConfig::parse_from(["test"]);
951 assert_eq!(config.tls.disable_verify, None);
952 }
953
954 #[test]
955 fn test_toml_failover_delay_survives_update_from() {
956 let toml = r#"
957 failover_delay = "1s"
958 "#;
959
960 let mut config: ClientConfig = toml::from_str(toml).unwrap();
961 assert_eq!(config.failover_delay, Some(std::time::Duration::from_secs(1)));
962
963 config.update_from(["test"]);
965 assert_eq!(config.failover_delay, Some(std::time::Duration::from_secs(1)));
966 }
967
968 #[test]
969 fn test_cli_failover_delay() {
970 let config = ClientConfig::parse_from(["test", "--client-failover-delay", "50ms"]);
971 assert_eq!(config.failover_delay, Some(std::time::Duration::from_millis(50)));
972 }
973
974 #[test]
975 fn test_toml_resolution_delay_survives_update_from() {
976 let toml = r#"
977 resolution_delay = "10ms"
978 "#;
979
980 let mut config: ClientConfig = toml::from_str(toml).unwrap();
981 assert_eq!(config.resolution_delay, Some(std::time::Duration::from_millis(10)));
982
983 config.update_from(["test"]);
985 assert_eq!(config.resolution_delay, Some(std::time::Duration::from_millis(10)));
986 }
987
988 #[test]
989 fn test_cli_resolution_delay() {
990 let config = ClientConfig::parse_from(["test", "--client-resolution-delay", "0s"]);
991 assert_eq!(config.resolution_delay, Some(std::time::Duration::ZERO));
992 assert_eq!(config.resolved_resolution_delay(), std::time::Duration::ZERO);
993 }
994
995 #[test]
996 fn resolution_delay_defaults_to_the_rfc_value() {
997 let config = ClientConfig::parse_from(["test"]);
998 assert_eq!(config.resolution_delay, None);
999 assert_eq!(config.resolved_resolution_delay(), std::time::Duration::from_millis(50));
1000 }
1001
1002 #[test]
1003 fn test_toml_fingerprint_survives_update_from() {
1004 let toml = r#"
1005 tls.fingerprint = ["abcd1234", "ef567890"]
1006 "#;
1007
1008 let mut config: ClientConfig = toml::from_str(toml).unwrap();
1009 assert_eq!(config.tls.fingerprint, vec!["abcd1234", "ef567890"]);
1010
1011 config.update_from(["test"]);
1013 assert_eq!(config.tls.fingerprint, vec!["abcd1234", "ef567890"]);
1014 }
1015
1016 #[test]
1017 fn test_toml_fingerprint_accepts_single_string() {
1018 let toml = r#"
1019 tls.fingerprint = "abcd1234"
1020 "#;
1021
1022 let config: ClientConfig = toml::from_str(toml).unwrap();
1023 assert_eq!(config.tls.fingerprint, vec!["abcd1234"]);
1024 }
1025
1026 #[test]
1027 fn test_cli_fingerprint() {
1028 let config = ClientConfig::parse_from(["test", "--client-tls-fingerprint", "abcd1234"]);
1029 assert_eq!(config.tls.fingerprint, vec!["abcd1234"]);
1030 }
1031
1032 #[test]
1033 fn test_toml_version_survives_update_from() {
1034 let toml = r#"
1035 version = ["moq-lite-02"]
1036 "#;
1037
1038 let mut config: ClientConfig = toml::from_str(toml).unwrap();
1039 assert_eq!(config.version, vec!["moq-lite-02".parse::<moq_net::Version>().unwrap()]);
1040
1041 config.update_from(["test"]);
1043 assert_eq!(config.version, vec!["moq-lite-02".parse::<moq_net::Version>().unwrap()]);
1044 }
1045
1046 #[test]
1047 fn test_cli_version() {
1048 let config = ClientConfig::parse_from(["test", "--client-version", "moq-lite-03"]);
1049 assert_eq!(config.version, vec!["moq-lite-03".parse::<moq_net::Version>().unwrap()]);
1050 }
1051
1052 #[test]
1053 fn test_cli_version_help_lists_every_parseable_name() {
1054 let help = ClientConfig::command().render_long_help().to_string();
1055 for name in moq_net::Version::names() {
1056 assert!(help.contains(name), "missing {name} from --client-version help");
1057 }
1058 }
1059
1060 #[test]
1061 fn test_toml_connect_survives_update_from() {
1062 let toml = r#"
1063 connect = "https://relay.example.com/anon"
1064 "#;
1065
1066 let mut config: ClientConfig = toml::from_str(toml).unwrap();
1067 assert_eq!(
1068 config.connect.as_ref().unwrap().as_str(),
1069 "https://relay.example.com/anon"
1070 );
1071
1072 config.update_from(["test"]);
1074 assert_eq!(
1075 config.connect.as_ref().unwrap().as_str(),
1076 "https://relay.example.com/anon"
1077 );
1078 }
1079
1080 #[test]
1081 fn test_cli_connect() {
1082 let config = ClientConfig::parse_from(["test", "--client-connect", "https://relay.example.com/anon"]);
1083 assert_eq!(
1084 config.connect.as_ref().unwrap().as_str(),
1085 "https://relay.example.com/anon"
1086 );
1087 }
1088
1089 #[test]
1090 fn test_toml_host_name_survives_update_from() {
1091 let toml = r#"
1092 tls.host_name = "example.host"
1093 "#;
1094
1095 let mut config: ClientConfig = toml::from_str(toml).unwrap();
1096 assert_eq!(config.tls.host_name.as_deref(), Some("example.host"));
1097
1098 config.update_from(["test"]);
1100 assert_eq!(config.tls.host_name.as_deref(), Some("example.host"));
1101 }
1102
1103 #[test]
1104 fn test_cli_host_name() {
1105 let config = ClientConfig::parse_from(["test", "--client-tls-host-name", "override.example"]);
1106 assert_eq!(config.tls.host_name.as_deref(), Some("override.example"));
1107 }
1108
1109 #[test]
1110 fn test_cli_no_version_defaults_to_all() {
1111 let config = ClientConfig::parse_from(["test"]);
1112 assert!(config.version.is_empty());
1113 assert_eq!(config.versions().alpns().len(), moq_net::ALPNS.len());
1115 }
1116
1117 #[cfg(all(feature = "websocket", any(feature = "noq", feature = "quinn", feature = "quiche")))]
1118 #[tokio::test]
1119 async fn race_transport_connect_stops_on_quic_auth_error() {
1120 let quic = async { Err::<usize, _>(crate::ConnectError::Unauthorized.into()) };
1121 let websocket = async {
1122 tokio::task::yield_now().await;
1124 Some(Ok(1usize))
1125 };
1126
1127 let err = super::race_transport_connect(quic, websocket).await.unwrap_err();
1128 assert_eq!(err.connect_error(), Some(crate::ConnectError::Unauthorized));
1129 }
1130
1131 #[cfg(all(feature = "websocket", any(feature = "noq", feature = "quinn", feature = "quiche")))]
1132 #[tokio::test]
1133 async fn race_transport_connect_keeps_websocket_after_quic_non_auth_error() {
1134 let quic = async { Err::<usize, _>(Error::ConnectFailed) };
1135 let websocket = async { Some(Ok(7usize)) };
1136
1137 let value = super::race_transport_connect(quic, websocket).await.unwrap();
1138 assert_eq!(value, super::TransportRace::WebSocket(7));
1139 }
1140
1141 #[cfg(all(feature = "websocket", any(feature = "noq", feature = "quinn", feature = "quiche")))]
1142 #[tokio::test]
1143 async fn race_transport_connect_returns_when_quic_transport_connects() {
1144 let quic = async { Ok("quic") };
1145 let websocket = std::future::pending::<Option<crate::Result<&str>>>();
1146
1147 let value = tokio::time::timeout(
1148 std::time::Duration::from_secs(1),
1149 super::race_transport_connect(quic, websocket),
1150 )
1151 .await
1152 .expect("race waited for WebSocket after QUIC transport connected")
1153 .unwrap();
1154 assert_eq!(value, super::TransportRace::Quic("quic"));
1155 }
1156
1157 #[test]
1160 fn failover_delay_defaults_to_the_rfc_8305_stagger() {
1161 let config = ClientConfig::parse_from(["test"]);
1162 assert_eq!(config.failover_delay, None);
1163 assert_eq!(config.resolved_failover_delay(), std::time::Duration::from_millis(250));
1164 }
1165
1166 #[cfg(all(
1169 feature = "iroh",
1170 not(any(feature = "noq", feature = "quinn", feature = "websocket"))
1171 ))]
1172 #[test]
1173 fn iroh_only_client_does_not_require_a_tls_provider() {
1174 ClientConfig::default().init().expect("iroh-only client");
1175 }
1176
1177 #[test]
1178 fn connect_timeout_defaults_to_thirty_seconds() {
1179 let config = ClientConfig::parse_from(["test"]);
1180 assert_eq!(config.timeout, None);
1181 assert_eq!(config.resolved_connect_timeout(), DEFAULT_CONNECT_TIMEOUT);
1182 }
1183
1184 #[cfg(feature = "websocket")]
1192 #[tokio::test]
1193 async fn connect_times_out_against_a_peer_that_never_speaks() {
1194 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1195 let addr = listener.local_addr().unwrap();
1196
1197 let timeout = DEFAULT_CONNECT_TIMEOUT;
1198 let mut config = ClientConfig {
1199 timeout: Some(timeout),
1200 ..Default::default()
1201 };
1202 config.websocket.delay = Some(std::time::Duration::ZERO);
1203 let client = config.init().unwrap();
1204
1205 let url: Url = format!("https://127.0.0.1:{}/", addr.port()).parse().unwrap();
1208
1209 let mut attempt = Box::pin(client.connect(url));
1210 let _silent = tokio::select! {
1211 res = &mut attempt => match res {
1212 Err(err) => panic!("connect failed before the silent peer accepted it: {err}"),
1213 Ok(_) => panic!("connected to a peer that never spoke"),
1214 },
1215 res = listener.accept() => res.unwrap().0,
1216 };
1217
1218 tokio::time::pause();
1221 tokio::time::advance(timeout).await;
1222
1223 let err = match attempt.await {
1224 Err(err) => err,
1225 Ok(_) => panic!("connected to a peer that never spoke"),
1226 };
1227
1228 assert!(matches!(err, Error::ConnectTimeout(_)), "unexpected error: {err}");
1229 }
1230}