camel_master/
consumer.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use camel_api::{CamelError, MetricsCollector, PlatformService};
5use camel_component_api::{Component, NetworkRetryPolicy};
6use tokio::task::JoinHandle;
7use tokio_util::sync::CancellationToken;
8
9pub(crate) struct MasterConsumer {
10    pub(crate) lock_name: String,
11    pub(crate) delegate_uri: String,
12    pub(crate) delegate_component: Arc<dyn Component>,
13    // TODO(MST-001): MetricsCollector is stored here but never used for emission.
14    // Wire into reconcile_event to record leadership transitions and delegate lifecycle.
15    pub(crate) metrics: Arc<dyn MetricsCollector>,
16    pub(crate) platform_service: Arc<dyn PlatformService>,
17    pub(crate) drain_timeout: Duration,
18    pub(crate) reconnect: NetworkRetryPolicy,
19    pub(crate) leadership_task: Option<JoinHandle<Result<(), CamelError>>>,
20    pub(crate) stop_token: Option<CancellationToken>,
21    pub(crate) runtime: Arc<dyn camel_component_api::RuntimeObservability>,
22}
23
24impl MasterConsumer {
25    #[allow(clippy::too_many_arguments)]
26    pub(crate) fn new(
27        lock_name: String,
28        delegate_uri: String,
29        delegate_component: Arc<dyn Component>,
30        metrics: Arc<dyn MetricsCollector>,
31        platform_service: Arc<dyn PlatformService>,
32        drain_timeout: Duration,
33        reconnect: NetworkRetryPolicy,
34        runtime: Arc<dyn camel_component_api::RuntimeObservability>,
35    ) -> Self {
36        Self {
37            lock_name,
38            delegate_uri,
39            delegate_component,
40            metrics,
41            platform_service,
42            drain_timeout,
43            reconnect,
44            leadership_task: None,
45            stop_token: None,
46            runtime,
47        }
48    }
49}
50
51pub(crate) enum DelegateState {
52    Inactive,
53    Active {
54        run_token: CancellationToken,
55        handle: JoinHandle<Result<(), CamelError>>,
56        /// Handle to the epoch-stamping bridge task. Joined by `stop_delegate`
57        /// within `drain_timeout`. Aborted on timeout to prevent detached
58        /// tasks from sending stale exchanges.
59        bridge_handle: Option<JoinHandle<()>>,
60    },
61}
camel_master/consumer.rs

camel_master/
consumer.rs
