base-station/software: Reimplement MQTT support and integrate with tokio.

base-station/software/Cargo.lock

@@ -790,7 +790,7 @@ dependencies = [
  "indexmap",
  "slab",
  "tokio 0.2.20",
- "tokio-util",
+ "tokio-util 0.3.1",
  "tracing",
  "tracing-futures",
 ]
@@ -1314,10 +1314,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "28e75553120607736d0a248f483fb0682e0279bb9a6f30a7c76c56a9a6ac9ade"
 dependencies = [
  "byteorder",
+ "bytes 1.0.1",
  "lazy_static",
  "log",
  "regex",
  "thiserror",
+ "tokio 1.10.1",
+ "tokio-util 0.6.7",
 ]
 
 [[package]]
@@ -1392,7 +1395,7 @@ dependencies = [
  "thiserror",
  "tokio 0.2.20",
  "tokio-serial",
- "tokio-util",
+ "tokio-util 0.3.1",
 ]
 
 [[package]]
@@ -2444,6 +2447,20 @@ dependencies = [
  "tokio 0.2.20",
 ]
 
+[[package]]
+name = "tokio-util"
+version = "0.6.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1caa0b0c8d94a049db56b5acf8cba99dc0623aab1b26d5b5f5e2d945846b3592"
+dependencies = [
+ "bytes 1.0.1",
+ "futures-core",
+ "futures-sink",
+ "log",
+ "pin-project-lite 0.2.4",
+ "tokio 1.10.1",
+]
+
 [[package]]
 name = "toml"
 version = "0.5.8"

base-station/software/Cargo.toml

@@ -15,7 +15,7 @@ nrf24l01-stick = [ "nrf24l01-stick-driver" ]
 linux-embedded-hal = "0.3"
 embedded-nrf24l01 = { git = "https://github.com/mgottschlag/embedded-nrf24l01", branch = "wip" }
 embedded-hal = "0.2.3"
-mqtt-protocol = "0.11"
+mqtt-protocol = { version = "0.11", features = [ "tokio-codec" ] }
 env_logger = "0.9"
 log = "0.4.14"
 thiserror = "1.0"

base-station/software/default_config.toml

@@ -3,3 +3,7 @@ enabled = false
 # address = "http://localhost:8086"
 # user = ""
 # password = ""
+
+[mqtt]
+enabled = false
+# address = "http://localhost:1883"

base-station/software/src/config.rs

@@ -6,6 +6,7 @@ use serde::Deserialize;
 #[derive(Deserialize)]
 pub struct Config {
     pub influxdb: InfluxDbConfig,
+    pub mqtt: MQTTConfig,
 }
 
 impl Config {
@@ -18,6 +19,7 @@ impl Config {
 
     fn validate(&self) -> Result<(), ConfigError> {
         self.influxdb.validate()?;
+        self.mqtt.validate()?;
         Ok(())
     }
 }
@@ -56,6 +58,29 @@ impl Default for InfluxDbConfig {
     }
 }
 
