allow authentication against mqtt server

This commit is contained in:
Robin Appelman 2020-12-16 20:12:14 +01:00
commit cc82cc88c7
2 changed files with 39 additions and 12 deletions

View file

@ -1,5 +1,6 @@
use crate::device::BDAddr; use crate::device::BDAddr;
use color_eyre::{eyre::WrapErr, Report, Result}; use color_eyre::{eyre::WrapErr, Report, Result};
use rumqttc::MqttOptions;
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::str::FromStr; use std::str::FromStr;
@ -9,6 +10,12 @@ pub struct Config {
pub mqtt_port: u16, pub mqtt_port: u16,
pub host_port: u16, pub host_port: u16,
pub mi_temp_names: BTreeMap<BDAddr, String>, pub mi_temp_names: BTreeMap<BDAddr, String>,
pub mqtt_credentials: Option<Credentials>,
}
pub struct Credentials {
username: String,
password: String,
} }
impl Config { impl Config {
@ -40,11 +47,37 @@ impl Config {
}) })
.collect::<Result<BTreeMap<BDAddr, String>, Report>>()?; .collect::<Result<BTreeMap<BDAddr, String>, Report>>()?;
let mqtt_credentials = match dotenv::var("MQTT_USERNAME") {
Ok(username) => {
let password = dotenv::var("MQTT_PASSWORD")
.wrap_err("MQTT_USERNAME set, but MQTT_PASSWORD not set")?;
Some(Credentials { username, password })
}
Err(_) => None,
};
Ok(Config { Ok(Config {
mqtt_host, mqtt_host,
mqtt_port, mqtt_port,
host_port, host_port,
mi_temp_names, mi_temp_names,
mqtt_credentials,
}) })
} }
pub fn mqtt(&self) -> Result<MqttOptions> {
let hostname = hostname::get()?
.into_string()
.map_err(|_| Report::msg("invalid hostname"))?;
let mut mqtt_options = MqttOptions::new(
format!("taspromto-{}", hostname),
&self.mqtt_host,
self.mqtt_port,
);
if let Some(credentials) = self.mqtt_credentials.as_ref() {
mqtt_options.set_credentials(&credentials.username, &credentials.password);
}
mqtt_options.set_keep_alive(5);
Ok(mqtt_options)
}
} }

View file

@ -7,10 +7,10 @@ use crate::config::Config;
use crate::device::{format_device_state, Device, DeviceState}; use crate::device::{format_device_state, Device, DeviceState};
use crate::mqtt::mqtt_stream; use crate::mqtt::mqtt_stream;
use crate::topic::Topic; use crate::topic::Topic;
use color_eyre::{eyre::WrapErr, Report, Result}; use color_eyre::{eyre::WrapErr, Result};
use dashmap::DashMap; use dashmap::DashMap;
use pin_utils::pin_mut; use pin_utils::pin_mut;
use rumqttc::{MqttOptions, QoS}; use rumqttc::QoS;
use std::sync::Arc; use std::sync::Arc;
use tokio::stream::StreamExt; use tokio::stream::StreamExt;
use tokio::time::Duration; use tokio::time::Duration;
@ -31,11 +31,10 @@ async fn main() -> Result<()> {
.expect("Error setting Ctrl-C handler"); .expect("Error setting Ctrl-C handler");
let states = device_states.clone(); let states = device_states.clone();
let mqtt_host = config.mqtt_host; let mi_temp_names = config.mi_temp_names.clone();
let mqtt_port = config.mqtt_port;
tokio::task::spawn(async move { tokio::task::spawn(async move {
loop { loop {
if let Err(e) = mqtt_client(&mqtt_host, mqtt_port, states.clone()).await { if let Err(e) = mqtt_client(&config, states.clone()).await {
eprintln!("lost mqtt collection: {:#}", e); eprintln!("lost mqtt collection: {:#}", e);
} }
eprintln!("reconnecting after 1s"); eprintln!("reconnecting after 1s");
@ -45,7 +44,6 @@ async fn main() -> Result<()> {
let state = warp::any().map(move || device_states.clone()); let state = warp::any().map(move || device_states.clone());
let mi_temp_names = config.mi_temp_names;
let metrics = warp::path!("metrics") let metrics = warp::path!("metrics")
.and(state) .and(state)
.map(move |state: DeviceStates| { .map(move |state: DeviceStates| {
@ -66,12 +64,8 @@ async fn main() -> Result<()> {
Ok(()) Ok(())
} }
async fn mqtt_client(host: &str, port: u16, device_states: DeviceStates) -> Result<()> { async fn mqtt_client(config: &Config, device_states: DeviceStates) -> Result<()> {
let hostname = hostname::get()? let mqtt_options = config.mqtt()?;
.into_string()
.map_err(|_| Report::msg("invalid hostname"))?;
let mut mqtt_options = MqttOptions::new(format!("taspromto-{}", hostname), host, port);
mqtt_options.set_keep_alive(5);
let (client, stream) = mqtt_stream(mqtt_options) let (client, stream) = mqtt_stream(mqtt_options)
.await .await