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    pub(crate) metrics: Arc<dyn MetricsCollector>,
14    pub(crate) platform_service: Arc<dyn PlatformService>,
15    pub(crate) drain_timeout: Duration,
16    pub(crate) reconnect: NetworkRetryPolicy,
17    pub(crate) leadership_task: Option<JoinHandle<Result<(), CamelError>>>,
18    pub(crate) stop_token: Option<CancellationToken>,
19    pub(crate) runtime: Arc<dyn camel_component_api::RuntimeObservability>,
20}
21
22impl MasterConsumer {
23    #[allow(clippy::too_many_arguments)]
24    pub(crate) fn new(
25        lock_name: String,
26        delegate_uri: String,
27        delegate_component: Arc<dyn Component>,
28        metrics: Arc<dyn MetricsCollector>,
29        platform_service: Arc<dyn PlatformService>,
30        drain_timeout: Duration,
31        reconnect: NetworkRetryPolicy,
32        runtime: Arc<dyn camel_component_api::RuntimeObservability>,
33    ) -> Self {
34        Self {
35            lock_name,
36            delegate_uri,
37            delegate_component,
38            metrics,
39            platform_service,
40            drain_timeout,
41            reconnect,
42            leadership_task: None,
43            stop_token: None,
44            runtime,
45        }
46    }
47}
48
49pub(crate) enum DelegateState {
50    Inactive,
51    Active {
52        run_token: CancellationToken,
53        handle: JoinHandle<Result<(), CamelError>>,
54        /// Handle to the epoch-stamping bridge task. Joined by `stop_delegate`
55        /// within `drain_timeout`. Aborted on timeout to prevent detached
56        /// tasks from sending stale exchanges.
57        bridge_handle: Option<JoinHandle<()>>,
58        /// Leader epoch this delegate was reconciled at (the value the
59        /// epoch-stamping bridge carries). A `StartedLeading` delivery whose
60        /// published epoch equals this stamp is an idempotent no-op; a
61        /// differing epoch forces a full re-reconciliation. See ADR-0035.
62        epoch: u64,
63    },
64}
camel_master/consumer.rs

camel_master/
consumer.rs
