camel_component_jms/
component.rs

1use std::collections::HashMap;
2use std::future::Future;
3use std::path::PathBuf;
4use std::pin::Pin;
5use std::sync::Arc;
6use std::sync::atomic::{AtomicBool, Ordering};
7use std::task::{Context, Poll};
8use std::time::{Duration, Instant};
9
10use camel_bridge::{
11    download::ensure_binary,
12    health::wait_for_health,
13    process::{BridgeProcess, BridgeProcessConfig, BrokerType},
14};
15use camel_component_api::{
16    BoxProcessor, CamelError, Component, Consumer, Endpoint, Exchange, NetworkRetryPolicy,
17    ProducerContext,
18};
19use dashmap::DashMap;
20use tokio::sync::{Mutex, watch};
21use tonic::transport::Channel;
22use tower::Service;
23use tracing::{info, warn};
24
25use crate::config::{BrokerConfig, JmsEndpointConfig, JmsPoolConfig};
26use crate::consumer::JmsConsumer;
27use crate::health::JmsHealthCheck;
28use crate::producer::JmsProducer;
29use crate::proto::{HealthRequest, bridge_service_client::BridgeServiceClient};
30
31// ── Transport error classification ───────────────────────────────────────────
32
33/// Shared constant prefix for all bridge transport errors.
34///
35/// Both error-producing sites (producer.rs, consumer.rs) and the detection
36/// helper (`is_bridge_transport_error`) reference this constant so that a
37/// format-string drift cannot silently break retry logic.
38pub const BRIDGE_TRANSPORT_ERROR_PREFIX: &str = "JMS gRPC ";
39const MAX_RESTART_ATTEMPTS: u32 = 10;
40
41// ── BridgeState ──────────────────────────────────────────────────────────────
42
43#[derive(Debug, Clone)]
44pub enum BridgeState {
45    Starting,
46    Ready { channel: Channel },
47    Degraded(String),
48    Restarting { attempt: u32, next_at: Instant },
49    Stopped,
50}
51
52// ── BridgeSlot ───────────────────────────────────────────────────────────────
53
54pub struct BridgeSlot {
55    pub name: String,
56    pub broker_url: String,
57    pub broker_type: BrokerType,
58    pub credentials: Option<(String, String)>,
59    pub state_rx: watch::Receiver<BridgeState>,
60    pub(crate) state_tx: watch::Sender<BridgeState>,
61    /// BridgeProcess::stop(mut self) takes ownership — Mutex<Option<>> is required.
62    pub process: Arc<tokio::sync::Mutex<Option<BridgeProcess>>>,
63    /// JoinHandle of the health monitor task for this slot.
64    /// Stored so that shutdown can await the monitor and observe panics.
65    pub(crate) health_monitor_handle: Arc<tokio::sync::Mutex<Option<tokio::task::JoinHandle<()>>>>,
66}
67
68impl std::fmt::Debug for BridgeSlot {
69    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
70        f.debug_struct("BridgeSlot")
71            .field("name", &self.name)
72            .field("broker_url", &self.broker_url)
73            .field("broker_type", &self.broker_type)
74            .finish()
75    }
76}
77
78// ── JmsBridgePool ────────────────────────────────────────────────────────────
79
80pub struct JmsBridgePool {
81    pub(crate) slots: DashMap<String, Arc<BridgeSlot>>,
82    pub(crate) config: HashMap<String, BrokerConfig>,
83    pub(crate) bridge_start_timeout_ms: u64,
84    pub(crate) reconnect: NetworkRetryPolicy,
85    pub(crate) health_check_interval_ms: u64,
86    pub(crate) bridge_version: String,
87    pub(crate) bridge_cache_dir: PathBuf,
88    /// Maximum number of concurrently active (Starting/Ready) bridges.
89    pub(crate) max_bridges: usize,
90    /// Serializes bridge admission (check + insert) to prevent race on max_bridges.
91    bridge_create_lock: Mutex<()>,
92    pub(crate) shutting_down: Arc<AtomicBool>,
93}
94
95impl JmsBridgePool {
96    pub fn from_config(pool_config: JmsPoolConfig) -> Result<Self, CamelError> {
97        pool_config.validate()?;
98        // Backward compat: broker_reconnect_interval_ms overrides
99        // reconnect.initial_delay when explicitly set (non-default).
100        let mut reconnect = pool_config.reconnect;
101        if pool_config.broker_reconnect_interval_ms
102            != crate::config::default_broker_reconnect_interval_ms()
103        {
104            reconnect.initial_delay =
105                Duration::from_millis(pool_config.broker_reconnect_interval_ms);
106        }
107        Ok(Self {
108            slots: DashMap::new(),
109            config: pool_config.brokers,
110            bridge_start_timeout_ms: pool_config.bridge_start_timeout_ms,
111            reconnect,
112            health_check_interval_ms: pool_config.health_check_interval_ms,
113            bridge_version: crate::BRIDGE_VERSION.to_string(),
114            bridge_cache_dir: pool_config.bridge_cache_dir,
115            max_bridges: pool_config.max_bridges,
116            bridge_create_lock: Mutex::new(()),
117            shutting_down: Arc::new(AtomicBool::new(false)),
118        })
119    }
120
121    /// Resolve broker name from the URI `broker=` param.
122    ///
123    /// - If `Some(name)` → validate it exists in config and return it.
124    /// - If `None` and exactly one broker is configured → use it implicitly.
125    /// - If `None` and multiple brokers are configured → error asking for `?broker=`.
126    /// - If `None` and no brokers are configured → error asking to declare brokers.
127    pub fn resolve_broker_name(&self, name: Option<&str>) -> Result<String, CamelError> {
128        match name {
129            Some(n) => {
130                if self.config.contains_key(n) {
131                    Ok(n.to_string())
132                } else {
133                    Err(CamelError::ProcessorError(format!(
134                        "Unknown JMS broker '{n}' — declare it in [components.jms.brokers] in Camel.toml",
135                    )))
136                }
137            }
138            None => match self.config.len() {
139                0 => Err(CamelError::ProcessorError(
140                    "No JMS brokers configured — declare at least one in [components.jms.brokers] in Camel.toml".to_string(),
141                )),
142                1 => Ok(self.config.keys().next().unwrap().clone()), // allow-unwrap
143                _ => Err(CamelError::ProcessorError(format!(
144                    "Multiple JMS brokers configured ({}); specify one with ?broker=<name> in the URI",
145                    self.config.keys().cloned().collect::<Vec<_>>().join(", ")
146                ))),
147            },
148        }
149    }
150
151    /// Resolve broker type: activemq/artemis schemes hard-override config type; jms uses config.
152    pub fn resolve_broker_type(&self, scheme: &str, broker_name: &str) -> BrokerType {
153        let config_type = self
154            .config
155            .get(broker_name)
156            .map(|c| c.broker_type.clone())
157            .unwrap_or(BrokerType::Generic);
158
159        match scheme {
160            "activemq" => {
161                if config_type != BrokerType::ActiveMq && config_type != BrokerType::Generic {
162                    warn!(
163                        "Scheme 'activemq' overrides configured broker_type '{:?}' for broker '{}'",
164                        config_type, broker_name
165                    );
166                }
167                BrokerType::ActiveMq
168            }
169            "artemis" => {
170                if config_type != BrokerType::Artemis && config_type != BrokerType::Generic {
171                    warn!(
172                        "Scheme 'artemis' overrides configured broker_type '{:?}' for broker '{}'",
173                        config_type, broker_name
174                    );
175                }
176                BrokerType::Artemis
177            }
178            _ => config_type,
179        }
180    }
181
182    /// Get or create a BridgeSlot for the given broker name.
183    /// If the slot doesn't exist, starts the bridge process and spawns the health monitor.
184    pub async fn get_or_create_slot(
185        &self,
186        broker_name: &str,
187    ) -> Result<Arc<BridgeSlot>, CamelError> {
188        if let Some(slot) = self.slots.get(broker_name) {
189            return Ok(Arc::clone(&*slot));
190        }
191
192        // Serialize admission: check + insert in single critical section to prevent
193        // concurrent creators from both seeing count below limit and exceeding max_bridges.
194        let _guard = self.bridge_create_lock.lock().await;
195
196        // Re-check after acquiring lock (another caller may have inserted while we waited).
197        if let Some(slot) = self.slots.get(broker_name) {
198            return Ok(Arc::clone(&*slot));
199        }
200
201        // Enforce max_bridges: count ALL slots in the map under the admission lock.
202        // We count total slots (not just Starting/Ready) because any inserted slot
203        // represents an allocated bridge — even Degraded/Restarting slots hold resources
204        // and the bridge process may still be running. Counting only active states would
205        // allow a race: slot A transitions Starting→Degraded between two callers' checks,
206        // letting both pass the limit.
207        let total_count = self.slots.len();
208        if total_count >= self.max_bridges {
209            return Err(CamelError::Config(format!(
210                "JMS bridge limit reached: {total_count} bridge(s) >= max_bridges ({})",
211                self.max_bridges
212            )));
213        }
214
215        let broker_config = self.config.get(broker_name).ok_or_else(|| {
216            CamelError::ProcessorError(format!("Unknown JMS broker '{}'", broker_name))
217        })?;
218
219        // Clone all required broker data before touching DashMap::entry().
220        let broker_url = broker_config.broker_url.clone();
221        let broker_type = broker_config.broker_type.clone();
222        let credentials = match (&broker_config.username, &broker_config.password) {
223            (Some(u), Some(p)) => Some((u.clone(), p.clone())),
224            _ => None,
225        };
226
227        let slot = match self.slots.entry(broker_name.to_string()) {
228            dashmap::Entry::Occupied(existing) => {
229                return Ok(Arc::clone(existing.get()));
230            }
231            dashmap::Entry::Vacant(entry) => {
232                let (state_tx, state_rx) = watch::channel(BridgeState::Starting);
233                let slot = Arc::new(BridgeSlot {
234                    name: broker_name.to_string(),
235                    broker_url: broker_url.clone(),
236                    broker_type: broker_type.clone(),
237                    credentials: credentials.clone(),
238                    state_rx,
239                    state_tx,
240                    process: Arc::new(tokio::sync::Mutex::new(None)),
241                    health_monitor_handle: Arc::new(tokio::sync::Mutex::new(None)),
242                });
243                entry.insert(Arc::clone(&slot));
244                slot
245            }
246        };
247
248        let start_result = Self::start_bridge_process(
249            &self.bridge_version,
250            &self.bridge_cache_dir,
251            self.bridge_start_timeout_ms,
252            &broker_url,
253            &broker_type,
254            &credentials,
255        )
256        .await;
257
258        match start_result {
259            Ok((process, channel)) => {
260                {
261                    let mut guard = slot.process.lock().await;
262                    *guard = Some(process);
263                }
264                let _ = slot.state_tx.send(BridgeState::Ready { channel });
265            }
266            Err(e) => {
267                let _ = slot
268                    .state_tx
269                    .send(BridgeState::Degraded(format!("Initial start failed: {e}")));
270            }
271        }
272
273        self.spawn_health_monitor(Arc::clone(&slot)).await;
274
275        Ok(slot)
276    }
277
278    /// Signal a slot to restart (called by producers on transport errors).
279    pub fn restart_slot(&self, broker_name: &str) {
280        if let Some(slot) = self.slots.get(broker_name) {
281            let _ = slot.state_tx.send(BridgeState::Restarting {
282                attempt: 0,
283                next_at: Instant::now(),
284            });
285        }
286    }
287
288    /// Recreate the tonic channel for an existing running bridge process.
289    ///
290    /// Useful when a channel becomes stale after transport-level failures while
291    /// the underlying bridge process is still alive.
292    pub async fn refresh_slot_channel(&self, broker_name: &str) -> Result<(), CamelError> {
293        let slot = self
294            .slots
295            .get(broker_name)
296            .map(|s| Arc::clone(&*s))
297            .ok_or_else(|| {
298                CamelError::ProcessorError(format!("Unknown JMS broker '{}'", broker_name))
299            })?;
300
301        // Lock is held during connect() (TLS handshake + retries) — widened from
302        // the old grpc_port() read. Bounded: localhost TLS (~10ms typical), per-slot
303        // lock (only blocks same-broker operations). Acceptable trade-off vs exposing
304        // pub(crate) TLS material to component crates.
305        let channel = {
306            let guard = slot.process.lock().await;
307            let process = guard.as_ref().ok_or_else(|| {
308                CamelError::ProcessorError(format!(
309                    "JMS broker '{}' has no running bridge process",
310                    broker_name
311                ))
312            })?;
313            process.connect().await.map_err(|e| {
314                CamelError::ProcessorError(format!(
315                    "JMS broker '{}' channel refresh failed: {}",
316                    broker_name, e
317                ))
318            })?
319        };
320
321        let _ = slot.state_tx.send(BridgeState::Ready { channel });
322        Ok(())
323    }
324
325    pub fn begin_shutdown(&self) {
326        self.shutting_down.store(true, Ordering::SeqCst);
327    }
328
329    /// Shutdown all slots: stop all bridge processes and await health monitors.
330    pub async fn shutdown(&self) -> Result<(), CamelError> {
331        self.begin_shutdown();
332        let names: Vec<String> = self.slots.iter().map(|e| e.key().clone()).collect();
333        let mut errors: Vec<String> = Vec::new();
334
335        for name in names {
336            if let Some((_, slot)) = self.slots.remove(&name) {
337                // Signal the health monitor to stop.
338                let _ = slot.state_tx.send(BridgeState::Stopped);
339
340                // Stop the bridge process.
341                let process = {
342                    let mut guard = slot.process.lock().await;
343                    guard.take()
344                };
345                if let Some(p) = process
346                    && let Err(e) = p.stop().await
347                {
348                    errors.push(format!("broker '{}': process stop failed: {e}", slot.name));
349                }
350
351                // Await the health monitor task with timeout; abort if it doesn't stop.
352                let monitor_handle = {
353                    let mut guard = slot.health_monitor_handle.lock().await;
354                    guard.take()
355                };
356                if let Some(mut h) = monitor_handle
357                    && tokio::time::timeout(Duration::from_secs(5), &mut h)
358                        .await
359                        .is_err()
360                {
361                    h.abort();
362                    let _ = h.await;
363                    warn!(
364                        "health monitor for '{}' did not stop in 5s; aborted",
365                        slot.name
366                    );
367                }
368            }
369        }
370
371        if errors.is_empty() {
372            Ok(())
373        } else {
374            Err(CamelError::ProcessorError(format!(
375                "JMS pool shutdown completed with {} error(s): {}",
376                errors.len(),
377                errors.join("; ")
378            )))
379        }
380    }
381
382    async fn spawn_health_monitor(&self, slot: Arc<BridgeSlot>) {
383        let health_interval = self.health_check_interval_ms;
384        let bridge_version = self.bridge_version.clone();
385        let bridge_cache_dir = self.bridge_cache_dir.clone();
386        let start_timeout_ms = self.bridge_start_timeout_ms;
387        let handle_ref = Arc::clone(&slot.health_monitor_handle);
388        let shutting_down = Arc::clone(&self.shutting_down);
389        let broker_name = slot.name.clone();
390
391        let handle = tokio::spawn(async move {
392            loop {
393                let state = slot.state_rx.borrow().clone();
394                match state {
395                    BridgeState::Stopped => {
396                        info!("Health monitor for '{}' exiting (Stopped)", slot.name);
397                        break;
398                    }
399                    BridgeState::Ready { ref channel } => {
400                        tokio::time::sleep(Duration::from_millis(health_interval)).await;
401                        let mut client = BridgeServiceClient::new(channel.clone());
402                        let health_timeout = Duration::from_secs(3);
403                        match tokio::time::timeout(health_timeout, client.health(HealthRequest {}))
404                            .await
405                        {
406                            Ok(Ok(_)) => {}
407                            Ok(Err(e)) => {
408                                warn!(
409                                    "Health check failed for broker '{}': {e}. Marking Degraded.",
410                                    slot.name
411                                );
412                                let _ = slot.state_tx.send(BridgeState::Degraded(e.to_string()));
413                            }
414                            Err(_) => {
415                                let msg = format!(
416                                    "health RPC timed out after {}ms",
417                                    health_timeout.as_millis()
418                                );
419                                warn!(
420                                    "Health check timed out for broker '{}': {}. Marking Degraded.",
421                                    slot.name, msg
422                                );
423                                let _ = slot.state_tx.send(BridgeState::Degraded(msg));
424                            }
425                        }
426                    }
427                    BridgeState::Degraded(_) | BridgeState::Starting => {
428                        if matches!(*slot.state_rx.borrow(), BridgeState::Stopped) {
429                            break;
430                        }
431                        if shutting_down.load(Ordering::SeqCst) {
432                            tracing::info!(
433                                "Pool shutting down — not restarting bridge for broker '{}'",
434                                broker_name
435                            );
436                            break;
437                        }
438                        let _ = slot.state_tx.send(BridgeState::Restarting {
439                            attempt: 0,
440                            next_at: Instant::now(),
441                        });
442                    }
443                    BridgeState::Restarting { attempt, next_at } => {
444                        if shutting_down.load(Ordering::SeqCst) {
445                            tracing::info!(
446                                "Pool shutting down — aborting restart for broker '{}'",
447                                broker_name
448                            );
449                            break;
450                        }
451
452                        let now = Instant::now();
453                        if now < next_at {
454                            tokio::time::sleep(next_at - now).await;
455                        }
456
457                        info!(
458                            "Restarting bridge for broker '{}' (attempt {})",
459                            slot.name,
460                            attempt + 1
461                        );
462
463                        if attempt >= MAX_RESTART_ATTEMPTS {
464                            // log-policy: system-broken
465                            tracing::error!(
466                                "Max restart attempts ({}) reached for broker '{}' — staying degraded",
467                                attempt,
468                                broker_name
469                            );
470                            let _ = slot.state_tx.send(BridgeState::Degraded(format!(
471                                "max restart attempts ({}) exceeded",
472                                attempt
473                            )));
474                            break;
475                        }
476
477                        let old_process = {
478                            let mut guard = slot.process.lock().await;
479                            guard.take()
480                        };
481                        if let Some(p) = old_process {
482                            let _ = p.stop().await;
483                        }
484
485                        let start_result = Self::start_bridge_process(
486                            &bridge_version,
487                            &bridge_cache_dir,
488                            start_timeout_ms,
489                            &slot.broker_url,
490                            &slot.broker_type,
491                            &slot.credentials,
492                        )
493                        .await;
494
495                        match start_result {
496                            Ok((process, channel)) => {
497                                // Guard: don't resurrect a stopped slot (shutdown may have run
498                                // while this async bridge start was in-flight).
499                                if matches!(*slot.state_rx.borrow(), BridgeState::Stopped) {
500                                    let _ = process.stop().await;
501                                    break;
502                                }
503                                {
504                                    let mut guard = slot.process.lock().await;
505                                    *guard = Some(process);
506                                }
507                                let _ = slot.state_tx.send(BridgeState::Ready { channel });
508                                info!("Broker '{}' bridge restarted successfully", slot.name);
509                            }
510                            Err(e) => {
511                                // Guard: don't schedule retries after shutdown.
512                                if matches!(*slot.state_rx.borrow(), BridgeState::Stopped) {
513                                    break;
514                                }
515                                let delay_secs = std::cmp::min(5 * 2u64.pow(attempt), 120);
516                                let next = Instant::now() + Duration::from_secs(delay_secs);
517                                warn!(
518                                    "Failed to restart bridge for '{}' (attempt {}): {e}. Retry in {delay_secs}s",
519                                    slot.name,
520                                    attempt + 1
521                                );
522                                let _ = slot.state_tx.send(BridgeState::Restarting {
523                                    attempt: attempt + 1,
524                                    next_at: next,
525                                });
526                            }
527                        }
528                    }
529                }
530            }
531        });
532
533        // Store the handle so shutdown can await the monitor.
534        let mut guard = handle_ref.lock().await;
535        *guard = Some(handle);
536    }
537
538    async fn start_bridge_process(
539        bridge_version: &str,
540        bridge_cache_dir: &std::path::Path,
541        start_timeout_ms: u64,
542        broker_url: &str,
543        broker_type: &BrokerType,
544        credentials: &Option<(String, String)>,
545    ) -> Result<(BridgeProcess, Channel), CamelError> {
546        info!(
547            "Starting JMS bridge process for {}...",
548            redact_url(broker_url)
549        );
550        let binary_path = ensure_binary(bridge_version, bridge_cache_dir)
551            .await
552            .map_err(|e| {
553                CamelError::ProcessorError(format!("JMS bridge binary unavailable: {e}"))
554            })?;
555
556        let process_config = BridgeProcessConfig::jms(
557            binary_path,
558            broker_url.to_string(),
559            broker_type.clone(),
560            credentials.as_ref().map(|(u, _)| u.clone()),
561            credentials
562                .as_ref()
563                .map(|(_, p)| camel_bridge::process::Redacted::new(p.clone())),
564            start_timeout_ms,
565        );
566
567        let total_timeout = Duration::from_millis(start_timeout_ms);
568        let result = tokio::time::timeout(total_timeout, async {
569            let (process, channel) = BridgeProcess::start_and_connect(&process_config)
570                .await
571                .map_err(|e| CamelError::ProcessorError(format!("JMS bridge start failed: {e}")))?;
572
573            wait_for_health(&channel, Duration::from_secs(10), |ch| {
574                let mut client = BridgeServiceClient::new(ch);
575                async move {
576                    let resp = client.health(HealthRequest {}).await?;
577                    Ok(resp.into_inner().healthy)
578                }
579            })
580            .await
581            .map_err(|e| {
582                CamelError::ProcessorError(format!("JMS bridge health check failed: {e}"))
583            })?;
584
585            Ok::<(BridgeProcess, Channel), CamelError>((process, channel))
586        })
587        .await
588        .map_err(|_| {
589            CamelError::ProcessorError(format!(
590                "JMS bridge start timed out after {}ms",
591                start_timeout_ms
592            ))
593        })??;
594
595        Ok(result)
596    }
597}
598
599// ── JmsComponent ─────────────────────────────────────────────────────────────
600
601#[derive(Clone)]
602pub struct JmsComponent {
603    scheme: String,
604    pool: Arc<JmsBridgePool>,
605}
606
607impl JmsComponent {
608    pub fn with_scheme(scheme: impl Into<String>, pool: Arc<JmsBridgePool>) -> Self {
609        Self {
610            scheme: scheme.into(),
611            pool,
612        }
613    }
614
615    pub fn scheme(&self) -> &str {
616        &self.scheme
617    }
618
619    /// Test helper: send a message directly without going through a route.
620    #[cfg(test)]
621    pub async fn send_for_test(
622        &self,
623        destination: &str,
624        body: &[u8],
625        content_type: &str,
626    ) -> Result<String, CamelError> {
627        let broker_name = self.pool.resolve_broker_name(None)?;
628        let slot = self.pool.get_or_create_slot(&broker_name).await?;
629        let channel = match &*slot.state_rx.borrow() {
630            BridgeState::Ready { channel } => channel.clone(),
631            other => {
632                return Err(CamelError::ProcessorError(format!(
633                    "Bridge not ready: {:?}",
634                    other
635                )));
636            }
637        };
638        let mut client = BridgeServiceClient::new(channel);
639        let r = client
640            .send(crate::proto::SendRequest {
641                destination: destination.to_string(),
642                body: body.to_vec(),
643                headers: Default::default(),
644                content_type: content_type.to_string(),
645            })
646            .await
647            .map_err(|e| CamelError::ProcessorError(format!("test send error: {e}")))?;
648        Ok(r.into_inner().message_id)
649    }
650}
651
652impl Component for JmsComponent {
653    fn scheme(&self) -> &str {
654        &self.scheme
655    }
656
657    fn create_endpoint(
658        &self,
659        uri: &str,
660        ctx: &dyn camel_component_api::ComponentContext,
661    ) -> Result<Box<dyn Endpoint>, CamelError> {
662        let endpoint_config = JmsEndpointConfig::from_uri(uri)?;
663        let broker_name = self
664            .pool
665            .resolve_broker_name(endpoint_config.broker_name.as_deref())?;
666        let resolved_broker_type = self.pool.resolve_broker_type(&self.scheme, &broker_name);
667
668        let health_check = JmsHealthCheck::new(Arc::clone(&self.pool), broker_name.clone());
669        ctx.register_current_route_health_check(Arc::new(health_check));
670
671        Ok(Box::new(JmsEndpoint {
672            pool: Arc::clone(&self.pool),
673            uri: uri.to_string(),
674            broker_name,
675            resolved_broker_type,
676            endpoint_config,
677        }))
678    }
679}
680
681// ── JmsEndpoint ──────────────────────────────────────────────────────────────
682
683struct JmsEndpoint {
684    pool: Arc<JmsBridgePool>,
685    uri: String,
686    broker_name: String,
687    resolved_broker_type: BrokerType,
688    endpoint_config: JmsEndpointConfig,
689}
690
691impl Endpoint for JmsEndpoint {
692    fn uri(&self) -> &str {
693        &self.uri
694    }
695
696    fn create_producer(
697        &self,
698        rt: Arc<dyn camel_component_api::RuntimeObservability>,
699        _ctx: &ProducerContext,
700    ) -> Result<BoxProcessor, CamelError> {
701        Ok(BoxProcessor::new(LazyJmsProducer {
702            pool: Arc::clone(&self.pool),
703            broker_name: self.broker_name.clone(),
704            endpoint_config: self.endpoint_config.clone(),
705            resolved_broker_type: self.resolved_broker_type.clone(),
706            runtime: rt,
707        }))
708    }
709
710    fn create_consumer(
711        &self,
712        rt: Arc<dyn camel_component_api::RuntimeObservability>,
713    ) -> Result<Box<dyn Consumer>, CamelError> {
714        Ok(Box::new(JmsConsumer::new(
715            Arc::clone(&self.pool),
716            self.broker_name.clone(),
717            self.endpoint_config.clone(),
718            self.pool.reconnect.clone(),
719            rt,
720        )))
721    }
722}
723
724#[derive(Clone)]
725struct LazyJmsProducer {
726    pool: Arc<JmsBridgePool>,
727    broker_name: String,
728    endpoint_config: JmsEndpointConfig,
729    #[allow(dead_code)]
730    resolved_broker_type: BrokerType,
731    /// Phase B will use this for `rt.metrics().increment_errors(...)` and
732    /// `rt.health().force_unhealthy_for_route(...)` calls per ADR-0012.
733    #[allow(dead_code)]
734    runtime: Arc<dyn camel_component_api::RuntimeObservability>,
735}
736
737impl Service<Exchange> for LazyJmsProducer {
738    type Response = Exchange;
739    type Error = CamelError;
740    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
741
742    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
743        // Check existing slot state if one is already present.
744        // If no slot exists yet, return Ready — call() will handle async bridge start.
745        if let Some(slot) = self.pool.slots.get(&self.broker_name) {
746            match &*slot.state_rx.borrow() {
747                BridgeState::Ready { .. } => return Poll::Ready(Ok(())),
748                BridgeState::Starting | BridgeState::Restarting { .. } => {
749                    // Register a waker so the executor is notified when bridge state
750                    // changes. Without this, Poll::Pending would stall callers that use
751                    // strict Tower semantics (poll_ready loop before call()).
752                    // Guard with try_current: fall back to wake_by_ref when no Tokio
753                    // runtime is active (e.g. unit tests).
754                    let waker = cx.waker().clone();
755                    let mut rx = slot.state_rx.clone();
756                    if let Ok(handle) = tokio::runtime::Handle::try_current() {
757                        handle.spawn(async move {
758                            let _ = rx.changed().await;
759                            waker.wake();
760                        });
761                    } else {
762                        waker.wake_by_ref();
763                    }
764                    return Poll::Pending;
765                }
766                BridgeState::Degraded(reason) => {
767                    return Poll::Ready(Err(CamelError::ProcessorError(format!(
768                        "JMS broker '{}' is degraded: {}",
769                        self.broker_name, reason
770                    ))));
771                }
772                BridgeState::Stopped => {
773                    return Poll::Ready(Err(CamelError::ProcessorError(format!(
774                        "JMS broker '{}' is stopped",
775                        self.broker_name
776                    ))));
777                }
778            }
779        }
780        Poll::Ready(Ok(()))
781    }
782
783    fn call(&mut self, exchange: Exchange) -> Self::Future {
784        let pool = Arc::clone(&self.pool);
785        let broker_name = self.broker_name.clone();
786        let endpoint_config = self.endpoint_config.clone();
787
788        Box::pin(async move {
789            let slot = pool.get_or_create_slot(&broker_name).await?;
790            let mut rx = slot.state_rx.clone();
791
792            loop {
793                let state = rx.borrow().clone();
794                match state {
795                    BridgeState::Ready { channel } => {
796                        let mut producer = JmsProducer::new(channel, endpoint_config.clone());
797                        match producer.call(exchange).await {
798                            Ok(done) => return Ok(done),
799                            Err(first_err) if is_bridge_transport_error(&first_err) => {
800                                warn!(
801                                    broker = %broker_name,
802                                    error = %first_err,
803                                    "JMS send transport error; refreshing channel (no automatic resend)"
804                                );
805
806                                if let Err(refresh_err) =
807                                    pool.refresh_slot_channel(&broker_name).await
808                                {
809                                    warn!(
810                                        broker = %broker_name,
811                                        error = %refresh_err,
812                                        "JMS channel refresh failed; requesting bridge restart"
813                                    );
814                                    pool.restart_slot(&broker_name);
815                                }
816
817                                // Do NOT automatically resend — the first send may have reached
818                                // the broker even though the ack failed. Resending non-idempotent
819                                // writes causes duplicates. Return the original error so the caller
820                                // can decide whether to retry.
821                                return Err(first_err);
822                            }
823                            Err(other_err) => return Err(other_err),
824                        }
825                    }
826                    BridgeState::Degraded(reason) => {
827                        return Err(CamelError::ProcessorError(format!(
828                            "JMS broker '{}' is degraded: {}",
829                            broker_name, reason
830                        )));
831                    }
832                    BridgeState::Stopped => {
833                        return Err(CamelError::ProcessorError(format!(
834                            "JMS broker '{}' is stopped",
835                            broker_name
836                        )));
837                    }
838                    BridgeState::Starting | BridgeState::Restarting { .. } => {
839                        if rx.changed().await.is_err() {
840                            return Err(CamelError::ProcessorError(format!(
841                                "JMS broker '{}' state channel closed",
842                                broker_name
843                            )));
844                        }
845                    }
846                }
847            }
848        })
849    }
850}
851
852// ── Helpers ──────────────────────────────────────────────────────────────────
853
854/// Redact userinfo (username:password@) from a broker URL for safe logging.
855/// Handles URLs like `tcp://user:pass@host:61616` → `tcp://***@host:61616`.
856fn redact_url(url: &str) -> String {
857    // Find the scheme separator (://)
858    if let Some(pos) = url.find("://") {
859        let scheme = &url[..pos + 3]; // includes "://"
860        let rest = &url[pos + 3..];
861        // Find @ in the remainder — everything before @ is userinfo
862        if let Some(at_pos) = rest.find('@') {
863            return format!("{}***@{}", scheme, &rest[at_pos + 1..]);
864        }
865    }
866    url.to_string()
867}
868
869pub fn is_bridge_transport_error(err: &CamelError) -> bool {
870    // Typed variant matching: only ProcessorError messages that start with
871    // the well-known transport prefix are classified as transport errors.
872    // This rejects Config errors, business errors, and other CamelError variants
873    // without relying on the Display wrapper formatting.
874    match err {
875        CamelError::ProcessorError(msg) => msg.starts_with(BRIDGE_TRANSPORT_ERROR_PREFIX),
876        _ => false,
877    }
878}
879
880// ── Unit tests ───────────────────────────────────────────────────────────────
881
882#[cfg(test)]
883mod tests {
884    use camel_component_api::test_support::PanicRuntimeObservability;
885    fn test_rt() -> std::sync::Arc<dyn camel_component_api::RuntimeObservability> {
886        std::sync::Arc::new(PanicRuntimeObservability)
887    }
888    fn rt() -> std::sync::Arc<dyn camel_component_api::RuntimeObservability> {
889        std::sync::Arc::new(PanicRuntimeObservability)
890    }
891
892    use super::*;
893    use crate::config::{BrokerConfig, JmsPoolConfig};
894    use std::collections::HashMap;
895
896    #[test]
897    fn from_config_accepts_empty_brokers() {
898        let pool_config = JmsPoolConfig::default();
899        let result = JmsBridgePool::from_config(pool_config);
900        assert!(result.is_ok());
901    }
902
903    #[test]
904    fn resolve_broker_name_with_explicit_name() {
905        let pool = JmsBridgePool::from_config(JmsPoolConfig::single_broker(
906            "tcp://localhost:61616",
907            BrokerType::ActiveMq,
908        ))
909        .unwrap();
910        assert_eq!(
911            pool.resolve_broker_name(Some("default")).unwrap(),
912            "default"
913        );
914    }
915
916    #[test]
917    fn resolve_broker_name_default() {
918        let pool = JmsBridgePool::from_config(JmsPoolConfig::single_broker(
919            "tcp://localhost:61616",
920            BrokerType::ActiveMq,
921        ))
922        .unwrap();
923        assert_eq!(pool.resolve_broker_name(None).unwrap(), "default");
924    }
925
926    #[test]
927    fn resolve_broker_name_unknown_returns_error() {
928        let pool = JmsBridgePool::from_config(JmsPoolConfig::single_broker(
929            "tcp://localhost:61616",
930            BrokerType::ActiveMq,
931        ))
932        .unwrap();
933        let err = pool.resolve_broker_name(Some("unknown")).unwrap_err();
934        assert!(
935            err.to_string().contains("Unknown JMS broker 'unknown'"),
936            "got: {}",
937            err
938        );
939    }
940
941    #[test]
942    fn resolve_broker_type_scheme_overrides() {
943        let pool = JmsBridgePool::from_config(JmsPoolConfig::single_broker(
944            "tcp://localhost:61616",
945            BrokerType::Generic,
946        ))
947        .unwrap();
948        assert_eq!(
949            pool.resolve_broker_type("activemq", "default"),
950            BrokerType::ActiveMq
951        );
952        assert_eq!(
953            pool.resolve_broker_type("artemis", "default"),
954            BrokerType::Artemis
955        );
956        assert_eq!(
957            pool.resolve_broker_type("jms", "default"),
958            BrokerType::Generic
959        );
960    }
961
962    #[test]
963    fn resolve_broker_type_activemq_scheme_overrides_artemis_config() {
964        let pool = JmsBridgePool::from_config(JmsPoolConfig {
965            brokers: HashMap::from([(
966                "main".to_string(),
967                BrokerConfig {
968                    broker_url: "tcp://localhost:61616".to_string(),
969                    broker_type: BrokerType::Artemis,
970                    username: None,
971                    password: None,
972                },
973            )]),
974            ..JmsPoolConfig::default()
975        })
976        .unwrap();
977        assert_eq!(
978            pool.resolve_broker_type("activemq", "main"),
979            BrokerType::ActiveMq
980        );
981        assert_eq!(pool.resolve_broker_type("jms", "main"), BrokerType::Artemis);
982    }
983
984    #[test]
985    fn create_endpoint_resolves_broker() {
986        let pool = Arc::new(
987            JmsBridgePool::from_config(JmsPoolConfig::single_broker(
988                "tcp://localhost:61616",
989                BrokerType::ActiveMq,
990            ))
991            .unwrap(),
992        );
993        let component = JmsComponent::with_scheme("jms", pool);
994        let endpoint = component.create_endpoint(
995            "jms:queue:orders",
996            &camel_component_api::NoOpComponentContext,
997        );
998        assert!(endpoint.is_ok(), "got: {:?}", endpoint.err());
999    }
1000
1001    #[test]
1002    fn create_endpoint_rejects_wrong_scheme() {
1003        let pool = Arc::new(
1004            JmsBridgePool::from_config(JmsPoolConfig::single_broker(
1005                "tcp://localhost:61616",
1006                BrokerType::ActiveMq,
1007            ))
1008            .unwrap(),
1009        );
1010        let component = JmsComponent::with_scheme("jms", pool);
1011        let err = component
1012            .create_endpoint("kafka:orders", &camel_component_api::NoOpComponentContext)
1013            .err()
1014            .unwrap();
1015        assert!(
1016            err.to_string()
1017                .contains("expected scheme 'jms', 'activemq', or 'artemis'"),
1018            "got: {}",
1019            err
1020        );
1021    }
1022
1023    #[test]
1024    fn create_endpoint_with_explicit_broker_param() {
1025        let pool = Arc::new(
1026            JmsBridgePool::from_config(JmsPoolConfig {
1027                brokers: HashMap::from([
1028                    (
1029                        "primary".to_string(),
1030                        BrokerConfig {
1031                            broker_url: "tcp://primary:61616".to_string(),
1032                            broker_type: BrokerType::ActiveMq,
1033                            username: None,
1034                            password: None,
1035                        },
1036                    ),
1037                    (
1038                        "secondary".to_string(),
1039                        BrokerConfig {
1040                            broker_url: "tcp://secondary:61616".to_string(),
1041                            broker_type: BrokerType::Artemis,
1042                            username: None,
1043                            password: None,
1044                        },
1045                    ),
1046                ]),
1047                ..JmsPoolConfig::default()
1048            })
1049            .unwrap(),
1050        );
1051        let component = JmsComponent::with_scheme("jms", Arc::clone(&pool));
1052        let endpoint = component.create_endpoint(
1053            "jms:queue:orders?broker=secondary",
1054            &camel_component_api::NoOpComponentContext,
1055        );
1056        assert!(endpoint.is_ok(), "got: {:?}", endpoint.err());
1057    }
1058
1059    #[tokio::test]
1060    async fn concurrent_get_or_create_slot_no_deadlock() {
1061        use tokio::time::timeout;
1062
1063        struct EnvGuard {
1064            key: &'static str,
1065            prev: Option<std::ffi::OsString>,
1066        }
1067        impl Drop for EnvGuard {
1068            fn drop(&mut self) {
1069                if let Some(v) = &self.prev {
1070                    // SAFETY: restoring process env in test scope.
1071                    unsafe { std::env::set_var(self.key, v) };
1072                } else {
1073                    // SAFETY: restoring process env in test scope.
1074                    unsafe { std::env::remove_var(self.key) };
1075                }
1076            }
1077        }
1078
1079        let env_key = "CAMEL_JMS_BRIDGE_BINARY_PATH";
1080        let _guard = EnvGuard {
1081            key: env_key,
1082            prev: std::env::var_os(env_key),
1083        };
1084        // SAFETY: test-scoped env mutation.
1085        unsafe { std::env::set_var(env_key, "/bin/false") };
1086
1087        let pool = Arc::new(
1088            JmsBridgePool::from_config(JmsPoolConfig {
1089                brokers: HashMap::from([(
1090                    "test".to_string(),
1091                    BrokerConfig {
1092                        broker_url: "tcp://localhost:61616".to_string(),
1093                        broker_type: BrokerType::ActiveMq,
1094                        username: None,
1095                        password: None,
1096                    },
1097                )]),
1098                bridge_start_timeout_ms: 100,
1099                ..JmsPoolConfig::default()
1100            })
1101            .unwrap(),
1102        );
1103
1104        let handles: Vec<_> = (0..5)
1105            .map(|_| {
1106                let pool = Arc::clone(&pool);
1107                tokio::spawn(async move {
1108                    let _ = pool.get_or_create_slot("test").await;
1109                })
1110            })
1111            .collect();
1112
1113        let result = timeout(Duration::from_secs(5), async {
1114            for h in handles {
1115                let _ = h.await;
1116            }
1117        })
1118        .await;
1119
1120        assert!(result.is_ok(), "Concurrent get_or_create_slot deadlocked!");
1121    }
1122
1123    #[tokio::test]
1124    async fn lazy_producer_reports_degraded_when_bridge_start_fails() {
1125        use tower::Service;
1126
1127        struct EnvGuard {
1128            key: &'static str,
1129            prev: Option<std::ffi::OsString>,
1130        }
1131        impl Drop for EnvGuard {
1132            fn drop(&mut self) {
1133                if let Some(v) = &self.prev {
1134                    // SAFETY: restoring process env in test scope.
1135                    unsafe { std::env::set_var(self.key, v) };
1136                } else {
1137                    // SAFETY: restoring process env in test scope.
1138                    unsafe { std::env::remove_var(self.key) };
1139                }
1140            }
1141        }
1142
1143        let env_key = "CAMEL_JMS_BRIDGE_BINARY_PATH";
1144        let _guard = EnvGuard {
1145            key: env_key,
1146            prev: std::env::var_os(env_key),
1147        };
1148        // SAFETY: test-scoped env mutation.
1149        unsafe { std::env::set_var(env_key, "/bin/false") };
1150
1151        let pool = Arc::new(
1152            JmsBridgePool::from_config(JmsPoolConfig {
1153                brokers: HashMap::from([(
1154                    "default".to_string(),
1155                    BrokerConfig {
1156                        broker_url: "tcp://localhost:61616".to_string(),
1157                        broker_type: BrokerType::ActiveMq,
1158                        username: None,
1159                        password: None,
1160                    },
1161                )]),
1162                bridge_start_timeout_ms: 100,
1163                ..JmsPoolConfig::default()
1164            })
1165            .unwrap(),
1166        );
1167
1168        let component = JmsComponent::with_scheme("jms", pool);
1169        let endpoint = component
1170            .create_endpoint(
1171                "jms:queue:orders",
1172                &camel_component_api::NoOpComponentContext,
1173            )
1174            .unwrap();
1175        let mut producer = endpoint
1176            .create_producer(rt(), &camel_component_api::ProducerContext::default())
1177            .unwrap();
1178
1179        let mut exchange = Exchange::default();
1180        exchange.input.body = camel_component_api::Body::Text("hello".to_string());
1181
1182        let err = producer.call(exchange).await.unwrap_err();
1183        assert!(err.to_string().contains("is degraded"), "got: {}", err);
1184    }
1185
1186    /// A send transport error should trigger a channel refresh attempt first.
1187    /// If refresh cannot be performed (e.g. no running bridge process metadata),
1188    /// the producer requests a bridge restart as fallback.
1189    #[tokio::test]
1190    async fn lazy_producer_requests_restart_when_refresh_unavailable() {
1191        use tokio::sync::watch;
1192        use tonic::transport::Endpoint as TonicEndpoint;
1193        use tower::Service;
1194
1195        // Build a lazy channel to a port where nothing is listening.
1196        // connect_lazy() succeeds immediately; the error manifests on the actual RPC call.
1197        let dead_channel = TonicEndpoint::from_static("http://127.0.0.1:1").connect_lazy();
1198
1199        let (state_tx, state_rx) = watch::channel(BridgeState::Ready {
1200            channel: dead_channel.clone(),
1201        });
1202
1203        let pool = Arc::new(
1204            JmsBridgePool::from_config(JmsPoolConfig::single_broker(
1205                "tcp://localhost:61616",
1206                BrokerType::ActiveMq,
1207            ))
1208            .unwrap(),
1209        );
1210
1211        // Manually insert a slot with the dead-channel in Ready state.
1212        let slot = Arc::new(BridgeSlot {
1213            name: "default".to_string(),
1214            broker_url: "tcp://localhost:61616".to_string(),
1215            broker_type: BrokerType::ActiveMq,
1216            credentials: None,
1217            state_rx: state_rx.clone(),
1218            state_tx: state_tx.clone(),
1219            process: Arc::new(tokio::sync::Mutex::new(None)),
1220            health_monitor_handle: Arc::new(tokio::sync::Mutex::new(None)),
1221        });
1222        pool.slots.insert("default".to_string(), Arc::clone(&slot));
1223
1224        let endpoint_config =
1225            crate::config::JmsEndpointConfig::from_uri("jms:queue:test-retry").unwrap();
1226
1227        let mut producer = LazyJmsProducer {
1228            pool: Arc::clone(&pool),
1229            broker_name: "default".to_string(),
1230            endpoint_config,
1231            resolved_broker_type: BrokerType::ActiveMq,
1232            runtime: test_rt(),
1233        };
1234
1235        let mut exchange = Exchange::default();
1236        exchange.input.body = camel_component_api::Body::Text("hello".to_string());
1237
1238        // The send will fail because the channel points to a dead port.
1239        let result = producer.call(exchange).await;
1240        assert!(result.is_err(), "expected send to fail");
1241
1242        // Refresh cannot run in this setup (slot has no BridgeProcess), so the
1243        // fallback path requests a restart.
1244        let state_after = state_rx.borrow().clone();
1245        assert!(
1246            matches!(state_after, BridgeState::Restarting { .. }),
1247            "slot must enter Restarting when refresh is unavailable; got: {:?}",
1248            state_after
1249        );
1250    }
1251
1252    // ── JMS-007: Transport error classification ──────────────────────────────
1253
1254    #[test]
1255    fn transport_error_detects_send_error() {
1256        let err = CamelError::ProcessorError(format!(
1257            "{}send error: connection refused",
1258            BRIDGE_TRANSPORT_ERROR_PREFIX
1259        ));
1260        assert!(
1261            is_bridge_transport_error(&err),
1262            "send error must be classified as transport"
1263        );
1264    }
1265
1266    #[test]
1267    fn transport_error_detects_subscribe_error() {
1268        let err = CamelError::ProcessorError(format!(
1269            "{}subscribe error: stream reset",
1270            BRIDGE_TRANSPORT_ERROR_PREFIX
1271        ));
1272        assert!(
1273            is_bridge_transport_error(&err),
1274            "subscribe error must be classified as transport"
1275        );
1276    }
1277
1278    #[test]
1279    fn transport_error_rejects_business_errors() {
1280        let err = CamelError::ProcessorError("JMS broker 'main' is degraded: timeout".to_string());
1281        assert!(
1282            !is_bridge_transport_error(&err),
1283            "degraded state error must NOT be transport"
1284        );
1285    }
1286
1287    #[test]
1288    fn transport_error_rejects_config_errors() {
1289        let err = CamelError::Config("bridge_start_timeout_ms must be > 0".to_string());
1290        assert!(
1291            !is_bridge_transport_error(&err),
1292            "config error must NOT be transport"
1293        );
1294    }
1295
1296    #[test]
1297    fn transport_error_prefix_is_used_by_producer_and_consumer() {
1298        // Verify the constant prefix matches what producer.rs and consumer.rs emit.
1299        // If this test fails, the constant has drifted from the error format strings.
1300        assert!(
1301            BRIDGE_TRANSPORT_ERROR_PREFIX.starts_with("JMS gRPC "),
1302            "prefix must start with 'JMS gRPC '"
1303        );
1304    }
1305
1306    // ── JMS-006: max_bridges enforcement ─────────────────────────────────────
1307
1308    #[tokio::test]
1309    async fn pool_enforces_max_bridges_limit() {
1310        use tokio::sync::watch;
1311
1312        let pool = Arc::new(
1313            JmsBridgePool::from_config(JmsPoolConfig {
1314                brokers: HashMap::from([
1315                    (
1316                        "b1".to_string(),
1317                        BrokerConfig {
1318                            broker_url: "tcp://b1:61616".to_string(),
1319                            broker_type: BrokerType::ActiveMq,
1320                            username: None,
1321                            password: None,
1322                        },
1323                    ),
1324                    (
1325                        "b2".to_string(),
1326                        BrokerConfig {
1327                            broker_url: "tcp://b2:61616".to_string(),
1328                            broker_type: BrokerType::ActiveMq,
1329                            username: None,
1330                            password: None,
1331                        },
1332                    ),
1333                    (
1334                        "b3".to_string(),
1335                        BrokerConfig {
1336                            broker_url: "tcp://b3:61616".to_string(),
1337                            broker_type: BrokerType::ActiveMq,
1338                            username: None,
1339                            password: None,
1340                        },
1341                    ),
1342                ]),
1343                max_bridges: 2,
1344                ..JmsPoolConfig::default()
1345            })
1346            .unwrap(),
1347        );
1348
1349        // Manually insert two slots to simulate existing bridges.
1350        // max_bridges counts ALL slots in the map (not just active states).
1351        for name in &["b1", "b2"] {
1352            let (state_tx, state_rx) = watch::channel(BridgeState::Starting);
1353            let slot = Arc::new(BridgeSlot {
1354                name: name.to_string(),
1355                broker_url: format!("tcp://{name}:61616"),
1356                broker_type: BrokerType::ActiveMq,
1357                credentials: None,
1358                state_rx,
1359                state_tx,
1360                process: Arc::new(tokio::sync::Mutex::new(None)),
1361                health_monitor_handle: Arc::new(tokio::sync::Mutex::new(None)),
1362            });
1363            pool.slots.insert(name.to_string(), slot);
1364        }
1365
1366        // Attempting to create a third slot should fail.
1367        let err = pool.get_or_create_slot("b3").await.unwrap_err();
1368        assert!(
1369            err.to_string().contains("max_bridges"),
1370            "expected max_bridges error, got: {}",
1371            err
1372        );
1373    }
1374
1375    #[tokio::test]
1376    async fn pool_allows_slot_when_below_max_bridges() {
1377        use tokio::sync::watch;
1378
1379        let pool = Arc::new(
1380            JmsBridgePool::from_config(JmsPoolConfig {
1381                brokers: HashMap::from([(
1382                    "b1".to_string(),
1383                    BrokerConfig {
1384                        broker_url: "tcp://b1:61616".to_string(),
1385                        broker_type: BrokerType::ActiveMq,
1386                        username: None,
1387                        password: None,
1388                    },
1389                )]),
1390                max_bridges: 2,
1391                bridge_start_timeout_ms: 100,
1392                ..JmsPoolConfig::default()
1393            })
1394            .unwrap(),
1395        );
1396
1397        // Insert one slot in Degraded state (not counted as active).
1398        let (state_tx, state_rx) = watch::channel(BridgeState::Degraded("test".to_string()));
1399        let slot = Arc::new(BridgeSlot {
1400            name: "b1".to_string(),
1401            broker_url: "tcp://b1:61616".to_string(),
1402            broker_type: BrokerType::ActiveMq,
1403            credentials: None,
1404            state_rx,
1405            state_tx,
1406            process: Arc::new(tokio::sync::Mutex::new(None)),
1407            health_monitor_handle: Arc::new(tokio::sync::Mutex::new(None)),
1408        });
1409        pool.slots.insert("b1".to_string(), slot);
1410
1411        // b1 is Degraded (not active), so creating b1's slot returns existing.
1412        // The max_bridges check only applies to new slots.
1413        let result = pool.get_or_create_slot("b1").await;
1414        assert!(result.is_ok(), "existing slot must be returned");
1415    }
1416
1417    // ── JMS-003: poll_ready reflects bridge state ────────────────────────────
1418
1419    #[tokio::test]
1420    async fn poll_ready_returns_pending_when_starting() {
1421        use tokio::sync::watch;
1422        use tower::Service;
1423
1424        let pool = Arc::new(
1425            JmsBridgePool::from_config(JmsPoolConfig::single_broker(
1426                "tcp://localhost:61616",
1427                BrokerType::ActiveMq,
1428            ))
1429            .unwrap(),
1430        );
1431
1432        let (state_tx, state_rx) = watch::channel(BridgeState::Starting);
1433        let slot = Arc::new(BridgeSlot {
1434            name: "default".to_string(),
1435            broker_url: "tcp://localhost:61616".to_string(),
1436            broker_type: BrokerType::ActiveMq,
1437            credentials: None,
1438            state_rx,
1439            state_tx,
1440            process: Arc::new(tokio::sync::Mutex::new(None)),
1441            health_monitor_handle: Arc::new(tokio::sync::Mutex::new(None)),
1442        });
1443        pool.slots.insert("default".to_string(), slot);
1444
1445        let endpoint_config = crate::config::JmsEndpointConfig::from_uri("jms:queue:test").unwrap();
1446        let mut producer = LazyJmsProducer {
1447            pool: Arc::clone(&pool),
1448            broker_name: "default".to_string(),
1449            endpoint_config,
1450            resolved_broker_type: BrokerType::ActiveMq,
1451            runtime: test_rt(),
1452        };
1453
1454        let result = producer.poll_ready(&mut Context::from_waker(futures::task::noop_waker_ref()));
1455        assert!(
1456            matches!(result, Poll::Pending),
1457            "poll_ready must be Pending when Starting; got: {:?}",
1458            result
1459        );
1460    }
1461
1462    #[tokio::test]
1463    async fn poll_ready_returns_error_when_degraded() {
1464        use tokio::sync::watch;
1465        use tower::Service;
1466
1467        let pool = Arc::new(
1468            JmsBridgePool::from_config(JmsPoolConfig::single_broker(
1469                "tcp://localhost:61616",
1470                BrokerType::ActiveMq,
1471            ))
1472            .unwrap(),
1473        );
1474
1475        let (state_tx, state_rx) =
1476            watch::channel(BridgeState::Degraded("health check failed".to_string()));
1477        let slot = Arc::new(BridgeSlot {
1478            name: "default".to_string(),
1479            broker_url: "tcp://localhost:61616".to_string(),
1480            broker_type: BrokerType::ActiveMq,
1481            credentials: None,
1482            state_rx,
1483            state_tx,
1484            process: Arc::new(tokio::sync::Mutex::new(None)),
1485            health_monitor_handle: Arc::new(tokio::sync::Mutex::new(None)),
1486        });
1487        pool.slots.insert("default".to_string(), slot);
1488
1489        let endpoint_config = crate::config::JmsEndpointConfig::from_uri("jms:queue:test").unwrap();
1490        let mut producer = LazyJmsProducer {
1491            pool: Arc::clone(&pool),
1492            broker_name: "default".to_string(),
1493            endpoint_config,
1494            resolved_broker_type: BrokerType::ActiveMq,
1495            runtime: test_rt(),
1496        };
1497
1498        let result = producer.poll_ready(&mut Context::from_waker(futures::task::noop_waker_ref()));
1499        assert!(
1500            matches!(result, Poll::Ready(Err(_))),
1501            "poll_ready must be Err when Degraded; got: {:?}",
1502            result
1503        );
1504        let err_msg = match result {
1505            Poll::Ready(Err(e)) => e.to_string(),
1506            _ => unreachable!(),
1507        };
1508        assert!(
1509            err_msg.contains("degraded"),
1510            "error must mention degraded: {}",
1511            err_msg
1512        );
1513    }
1514
1515    #[tokio::test]
1516    async fn poll_ready_returns_error_when_stopped() {
1517        use tokio::sync::watch;
1518        use tower::Service;
1519
1520        let pool = Arc::new(
1521            JmsBridgePool::from_config(JmsPoolConfig::single_broker(
1522                "tcp://localhost:61616",
1523                BrokerType::ActiveMq,
1524            ))
1525            .unwrap(),
1526        );
1527
1528        let (state_tx, state_rx) = watch::channel(BridgeState::Stopped);
1529        let slot = Arc::new(BridgeSlot {
1530            name: "default".to_string(),
1531            broker_url: "tcp://localhost:61616".to_string(),
1532            broker_type: BrokerType::ActiveMq,
1533            credentials: None,
1534            state_rx,
1535            state_tx,
1536            process: Arc::new(tokio::sync::Mutex::new(None)),
1537            health_monitor_handle: Arc::new(tokio::sync::Mutex::new(None)),
1538        });
1539        pool.slots.insert("default".to_string(), slot);
1540
1541        let endpoint_config = crate::config::JmsEndpointConfig::from_uri("jms:queue:test").unwrap();
1542        let mut producer = LazyJmsProducer {
1543            pool: Arc::clone(&pool),
1544            broker_name: "default".to_string(),
1545            endpoint_config,
1546            resolved_broker_type: BrokerType::ActiveMq,
1547            runtime: test_rt(),
1548        };
1549
1550        let result = producer.poll_ready(&mut Context::from_waker(futures::task::noop_waker_ref()));
1551        assert!(
1552            matches!(result, Poll::Ready(Err(_))),
1553            "poll_ready must be Err when Stopped; got: {:?}",
1554            result
1555        );
1556    }
1557
1558    #[tokio::test]
1559    async fn poll_ready_returns_ready_when_slot_ready() {
1560        use tokio::sync::watch;
1561        use tonic::transport::Endpoint as TonicEndpoint;
1562        use tower::Service;
1563
1564        let pool = Arc::new(
1565            JmsBridgePool::from_config(JmsPoolConfig::single_broker(
1566                "tcp://localhost:61616",
1567                BrokerType::ActiveMq,
1568            ))
1569            .unwrap(),
1570        );
1571
1572        let lazy_channel = TonicEndpoint::from_static("http://127.0.0.1:1").connect_lazy();
1573        let (state_tx, state_rx) = watch::channel(BridgeState::Ready {
1574            channel: lazy_channel,
1575        });
1576        let slot = Arc::new(BridgeSlot {
1577            name: "default".to_string(),
1578            broker_url: "tcp://localhost:61616".to_string(),
1579            broker_type: BrokerType::ActiveMq,
1580            credentials: None,
1581            state_rx,
1582            state_tx,
1583            process: Arc::new(tokio::sync::Mutex::new(None)),
1584            health_monitor_handle: Arc::new(tokio::sync::Mutex::new(None)),
1585        });
1586        pool.slots.insert("default".to_string(), slot);
1587
1588        let endpoint_config = crate::config::JmsEndpointConfig::from_uri("jms:queue:test").unwrap();
1589        let mut producer = LazyJmsProducer {
1590            pool: Arc::clone(&pool),
1591            broker_name: "default".to_string(),
1592            endpoint_config,
1593            resolved_broker_type: BrokerType::ActiveMq,
1594            runtime: test_rt(),
1595        };
1596
1597        let result = producer.poll_ready(&mut Context::from_waker(futures::task::noop_waker_ref()));
1598        assert!(
1599            matches!(result, Poll::Ready(Ok(()))),
1600            "poll_ready must be Ready(Ok) when bridge is Ready; got: {:?}",
1601            result
1602        );
1603    }
1604
1605    #[tokio::test]
1606    async fn poll_ready_returns_ready_when_no_slot_exists() {
1607        use tower::Service;
1608
1609        let pool = Arc::new(
1610            JmsBridgePool::from_config(JmsPoolConfig::single_broker(
1611                "tcp://localhost:61616",
1612                BrokerType::ActiveMq,
1613            ))
1614            .unwrap(),
1615        );
1616
1617        let endpoint_config = crate::config::JmsEndpointConfig::from_uri("jms:queue:test").unwrap();
1618        let mut producer = LazyJmsProducer {
1619            pool: Arc::clone(&pool),
1620            broker_name: "default".to_string(),
1621            endpoint_config,
1622            resolved_broker_type: BrokerType::ActiveMq,
1623            runtime: test_rt(),
1624        };
1625
1626        // No slot exists yet — poll_ready should return Ready so call() can start the bridge.
1627        let result = producer.poll_ready(&mut Context::from_waker(futures::task::noop_waker_ref()));
1628        assert!(
1629            matches!(result, Poll::Ready(Ok(()))),
1630            "poll_ready must be Ready(Ok) when no slot exists; got: {:?}",
1631            result
1632        );
1633    }
1634
1635    // ── JMS-001: Health monitor lifecycle ────────────────────────────────────
1636
1637    #[tokio::test]
1638    async fn pool_shutdown_awaits_health_monitor() {
1639        use tokio::sync::watch;
1640
1641        let pool = Arc::new(
1642            JmsBridgePool::from_config(JmsPoolConfig {
1643                brokers: HashMap::from([(
1644                    "default".to_string(),
1645                    BrokerConfig {
1646                        broker_url: "tcp://localhost:61616".to_string(),
1647                        broker_type: BrokerType::ActiveMq,
1648                        username: None,
1649                        password: None,
1650                    },
1651                )]),
1652                health_check_interval_ms: 100,
1653                ..JmsPoolConfig::default()
1654            })
1655            .unwrap(),
1656        );
1657
1658        // Create a slot manually with a spawned health monitor task.
1659        let (state_tx, state_rx) = watch::channel(BridgeState::Ready {
1660            channel: tonic::transport::Endpoint::from_static("http://127.0.0.1:1").connect_lazy(),
1661        });
1662        let monitor_handle_ref: Arc<tokio::sync::Mutex<Option<tokio::task::JoinHandle<()>>>> =
1663            Arc::new(tokio::sync::Mutex::new(None));
1664
1665        // Spawn a simple monitor that exits on Stopped.
1666        let state_rx_clone = state_rx.clone();
1667        let handle = tokio::spawn(async move {
1668            loop {
1669                if matches!(*state_rx_clone.borrow(), BridgeState::Stopped) {
1670                    break;
1671                }
1672                tokio::time::sleep(Duration::from_millis(50)).await;
1673            }
1674        });
1675        *monitor_handle_ref.lock().await = Some(handle);
1676
1677        let slot = Arc::new(BridgeSlot {
1678            name: "default".to_string(),
1679            broker_url: "tcp://localhost:61616".to_string(),
1680            broker_type: BrokerType::ActiveMq,
1681            credentials: None,
1682            state_rx,
1683            state_tx,
1684            process: Arc::new(tokio::sync::Mutex::new(None)),
1685            health_monitor_handle: monitor_handle_ref,
1686        });
1687        pool.slots.insert("default".to_string(), slot);
1688
1689        // Shutdown should complete without hanging — the monitor exits on Stopped.
1690        let result = pool.shutdown().await;
1691        // May report errors from bridge process (none in this test), but must not hang.
1692        let _ = result;
1693    }
1694
1695    #[tokio::test]
1696    async fn health_monitor_handle_stored_after_spawn() {
1697        use tokio::sync::watch;
1698
1699        let pool = Arc::new(
1700            JmsBridgePool::from_config(JmsPoolConfig {
1701                brokers: HashMap::from([(
1702                    "default".to_string(),
1703                    BrokerConfig {
1704                        broker_url: "tcp://localhost:61616".to_string(),
1705                        broker_type: BrokerType::ActiveMq,
1706                        username: None,
1707                        password: None,
1708                    },
1709                )]),
1710                health_check_interval_ms: 100,
1711                bridge_start_timeout_ms: 100,
1712                ..JmsPoolConfig::default()
1713            })
1714            .unwrap(),
1715        );
1716
1717        let (state_tx, state_rx) = watch::channel(BridgeState::Stopped);
1718        let slot = Arc::new(BridgeSlot {
1719            name: "default".to_string(),
1720            broker_url: "tcp://localhost:61616".to_string(),
1721            broker_type: BrokerType::ActiveMq,
1722            credentials: None,
1723            state_rx,
1724            state_tx,
1725            process: Arc::new(tokio::sync::Mutex::new(None)),
1726            health_monitor_handle: Arc::new(tokio::sync::Mutex::new(None)),
1727        });
1728        pool.slots.insert("default".to_string(), Arc::clone(&slot));
1729
1730        pool.spawn_health_monitor(Arc::clone(&slot)).await;
1731
1732        tokio::time::sleep(Duration::from_millis(50)).await;
1733        let guard = slot.health_monitor_handle.lock().await;
1734        assert!(
1735            guard.is_some(),
1736            "health monitor handle must be stored after spawn_health_monitor"
1737        );
1738    }
1739
1740    // ── JMS-008: URL redaction for safe logging ──────────────────────────────
1741
1742    #[test]
1743    fn redact_url_strips_userinfo_with_password() {
1744        assert_eq!(
1745            redact_url("tcp://admin:s3cret@broker:61616"),
1746            "tcp://***@broker:61616"
1747        );
1748    }
1749
1750    #[test]
1751    fn redact_url_strips_userinfo_without_password() {
1752        assert_eq!(
1753            redact_url("tcp://admin@broker:61616"),
1754            "tcp://***@broker:61616"
1755        );
1756    }
1757
1758    #[test]
1759    fn redact_url_passes_clean_url_unchanged() {
1760        assert_eq!(redact_url("tcp://localhost:61616"), "tcp://localhost:61616");
1761    }
1762
1763    #[test]
1764    fn redact_url_handles_ssl_scheme() {
1765        assert_eq!(
1766            redact_url("ssl://user:pass@secure-broker:61617"),
1767            "ssl://***@secure-broker:61617"
1768        );
1769    }
1770
1771    // ── JMS-009: max_bridges race condition under concurrency ────────────────
1772
1773    #[tokio::test]
1774    async fn concurrent_slot_creation_respects_max_bridges() {
1775        let pool = Arc::new(
1776            JmsBridgePool::from_config(JmsPoolConfig {
1777                brokers: HashMap::from([
1778                    (
1779                        "b1".to_string(),
1780                        BrokerConfig {
1781                            broker_url: "tcp://b1:61616".to_string(),
1782                            broker_type: BrokerType::ActiveMq,
1783                            username: None,
1784                            password: None,
1785                        },
1786                    ),
1787                    (
1788                        "b2".to_string(),
1789                        BrokerConfig {
1790                            broker_url: "tcp://b2:61616".to_string(),
1791                            broker_type: BrokerType::ActiveMq,
1792                            username: None,
1793                            password: None,
1794                        },
1795                    ),
1796                    (
1797                        "b3".to_string(),
1798                        BrokerConfig {
1799                            broker_url: "tcp://b3:61616".to_string(),
1800                            broker_type: BrokerType::ActiveMq,
1801                            username: None,
1802                            password: None,
1803                        },
1804                    ),
1805                ]),
1806                max_bridges: 2,
1807                bridge_start_timeout_ms: 100,
1808                ..JmsPoolConfig::default()
1809            })
1810            .unwrap(),
1811        );
1812
1813        let (state_tx, state_rx) = watch::channel(BridgeState::Starting);
1814        for name in &["b1", "b2"] {
1815            let slot = Arc::new(BridgeSlot {
1816                name: name.to_string(),
1817                broker_url: format!("tcp://{name}:61616"),
1818                broker_type: BrokerType::ActiveMq,
1819                credentials: None,
1820                state_rx: state_rx.clone(),
1821                state_tx: state_tx.clone(),
1822                process: Arc::new(tokio::sync::Mutex::new(None)),
1823                health_monitor_handle: Arc::new(tokio::sync::Mutex::new(None)),
1824            });
1825            pool.slots.insert(name.to_string(), slot);
1826        }
1827
1828        assert_eq!(pool.slots.len(), 2);
1829
1830        let guard = pool.bridge_create_lock.lock().await;
1831        let total_count = pool.slots.len();
1832        assert!(total_count >= pool.max_bridges);
1833        let result = if total_count >= pool.max_bridges {
1834            Err(CamelError::Config(format!(
1835                "JMS bridge limit reached: {total_count} bridge(s) >= max_bridges ({})",
1836                pool.max_bridges
1837            )))
1838        } else {
1839            Ok(())
1840        };
1841        drop(guard);
1842
1843        assert!(result.is_err(), "3rd broker should be rejected");
1844        let err_msg = result.unwrap_err().to_string();
1845        assert!(
1846            err_msg.contains("max_bridges"),
1847            "error must mention max_bridges, got: {err_msg}"
1848        );
1849    }
1850
1851    // ── JMS-010: Transport error does NOT auto-resend ────────────────────────
1852
1853    #[tokio::test]
1854    async fn transport_error_refreshes_channel_but_does_not_resend() {
1855        use tokio::sync::watch;
1856        use tonic::transport::Endpoint as TonicEndpoint;
1857        use tower::Service;
1858
1859        let dead_channel = TonicEndpoint::from_static("http://127.0.0.1:1").connect_lazy();
1860
1861        let (state_tx, state_rx) = watch::channel(BridgeState::Ready {
1862            channel: dead_channel.clone(),
1863        });
1864
1865        let pool = Arc::new(
1866            JmsBridgePool::from_config(JmsPoolConfig::single_broker(
1867                "tcp://localhost:61616",
1868                BrokerType::ActiveMq,
1869            ))
1870            .unwrap(),
1871        );
1872
1873        let slot = Arc::new(BridgeSlot {
1874            name: "default".to_string(),
1875            broker_url: "tcp://localhost:61616".to_string(),
1876            broker_type: BrokerType::ActiveMq,
1877            credentials: None,
1878            state_rx: state_rx.clone(),
1879            state_tx: state_tx.clone(),
1880            process: Arc::new(tokio::sync::Mutex::new(None)),
1881            health_monitor_handle: Arc::new(tokio::sync::Mutex::new(None)),
1882        });
1883        pool.slots.insert("default".to_string(), Arc::clone(&slot));
1884
1885        let endpoint_config =
1886            crate::config::JmsEndpointConfig::from_uri("jms:queue:test-no-resend").unwrap();
1887
1888        let mut producer = LazyJmsProducer {
1889            pool: Arc::clone(&pool),
1890            broker_name: "default".to_string(),
1891            endpoint_config,
1892            resolved_broker_type: BrokerType::ActiveMq,
1893            runtime: test_rt(),
1894        };
1895
1896        let mut exchange = Exchange::default();
1897        exchange.input.body = camel_component_api::Body::Text("hello".to_string());
1898
1899        let result = producer.call(exchange).await;
1900        assert!(result.is_err(), "expected send to fail");
1901
1902        let state_after = state_rx.borrow().clone();
1903        assert!(
1904            matches!(state_after, BridgeState::Restarting { .. }),
1905            "slot must enter Restarting; got: {:?}",
1906            state_after
1907        );
1908
1909        let err_msg = result.unwrap_err().to_string();
1910        assert!(
1911            err_msg.contains(BRIDGE_TRANSPORT_ERROR_PREFIX),
1912            "error must be original transport error, got: {}",
1913            err_msg
1914        );
1915    }
1916}
camel_component_jms/component.rs

camel_component_jms/
component.rs
