1use std::time::Duration;
2
3use camel_component_api::{CamelError, NetworkRetryPolicy};
4
5#[derive(Debug, Clone, Default, serde::Deserialize)]
6pub struct WsConfig {
7 pub max_connections: Option<u32>,
8 pub max_message_size: Option<u32>,
9 pub heartbeat_interval_ms: Option<u64>,
10 pub idle_timeout_ms: Option<u64>,
11 pub connect_timeout_ms: Option<u64>,
12 pub response_timeout_ms: Option<u64>,
13 pub send_timeout_ms: Option<u64>,
14 pub binary_payload: Option<bool>,
15 pub subprotocols: Option<Vec<String>>,
16}
17
18#[derive(Clone)]
19pub struct WsEndpointConfig {
20 pub scheme: String,
21 pub host: String,
22 pub port: u16,
23 pub path: String,
24 pub max_connections: u32,
25 pub max_message_size: u32,
26 pub send_to_all: bool,
27 pub heartbeat_interval: Duration,
28 pub idle_timeout: Duration,
29 pub connect_timeout: Duration,
30 pub response_timeout: Duration,
31 pub allow_origin: String,
32 pub tls_cert: Option<String>,
33 pub tls_key: Option<String>,
34 pub reconnect: bool,
35 pub reconnect_max_attempts: u32,
36 pub reconnect_delay_ms: u64,
37 pub send_timeout: Duration,
38 pub binary_payload: bool,
39 pub subprotocols: Vec<String>,
40 pub reconnect_policy: NetworkRetryPolicy,
45}
46
47fn redacted_opt(opt: &Option<String>) -> Option<&'static str> {
48 if opt.is_some() { Some("***") } else { None }
49}
50
51impl std::fmt::Debug for WsEndpointConfig {
52 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
53 f.debug_struct("WsEndpointConfig")
54 .field("scheme", &self.scheme)
55 .field("host", &self.host)
56 .field("port", &self.port)
57 .field("path", &self.path)
58 .field("max_connections", &self.max_connections)
59 .field("max_message_size", &self.max_message_size)
60 .field("send_to_all", &self.send_to_all)
61 .field("heartbeat_interval", &self.heartbeat_interval)
62 .field("idle_timeout", &self.idle_timeout)
63 .field("connect_timeout", &self.connect_timeout)
64 .field("response_timeout", &self.response_timeout)
65 .field("allow_origin", &self.allow_origin)
66 .field("tls_cert", &redacted_opt(&self.tls_cert))
67 .field("tls_key", &redacted_opt(&self.tls_key))
68 .field("reconnect", &self.reconnect)
69 .field("reconnect_max_attempts", &self.reconnect_max_attempts)
70 .field("reconnect_delay_ms", &self.reconnect_delay_ms)
71 .field("reconnect_policy", &self.reconnect_policy)
72 .field("send_timeout", &self.send_timeout)
73 .field("binary_payload", &self.binary_payload)
74 .field("subprotocols", &self.subprotocols)
75 .finish()
76 }
77}
78
79impl Default for WsEndpointConfig {
80 fn default() -> Self {
81 Self {
82 scheme: "ws".into(),
83 host: "0.0.0.0".into(),
84 port: 8080,
85 path: "/".into(),
86 max_connections: 100,
87 max_message_size: 65536,
88 send_to_all: false,
89 heartbeat_interval: Duration::ZERO,
90 idle_timeout: Duration::ZERO,
91 connect_timeout: Duration::from_secs(10),
92 response_timeout: Duration::from_secs(30),
93 allow_origin: "*".into(),
94 tls_cert: None,
95 tls_key: None,
96 reconnect: true,
97 reconnect_max_attempts: 5,
98 reconnect_delay_ms: 1000,
99 send_timeout: Duration::from_secs(30),
100 binary_payload: false,
101 subprotocols: Vec::new(),
102 reconnect_policy: NetworkRetryPolicy {
103 enabled: true,
104 max_attempts: 5,
105 initial_delay: Duration::from_millis(1000),
106 multiplier: 2.0,
107 max_delay: Duration::from_secs(30),
108 jitter_factor: 0.0, max_attempts_absolute: None,
110 },
111 }
112 }
113}
114
115#[derive(Debug, Clone)]
116pub struct WsServerConfig {
117 pub inner: WsEndpointConfig,
118}
119
120#[derive(Debug, Clone)]
121pub struct WsClientConfig {
122 pub inner: WsEndpointConfig,
123}
124
125impl WsConfig {
126 pub fn validate(&self) -> Result<(), CamelError> {
131 if let Some(0) = self.max_connections {
132 return Err(CamelError::Config(
133 "maxConnections must be >= 1 when specified".into(),
134 ));
135 }
136 if let Some(0) = self.max_message_size {
137 return Err(CamelError::Config(
138 "maxMessageSize must be >= 1 when specified".into(),
139 ));
140 }
141 Ok(())
142 }
143}
144
145impl WsEndpointConfig {
146 pub fn from_uri(uri: &str) -> Result<Self, CamelError> {
147 let parsed = camel_component_api::parse_uri(uri)
148 .map_err(|e| CamelError::EndpointCreationFailed(e.to_string()))?;
149
150 let scheme = parsed.scheme;
151 if scheme != "ws" && scheme != "wss" {
152 return Err(CamelError::EndpointCreationFailed(format!(
153 "Invalid WebSocket scheme: {scheme}"
154 )));
155 }
156
157 let host_port_path = parsed.path;
158 let host_port_path = host_port_path.strip_prefix("//").unwrap_or(&host_port_path);
159 let (host_port, path) = match host_port_path.split_once('/') {
160 Some((hp, p)) => (hp, format!("/{p}")),
161 None => (host_port_path, "/".to_string()),
162 };
163
164 let (host, port) = match host_port.rsplit_once(':') {
165 Some((h, p)) if p.parse::<u16>().is_ok() => {
166 let parsed_port = p.parse::<u16>().unwrap(); (h.to_string(), parsed_port)
168 }
169 _ => (
170 host_port.to_string(),
171 if scheme == "wss" { 443 } else { 80 },
172 ),
173 };
174
175 let mut cfg = Self {
176 scheme,
177 host: if host.is_empty() {
178 "0.0.0.0".to_string()
179 } else {
180 host
181 },
182 port,
183 path,
184 ..Self::default()
185 };
186
187 let params = parsed.params;
188 if let Some(raw) = params.get("maxConnections") {
190 let v = raw.parse::<u32>().map_err(|_| {
191 CamelError::InvalidUri(format!(
192 "maxConnections must be an unsigned integer, got '{raw}'"
193 ))
194 })?;
195 if v == 0 {
196 return Err(CamelError::InvalidUri("maxConnections must be >= 1".into()));
197 }
198 cfg.max_connections = v;
199 }
200 if let Some(raw) = params.get("maxMessageSize") {
202 let v = raw.parse::<u32>().map_err(|_| {
203 CamelError::InvalidUri(format!(
204 "maxMessageSize must be an unsigned integer, got '{raw}'"
205 ))
206 })?;
207 if v == 0 {
208 return Err(CamelError::InvalidUri("maxMessageSize must be > 0".into()));
209 }
210 cfg.max_message_size = v;
211 }
212 if let Some(raw) = params.get("sendToAll") {
213 let v = raw.parse::<bool>().map_err(|_| {
214 CamelError::InvalidUri(format!(
215 "sendToAll must be a boolean ('true' or 'false'), got '{raw}'"
216 ))
217 })?;
218 cfg.send_to_all = v;
219 }
220 if let Some(raw) = params.get("heartbeatIntervalMs") {
221 let v = raw.parse::<u64>().map_err(|_| {
222 CamelError::InvalidUri(format!(
223 "heartbeatIntervalMs must be an unsigned integer, got '{raw}'"
224 ))
225 })?;
226 cfg.heartbeat_interval = Duration::from_millis(v);
227 }
228 if let Some(raw) = params.get("idleTimeoutMs") {
229 let v = raw.parse::<u64>().map_err(|_| {
230 CamelError::InvalidUri(format!(
231 "idleTimeoutMs must be an unsigned integer, got '{raw}'"
232 ))
233 })?;
234 cfg.idle_timeout = Duration::from_millis(v);
235 }
236 if let Some(raw) = params.get("connectTimeoutMs") {
237 let v = raw.parse::<u64>().map_err(|_| {
238 CamelError::InvalidUri(format!(
239 "connectTimeoutMs must be an unsigned integer, got '{raw}'"
240 ))
241 })?;
242 cfg.connect_timeout = Duration::from_millis(v);
243 }
244 if let Some(raw) = params.get("responseTimeoutMs") {
245 let v = raw.parse::<u64>().map_err(|_| {
246 CamelError::InvalidUri(format!(
247 "responseTimeoutMs must be an unsigned integer, got '{raw}'"
248 ))
249 })?;
250 cfg.response_timeout = Duration::from_millis(v);
251 }
252 if let Some(v) = params.get("allowOrigin") {
253 if v.is_empty() {
254 return Err(CamelError::InvalidUri(
255 "allowOrigin must not be empty when specified".into(),
256 ));
257 }
258 cfg.allow_origin = v.to_string();
259 }
260 if let Some(v) = params.get("tlsCert") {
261 cfg.tls_cert = Some(v.to_string());
262 }
263 if let Some(v) = params.get("tlsKey") {
264 cfg.tls_key = Some(v.to_string());
265 }
266 let mut reconnect_explicit = false;
271 let mut reconnect_max_attempts_explicit = false;
272 let mut reconnect_delay_ms_explicit = false;
273
274 if let Some(raw) = params.get("reconnect") {
275 cfg.reconnect = raw.parse::<bool>().map_err(|_| {
276 CamelError::InvalidUri(format!(
277 "reconnect must be a boolean ('true' or 'false'), got '{raw}'"
278 ))
279 })?;
280 reconnect_explicit = true;
281 }
282 if let Some(raw) = params.get("reconnectMaxAttempts") {
283 cfg.reconnect_max_attempts = raw.parse::<u32>().map_err(|_| {
284 CamelError::InvalidUri(format!(
285 "reconnectMaxAttempts must be an unsigned integer, got '{raw}'"
286 ))
287 })?;
288 reconnect_max_attempts_explicit = true;
289 }
290 if let Some(raw) = params.get("reconnectDelayMs") {
291 cfg.reconnect_delay_ms = raw.parse::<u64>().map_err(|_| {
292 CamelError::InvalidUri(format!(
293 "reconnectDelayMs must be an unsigned integer, got '{raw}'"
294 ))
295 })?;
296 reconnect_delay_ms_explicit = true;
297 }
298 if let Some(raw) = params.get("sendTimeoutMs") {
299 let v = raw.parse::<u64>().map_err(|_| {
300 CamelError::InvalidUri(format!(
301 "sendTimeoutMs must be an unsigned integer, got '{raw}'"
302 ))
303 })?;
304 cfg.send_timeout = Duration::from_millis(v);
305 }
306 if let Some(raw) = params.get("binaryPayload") {
307 cfg.binary_payload = raw.parse::<bool>().map_err(|_| {
308 CamelError::InvalidUri(format!(
309 "binaryPayload must be a boolean ('true' or 'false'), got '{raw}'"
310 ))
311 })?;
312 }
313 if let Some(raw) = params.get("subprotocols") {
314 cfg.subprotocols = raw
315 .split(',')
316 .map(|s| s.trim().to_string())
317 .filter(|s| !s.is_empty())
318 .collect();
319 }
320
321 if reconnect_explicit {
330 cfg.reconnect_policy.enabled = cfg.reconnect;
331 }
332 if reconnect_max_attempts_explicit {
333 cfg.reconnect_policy.max_attempts = cfg.reconnect_max_attempts;
334 }
335 if reconnect_delay_ms_explicit {
336 cfg.reconnect_policy.initial_delay = Duration::from_millis(cfg.reconnect_delay_ms);
337 }
338
339 Ok(cfg)
340 }
341
342 pub fn server_config(&self) -> WsServerConfig {
343 WsServerConfig {
344 inner: self.clone(),
345 }
346 }
347
348 pub fn client_config(&self) -> WsClientConfig {
349 WsClientConfig {
350 inner: self.clone(),
351 }
352 }
353
354 pub fn canonical_host(&self) -> String {
355 match self.host.as_str() {
356 "0.0.0.0" | "localhost" => "127.0.0.1".to_string(),
357 h => h.to_string(),
358 }
359 }
360}
361
362#[cfg(test)]
363mod config_validation_tests {
364 use super::*;
365
366 #[test]
367 fn test_rejects_zero_max_connections() {
368 let cfg = WsConfig {
369 max_connections: Some(0),
370 ..WsConfig::default()
371 };
372 assert!(cfg.validate().is_err());
373 }
374
375 #[test]
376 fn test_rejects_zero_max_message_size() {
377 let cfg = WsConfig {
378 max_message_size: Some(0),
379 ..WsConfig::default()
380 };
381 assert!(cfg.validate().is_err());
382 }
383
384 #[test]
385 fn test_accepts_valid_config() {
386 let cfg = WsConfig::default();
387 assert!(cfg.validate().is_ok());
388 }
389
390 #[test]
391 fn test_accepts_nonzero_max_connections() {
392 let cfg = WsConfig {
393 max_connections: Some(50),
394 ..WsConfig::default()
395 };
396 assert!(cfg.validate().is_ok());
397 }
398
399 #[test]
400 fn test_accepts_nonzero_max_message_size() {
401 let cfg = WsConfig {
402 max_message_size: Some(1024),
403 ..WsConfig::default()
404 };
405 assert!(cfg.validate().is_ok());
406 }
407
408 #[test]
409 fn test_from_uri_rejects_invalid_send_to_all() {
410 let err = WsEndpointConfig::from_uri("ws://localhost:8080?sendToAll=yes").unwrap_err();
411 assert!(err.to_string().contains("sendToAll"));
412 }
413
414 #[test]
415 fn test_from_uri_rejects_invalid_max_connections_numeric() {
416 let err = WsEndpointConfig::from_uri("ws://localhost:8080?maxConnections=abc").unwrap_err();
417 assert!(err.to_string().contains("maxConnections"));
418 }
419
420 #[test]
421 fn test_from_uri_rejects_invalid_max_message_size_numeric() {
422 let err = WsEndpointConfig::from_uri("ws://localhost:8080?maxMessageSize=abc").unwrap_err();
423 assert!(err.to_string().contains("maxMessageSize"));
424 }
425
426 #[test]
427 fn test_from_uri_rejects_invalid_heartbeat_interval_numeric() {
428 let err =
429 WsEndpointConfig::from_uri("ws://localhost:8080?heartbeatIntervalMs=abc").unwrap_err();
430 assert!(err.to_string().contains("heartbeatIntervalMs"));
431 }
432
433 #[test]
434 fn test_from_uri_rejects_invalid_idle_timeout_numeric() {
435 let err = WsEndpointConfig::from_uri("ws://localhost:8080?idleTimeoutMs=abc").unwrap_err();
436 assert!(err.to_string().contains("idleTimeoutMs"));
437 }
438
439 #[test]
440 fn test_from_uri_rejects_invalid_connect_timeout_numeric() {
441 let err =
442 WsEndpointConfig::from_uri("ws://localhost:8080?connectTimeoutMs=abc").unwrap_err();
443 assert!(err.to_string().contains("connectTimeoutMs"));
444 }
445
446 #[test]
447 fn test_from_uri_rejects_invalid_response_timeout_numeric() {
448 let err =
449 WsEndpointConfig::from_uri("ws://localhost:8080?responseTimeoutMs=abc").unwrap_err();
450 assert!(err.to_string().contains("responseTimeoutMs"));
451 }
452
453 #[test]
455 fn test_from_uri_parses_send_timeout_ms() {
456 let cfg = WsEndpointConfig::from_uri("ws://localhost:8080?sendTimeoutMs=7500").unwrap();
457 assert_eq!(cfg.send_timeout, Duration::from_millis(7500));
458 }
459
460 #[test]
461 fn test_from_uri_rejects_invalid_send_timeout_ms() {
462 let err = WsEndpointConfig::from_uri("ws://localhost:8080?sendTimeoutMs=xyz").unwrap_err();
463 assert!(err.to_string().contains("sendTimeoutMs"));
464 }
465
466 #[test]
468 fn test_from_uri_parses_binary_payload_true() {
469 let cfg = WsEndpointConfig::from_uri("ws://localhost:8080?binaryPayload=true").unwrap();
470 assert!(cfg.binary_payload);
471 }
472
473 #[test]
474 fn test_from_uri_parses_binary_payload_false() {
475 let cfg = WsEndpointConfig::from_uri("ws://localhost:8080?binaryPayload=false").unwrap();
476 assert!(!cfg.binary_payload);
477 }
478
479 #[test]
480 fn test_from_uri_rejects_invalid_binary_payload() {
481 let err = WsEndpointConfig::from_uri("ws://localhost:8080?binaryPayload=sure").unwrap_err();
482 assert!(err.to_string().contains("binaryPayload"));
483 }
484
485 #[test]
487 fn test_from_uri_parses_subprotocols() {
488 let cfg =
489 WsEndpointConfig::from_uri("ws://localhost:8080?subprotocols=json,protobuf").unwrap();
490 assert_eq!(cfg.subprotocols, vec!["json", "protobuf"]);
491 }
492
493 #[test]
494 fn test_from_uri_subprotocols_trims_whitespace() {
495 let cfg = WsEndpointConfig::from_uri("ws://localhost:8080?subprotocols=a, b").unwrap();
496 assert_eq!(cfg.subprotocols, vec!["a", "b"]);
497 }
498
499 #[test]
500 fn test_from_uri_subprotocols_empty_when_not_specified() {
501 let cfg = WsEndpointConfig::from_uri("ws://localhost:8080").unwrap();
502 assert!(cfg.subprotocols.is_empty());
503 }
504
505 #[test]
506 fn ws_endpoint_config_debug_redacts_tls() {
507 let config = WsEndpointConfig {
508 tls_cert: Some("/secret/cert.pem".to_string()),
509 tls_key: Some("/secret/key.pem".to_string()),
510 ..WsEndpointConfig::default()
511 };
512 let debug = format!("{:?}", config);
513 assert!(
514 !debug.contains("/secret/"),
515 "TLS paths must be redacted: {debug}"
516 );
517 assert!(
518 debug.contains("tls_cert"),
519 "field name should appear: {debug}"
520 );
521 assert!(
522 debug.contains("tls_key"),
523 "field name should appear: {debug}"
524 );
525 }
526
527 #[test]
528 fn ws_endpoint_config_has_reconnect_policy_field() {
529 let cfg = WsEndpointConfig::default();
530 assert!(cfg.reconnect_policy.enabled);
531 assert_eq!(cfg.reconnect_policy.max_attempts, 5);
532 assert_eq!(
533 cfg.reconnect_policy.initial_delay,
534 std::time::Duration::from_millis(1000)
535 );
536 }
537
538 #[test]
539 fn ws_endpoint_uri_bridges_flat_fields_to_policy() {
540 let uri =
542 "ws://localhost:9001/test?reconnect=false&reconnectMaxAttempts=7&reconnectDelayMs=250";
543 let cfg = WsEndpointConfig::from_uri(uri).unwrap();
544 assert!(!cfg.reconnect);
545 assert_eq!(cfg.reconnect_max_attempts, 7);
546 assert_eq!(cfg.reconnect_delay_ms, 250);
547 assert!(!cfg.reconnect_policy.enabled);
549 assert_eq!(cfg.reconnect_policy.max_attempts, 7);
550 assert_eq!(
551 cfg.reconnect_policy.initial_delay,
552 std::time::Duration::from_millis(250)
553 );
554 }
555
556 #[test]
557 fn ws_endpoint_policy_defaults_match_old_flat_defaults() {
558 let cfg = WsEndpointConfig::default();
559 assert!(cfg.reconnect_policy.enabled);
560 assert_eq!(cfg.reconnect_policy.max_attempts, 5);
561 assert_eq!(
562 cfg.reconnect_policy.initial_delay,
563 std::time::Duration::from_millis(1000)
564 );
565 assert!((cfg.reconnect_policy.multiplier - 2.0).abs() < f64::EPSILON);
566 assert_eq!(
567 cfg.reconnect_policy.max_delay,
568 std::time::Duration::from_secs(30)
569 );
570 assert!((cfg.reconnect_policy.jitter_factor - 0.0).abs() < f64::EPSILON);
571 }
572
573 #[test]
579 fn ws_endpoint_policy_preserved_when_no_flat_fields() {
580 let cfg = WsEndpointConfig::from_uri("ws://localhost:8080/echo").unwrap();
581 assert!(cfg.reconnect_policy.enabled);
583 assert_eq!(cfg.reconnect_policy.max_attempts, 5);
584 assert_eq!(
585 cfg.reconnect_policy.initial_delay,
586 std::time::Duration::from_millis(1000)
587 );
588 }
589
590 #[test]
600 fn ws_endpoint_policy_from_toml_preserved_across_from_uri() {
601 let toml_policy = NetworkRetryPolicy {
604 enabled: true,
605 max_attempts: 10,
606 initial_delay: std::time::Duration::from_millis(250),
607 multiplier: 3.0,
608 max_delay: std::time::Duration::from_secs(60),
609 jitter_factor: 0.1,
610 max_attempts_absolute: None,
611 };
612 let mut cfg = WsEndpointConfig::from_uri("ws://localhost:8080/echo").unwrap();
617 cfg.reconnect_policy = toml_policy.clone();
618 let cfg2 = WsEndpointConfig::from_uri("ws://localhost:8080/echo").unwrap();
621 let default_policy = WsEndpointConfig::default().reconnect_policy;
626 assert_eq!(cfg2.reconnect_policy, default_policy);
627 cfg.reconnect_policy = toml_policy.clone();
630 assert_eq!(cfg.reconnect_policy.max_attempts, 10);
631 assert_eq!(
632 cfg.reconnect_policy.initial_delay,
633 std::time::Duration::from_millis(250)
634 );
635 }
636
637 #[test]
640 fn ws_endpoint_policy_partial_bridge() {
641 let uri = "ws://localhost:9001/test?reconnectMaxAttempts=10";
642 let cfg = WsEndpointConfig::from_uri(uri).unwrap();
643 assert_eq!(cfg.reconnect_policy.max_attempts, 10);
645 assert!(cfg.reconnect_policy.enabled);
647 assert_eq!(
648 cfg.reconnect_policy.initial_delay,
649 std::time::Duration::from_millis(1000)
650 );
651 }
652
653 #[test]
656 fn network_retry_policy_from_toml() {
657 let toml_str = r#"
658 enabled = true
659 max_attempts = 10
660 initial_delay_ms = 250
661 multiplier = 3.0
662 max_delay_ms = 60000
663 jitter_factor = 0.1
664 "#;
665 let policy: NetworkRetryPolicy = toml::from_str(toml_str).unwrap();
666 assert!(policy.enabled);
667 assert_eq!(policy.max_attempts, 10);
668 assert_eq!(policy.initial_delay, std::time::Duration::from_millis(250));
669 assert!((policy.multiplier - 3.0).abs() < f64::EPSILON);
670 assert_eq!(policy.max_delay, std::time::Duration::from_millis(60_000));
671 assert!((policy.jitter_factor - 0.1).abs() < f64::EPSILON);
672 }
673}