(conf: &Config)
| 46 | |
| 47 | impl<'a> Integration<'a> { |
| 48 | pub async fn new(conf: &Config) -> Result<Integration<'a>> { |
| 49 | info!("Initializing MQTT integration"); |
| 50 | |
| 51 | // topic templates |
| 52 | let mut templates = Handlebars::new(); |
| 53 | templates.register_escape_fn(handlebars::no_escape); |
| 54 | templates.register_template_string("event_topic", &conf.event_topic)?; |
| 55 | templates.register_template_string("command_topic", &conf.command_topic)?; |
| 56 | |
| 57 | let command_topic = templates.render( |
| 58 | "command_topic", |
| 59 | &CommandTopicContext { |
| 60 | application_id: "+".into(), |
| 61 | dev_eui: "+".into(), |
| 62 | command: "+".into(), |
| 63 | }, |
| 64 | )?; |
| 65 | |
| 66 | // get client id, this will generate a random client_id when no client_id has been |
| 67 | // configured. |
| 68 | let client_id = if conf.client_id.is_empty() { |
| 69 | let mut rnd = rand::rng(); |
| 70 | let client_id: u64 = rnd.random(); |
| 71 | format!("{:x}", client_id) |
| 72 | } else { |
| 73 | conf.client_id.clone() |
| 74 | }; |
| 75 | |
| 76 | // Get QoS |
| 77 | let qos = match conf.qos { |
| 78 | 0 => QoS::AtMostOnce, |
| 79 | 1 => QoS::AtLeastOnce, |
| 80 | 2 => QoS::ExactlyOnce, |
| 81 | _ => return Err(anyhow!("Invalid QoS: {}", conf.qos)), |
| 82 | }; |
| 83 | |
| 84 | // Create connect channel |
| 85 | // We need to re-subscribe on (re)connect to be sure we have a subscription. Even |
| 86 | // in case of a persistent MQTT session, there is no guarantee that the MQTT persisted the |
| 87 | // session and that a re-connect would recover the subscription. |
| 88 | let (connect_tx, mut connect_rx) = mpsc::channel(10); |
| 89 | |
| 90 | // Create client |
| 91 | let mut mqtt_opts = |
| 92 | MqttOptions::parse_url(format!("{}?client_id={}", conf.server, client_id))?; |
| 93 | mqtt_opts.set_clean_start(conf.clean_session); |
| 94 | mqtt_opts.set_keep_alive(conf.keep_alive_interval); |
| 95 | if !conf.username.is_empty() || !conf.password.is_empty() { |
| 96 | mqtt_opts.set_credentials(&conf.username, &conf.password); |
| 97 | } |
| 98 | |
| 99 | if !conf.ca_cert.is_empty() || !conf.tls_cert.is_empty() || !conf.tls_key.is_empty() { |
| 100 | info!( |
| 101 | "Configuring client with TLS certificate, ca_cert: {}, tls_cert: {}, tls_key: {}", |
| 102 | conf.ca_cert, conf.tls_cert, conf.tls_key |
| 103 | ); |
| 104 | |
| 105 | let root_certs = get_root_certs(if conf.ca_cert.is_empty() { |
nothing calls this directly
no test coverage detected