+#[derive(Clone, Deserialize)]
+pub struct MQTTConfig {
+    #[serde(default)]
+    pub enabled: bool,
+    #[serde(default)]
+    pub address: String,
+}
+
+impl MQTTConfig {
+    fn validate(&self) -> Result<(), ConfigError> {
+        Ok(())
+    }
+}
+
+impl Default for MQTTConfig {
+    fn default() -> MQTTConfig {
+        MQTTConfig {
+            enabled: false,
+            address: "http://localhost:1883".to_owned(),
+        }
+    }
+}
+
 #[derive(thiserror::Error, Debug)]
 pub enum ConfigError {
     #[error("I/O error")]

base-station/software/src/main.rs

@@ -66,7 +66,11 @@ async fn main() {
     } else {
         None
     };
-    let publish = Publish::init();
+    let publish = if config.influxdb.enabled {
+        Some(Publish::init(&config.mqtt))
+    } else {
+        None
+    };
 
     let mut app = Application {
         radio: Arc::new(Mutex::new(radio)),
@@ -89,7 +93,7 @@ async fn main() {
 /// Type which implements packet handling and responds to requests from devices.
 struct Application {
     radio: Arc<Mutex<Radio>>,
-    publish: Arc<Mutex<Publish>>,
+    publish: Arc<Mutex<Option<Publish>>>,
     tsdb: Arc<Mutex<Option<TimeSeriesDatabase>>>,
     current: CurrentValues,
     rng: ThreadRng,
@@ -161,7 +165,9 @@ impl Application {
         let r = report.clone();
         tokio::spawn(async move {
             let mut publish = publish.lock().await;
-            publish.publish(location, &r.values[0..count]).await;
+            if let Some(publish) = publish.as_mut() {
+                publish.publish(location, &r.values[0..count]).await;
+            }
         });
 
         // Insert values into TSDB.

base-station/software/src/mqtt.rs

@@ -1,86 +0,0 @@
-use std::io::Write;
-use std::net::TcpStream;
-use std::thread;
-
-use log::{error, info};
-use mqtt::control::variable_header::ConnectReturnCode;
-use mqtt::packet::*;
-use mqtt::{Decodable, Encodable, QualityOfService};
-use mqtt::{TopicFilter, TopicName};
-
-use super::Error;
-
-pub(crate) struct MQTT {
-    stream: TcpStream,
-}
-
-impl MQTT {
-    pub(crate) fn connect(server_addr: &str) -> Result<MQTT, Error> {
-        info!("Connecting to MQTT broker at {}...", server_addr);
-        let mut stream = TcpStream::connect(server_addr)?;
-
-        let mut conn = ConnectPacket::new("MQTT", "72711cd2-faaf-44f0-8490-b92eb063ff4b");
-        conn.set_clean_session(true);
-        let mut buf = Vec::new();
-        conn.encode(&mut buf).unwrap();
-        stream.write_all(&buf[..])?;
-
-        let connack = ConnackPacket::decode(&mut stream).unwrap();
-        if connack.connect_return_code() != ConnectReturnCode::ConnectionAccepted {
-            panic!(
-                "Failed to connect to server, return code {:?}",
-                connack.connect_return_code()
-            );
-        }
-        info!("Connected to MQTT broker.");
-
-        // We need to listen for packets from the server to reply to ping requests.
-        // TODO: More protocol handling.
-        let mut cloned_stream = stream.try_clone().unwrap();
-        thread::spawn(move || {
-            loop {
-                let packet = match VariablePacket::decode(&mut cloned_stream) {
-                    Ok(pk) => pk,
-                    Err(err) => {
-                        error!("could not decode received packet: {:?}", err);
-                        continue;
-                    }
-                };
-
-                match packet {
-                    VariablePacket::PingreqPacket(..) => {
-                        let pingresp = PingrespPacket::new();
-                        info!("Sending Ping response {:?}", pingresp);
-                        pingresp.encode(&mut cloned_stream).unwrap();
-                    }
-                    VariablePacket::DisconnectPacket(..) => {
-                        // Nothing to do here, the other end closes the connection.
-                        break;
-                    }
-                    _ => {}
-                }
-            }
-        });
-
-        Ok(MQTT { stream })
-    }
-
-    pub(crate) fn publish(&mut self, topic: &str, value: &str) -> Result<(), Error> {
-        // Publish the value.
-        let topic = TopicName::new(topic).unwrap();
-        let publish_packet = PublishPacket::new(topic, QoSWithPacketIdentifier::Level0, value);
-        let mut buf = Vec::new();
-        publish_packet.encode(&mut buf).unwrap();
-        self.stream.write_all(&buf[..])?;
-
-        // Check for any packets from the server to see whether we were disconnected.
-        // TODO
-        Ok(())
-    }
-}
-
-impl Drop for MQTT {
-    fn drop(&mut self) {
-        // TODO: Stop the receive thread.
-    }
-}

base-station/software/src/publish.rs

@@ -1,12 +1,32 @@
 //! Interface to a publish values via MQTT.
 
-use protocol::{Location, Value};
+use std::collections::HashMap;
+use std::mem;
+use std::ops::DerefMut;
+use std::sync::Arc;
+use std::time::Duration;
+
+use log::{error, info};
+use mqtt::control::variable_header::ConnectReturnCode;
+use mqtt::packet::*;
+use mqtt::{Encodable, TopicName};
+use tokio::io::AsyncWriteExt;
+use tokio::net::TcpStream;
+use tokio::sync::{Mutex, Notify};
+use tokio::time::sleep;
 
-// TODO: Configuration mechanism.
+use super::config::MQTTConfig;
+use protocol::{Location, Value};
 
 /// Interface to MQTT.
 pub struct Publish {
-    // TODO
+    // We only want to send the latest values if we do not have a connection, so we simply buffer
+    // the values until the sender thread collects them.
+    to_be_published: Arc<Mutex<HashMap<Location, Vec<Value>>>>,
+    trigger: Arc<Notify>,
+    /*config: MQTTConfig,
+    //publish_send: UnboundedSender<(Location, Vec<Value>)>,
+    stream: Option<OwnedWriteHalf>,*/
 }
 
 impl Publish {
@@ -15,9 +35,184 @@ impl Publish {
     /// The function does not return an error. Instead, the code will automatically retry
     /// connection in case the connection cannot be established or the server closes the
     /// connection.
-    pub fn init() -> Publish {
-        // TODO
-        Publish {}
+    pub fn init(config: &MQTTConfig) -> Publish {
+        let to_be_published = Arc::new(Mutex::new(HashMap::new()));
+        let to_be_published2 = to_be_published.clone();
+        let trigger = Arc::new(Notify::new());
+        let trigger2 = trigger.clone();
+        let config = config.clone();
+
+        tokio::spawn(async move {
+            let mut currently_sending = HashMap::new();
+            loop {
+                // When we are not connected, try to connect.
+                let stream = match Self::connect(&config).await {
+                    Some(stream) => stream,
+                    None => {
+                        sleep(Duration::from_millis(500)).await;
+                        continue;
+                    }
+                };
+                let (mut read, write) = stream.into_split();
+                let write = Arc::new(Mutex::new(write));
+
+                // When we are connected, listen for ping packets and send a response.
+                {
+                    let write = write.clone();
+                    tokio::spawn(async move {
+                        loop {
+                            let packet = match VariablePacket::parse(&mut read).await {
+                                Ok(packet) => packet,
+                                Err(e) => {
+                                    error!("could not receive MQTT packet: {:?}", e);
+                                    return;
+                                }
+                            };
+
+                            match packet {
+                                VariablePacket::PingreqPacket(..) => {
+                                    let pingresp = PingrespPacket::new();
+                                    let mut data = Vec::new();
+                                    pingresp.encode(&mut data).unwrap();
+                                    let mut write = write.lock().await;
+                                    if let Err(e) = write.write_all(&data[..]).await {
+                                        error!("could not reply to MQTT ping packet: {:?}", e);
+                                        return;
+                                    }
+                                }
+                                VariablePacket::DisconnectPacket(..) => {
+                                    // Nothing to do here, the other end closes the connection.
+                                    error!("MQTT server closed the connection.");
+                                    return;
+                                }
+                                _ => {}
+                            }
+                        }
+                    });
+                }
+
+                // Whenever we receive values to be sent, send them until the server disconnects or
+                // until we are told to shut down.
+                loop {
+                    // Fetch values to be sent.
+                    let to_be_published = {
+                        let mut new = HashMap::<Location, Vec<Value>>::new();
+                        mem::swap(&mut new, to_be_published2.lock().await.deref_mut());
+                        new
+                    };
+                    for (location, values) in to_be_published {
+                        // TODO: Merge values.
+                        currently_sending.insert(location, values);
+                    }
+
+                    // If there is nothing else to send, wait and try again.
+                    if currently_sending.len() == 0 {
+                        trigger2.notified().await;
+                        continue;
+                    }
+
+                    // Publish one value, then start again.
+                    let first_location = *currently_sending.keys().next().unwrap();
+                    let count = currently_sending[&first_location].len();
+                    let first_value = currently_sending[&first_location][count - 1];
+                    let (label, value) = match first_value {
+                        Value::Temperature(temperature) => {
+                            let temperature =
+                                temperature as f32 / first_value.decimal_factor() as f32;
+                            ("temperature", temperature)
+                        }
+                        Value::Pressure(pressure) => {
+                            let pressure = pressure as f32 / first_value.decimal_factor() as f32;
+                            ("pressure", pressure)
+                        }
+                        Value::Humidity(humidity) => {
+                            let humidity = humidity as f32 / first_value.decimal_factor() as f32;
+                            ("humidity", humidity)
+                        }
+                        _ => {
+                            // Invalid value, remove it from the queue and try again.
+                            currently_sending.get_mut(&first_location).unwrap().pop();
+                            if currently_sending[&first_location].len() == 0 {
+                                currently_sending.remove(&first_location);
+                            }
+                            continue;
+                        }
+                    };
+                    let topic = format!(
+                        "smarthome/{}/{}",
+                        first_location.to_str().to_lowercase(),
+                        label
+                    );
+                    let topic = TopicName::new(topic).unwrap();
+                    let publish_packet = PublishPacket::new(
+                        topic,
+                        QoSWithPacketIdentifier::Level0,
+                        format!("{}", value),
+                    );
+                    let mut buf = Vec::new();
+                    publish_packet.encode(&mut buf).unwrap();
+                    if let Err(e) = write.lock().await.write_all(&buf[..]).await {
+                        error!("could not send value to MQTT: {:?}", e);
+                        break;
+                    } else {
+                        // The value was sent, remove it from the queue.
+                        currently_sending.get_mut(&first_location).unwrap().pop();
+                        if currently_sending[&first_location].len() == 0 {
+                            currently_sending.remove(&first_location);
+                        }
+                    }
+                }
+            }
+        });
+        Publish {
+            to_be_published,
+            trigger,
+        }
+    }
+
+    async fn connect(config: &MQTTConfig) -> Option<TcpStream> {
+        info!("Connecting to MQTT broker...");
+        let mut stream = match TcpStream::connect(&config.address).await {
+            Ok(stream) => stream,
+            Err(e) => {
+                error!("Failed to connect to the MQTT broker: {:?}", e);
+                return None;
+            }
+        };
+
+        let mut conn = ConnectPacket::new("72711cd2-faaf-44f0-8490-b92eb063ff4b");
+        conn.set_clean_session(true);
+        let mut buf = Vec::new();
+        conn.encode(&mut buf).unwrap();
+        match stream.write_all(&buf[..]).await {
+            Ok(stream) => stream,
+            Err(e) => {
+                error!("Failed to send MQTT connect packet: {:?}", e);
+                return None;
+            }
+        };
+
+        let connack = match VariablePacket::parse(&mut stream).await {
+            Ok(VariablePacket::ConnackPacket(connack)) => connack,
+            Ok(_) => {
+                error!("Received wrong MQTT packet, expected CONNACK!");
+                return None;
+            }
+            Err(e) => {
+                error!("Failed to receive MQTT CONNACK packet: {:?}", e);
+                return None;
+            }
+        };
+        if connack.connect_return_code() != ConnectReturnCode::ConnectionAccepted {
+            error!(
+                "Failed to connect to server, return code {:?}",
+                connack.connect_return_code()
+            );
+            return None;
+        }
+
+        info!("Connected to MQTT broker.");
+        Some(stream)
     }
 
     /// Publishes values received from a device.
@@ -30,7 +225,16 @@ impl Publish {
     ///
     /// * `location` - the location of the device
     /// * `values` - a list of values to be published
-    pub async fn publish(&mut self, _location: Location, _values: &[Value]) {
-        // TODO
+    pub async fn publish(&mut self, location: Location, values: &[Value]) {
+        let mut buffer = self.to_be_published.lock().await;
+        // TODO: Merge the values?
+        *buffer.entry(location).or_insert(Vec::new()) = values.to_vec();
+        self.trigger.notify_one();
+    }
+}
+
+impl Drop for Publish {
+    fn drop(&mut self) {
+        // TODO: Stop the thread.
     }
 }