camel_component_mqtt/
lib.rs

1pub mod client_id;
2pub mod config;
3pub mod consumer;
4pub mod endpoint;
5pub mod headers;
6pub mod producer;
7#[cfg(feature = "tls")]
8pub mod tls;
9pub mod uri;
10
11use std::sync::Arc;
12
13use camel_api::CamelError;
14use camel_component_api::{
15    BoxProcessor, Component, ComponentBundle, ComponentContext, ComponentRegistrar, Consumer,
16    Endpoint, ProducerContext, RuntimeObservability,
17};
18use config::{MqttBrokerConfig, MqttConfig, MqttEndpointConfig};
19use consumer::MqttConsumer;
20use producer::MqttProducer;
21use uri::parse_mqtt_uri;
22
23// ─── Component ─────────────────────────────────────────────────────────────
24
25pub struct MqttComponent {
26    config: MqttConfig,
27}
28
29impl MqttComponent {
30    pub fn new() -> Self {
31        Self {
32            config: MqttConfig::default(),
33        }
34    }
35
36    pub fn with_config(config: MqttConfig) -> Result<Self, CamelError> {
37        for (name, broker) in &config.brokers {
38            broker
39                .validate()
40                .map_err(|e| CamelError::Config(format!("mqtt broker '{name}': {e}")))?;
41        }
42        Ok(Self { config })
43    }
44
45    fn resolve_broker(&self, broker_name: &str) -> Result<MqttBrokerConfig, CamelError> {
46        self.config
47            .brokers
48            .get(broker_name)
49            .cloned()
50            .ok_or_else(|| {
51                CamelError::Config(format!(
52                    "mqtt: broker '{broker_name}' not found in Camel.toml [components.mqtt.brokers]"
53                ))
54            })
55    }
56}
57
58impl Default for MqttComponent {
59    fn default() -> Self {
60        Self::new()
61    }
62}
63
64impl Component for MqttComponent {
65    fn scheme(&self) -> &str {
66        "mqtt"
67    }
68
69    fn create_endpoint(
70        &self,
71        uri: &str,
72        _ctx: &dyn ComponentContext,
73    ) -> Result<Box<dyn Endpoint>, CamelError> {
74        let ep_config = parse_mqtt_uri(uri)?;
75        ep_config.validate()?;
76        let broker = self.resolve_broker(&ep_config.broker_name)?;
77        Ok(Box::new(MqttEndpoint {
78            uri: uri.to_string(),
79            config: ep_config,
80            broker,
81            client_id_prefix: self.config.client_id_prefix.clone(),
82            fallback_reconnect: self.config.reconnect.clone(),
83        }))
84    }
85}
86
87// ─── Endpoint ──────────────────────────────────────────────────────────────
88
89pub struct MqttEndpoint {
90    uri: String,
91    config: MqttEndpointConfig,
92    broker: MqttBrokerConfig,
93    client_id_prefix: String,
94    fallback_reconnect: camel_component_api::NetworkRetryPolicy,
95}
96
97impl Endpoint for MqttEndpoint {
98    fn uri(&self) -> &str {
99        &self.uri
100    }
101
102    fn create_consumer(
103        &self,
104        rt: Arc<dyn RuntimeObservability>,
105    ) -> Result<Box<dyn Consumer>, CamelError> {
106        consumer::validate_for_consumer(&self.config)?;
107        Ok(Box::new(MqttConsumer::new(
108            self.config.clone(),
109            rt,
110            self.broker.clone(),
111            self.client_id_prefix.clone(),
112            self.fallback_reconnect.clone(),
113        )))
114    }
115
116    fn create_producer(
117        &self,
118        _rt: Arc<dyn RuntimeObservability>,
119        ctx: &ProducerContext,
120    ) -> Result<BoxProcessor, CamelError> {
121        let producer = MqttProducer::new(
122            self.config.clone(),
123            self.broker.clone(),
124            &self.uri,
125            &self.client_id_prefix,
126            ctx.route_id(),
127            self.fallback_reconnect.clone(),
128        )?;
129        Ok(BoxProcessor::new(producer))
130    }
131}
132
133// ─── Bundle ────────────────────────────────────────────────────────────────
134
135pub struct MqttBundle {
136    config: MqttConfig,
137}
138
139impl ComponentBundle for MqttBundle {
140    fn config_key() -> &'static str {
141        "mqtt"
142    }
143
144    fn from_toml(value: toml::Value) -> Result<Self, CamelError> {
145        let config: MqttConfig = value
146            .try_into()
147            .map_err(|e: toml::de::Error| CamelError::Config(e.to_string()))?;
148        Ok(Self { config })
149    }
150
151    fn register_all(self, ctx: &mut dyn ComponentRegistrar) {
152        match MqttComponent::with_config(self.config) {
153            Ok(comp) => ctx.register_component_dyn(Arc::new(comp)),
154            Err(e) => {
155                // log-policy: system-broken
156                tracing::error!("MqttComponent registration failed: {e}");
157            }
158        }
159    }
160}
camel_component_mqtt/lib.rs

camel_component_mqtt/
lib.rs
