base-station: Add code to publish data to MQTT.

base-station/software/Cargo.lock

@@ -30,6 +30,7 @@ dependencies = [
  "linux-embedded-hal",
  "log",
  "mqtt-protocol",
+ "thiserror",
 ]
 
 [[package]]
@@ -216,12 +217,30 @@ dependencies = [
  "void",
 ]
 
+[[package]]
+name = "proc-macro2"
+version = "1.0.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df246d292ff63439fea9bc8c0a270bed0e390d5ebd4db4ba15aba81111b5abe3"
+dependencies = [
+ "unicode-xid",
+]
+
 [[package]]
 name = "quick-error"
 version = "1.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0"
 
+[[package]]
+name = "quote"
+version = "1.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2bdc6c187c65bca4260c9011c9e3132efe4909da44726bad24cf7572ae338d7f"
+dependencies = [
+ "proc-macro2",
+]
+
 [[package]]
 name = "regex"
 version = "1.3.7"
@@ -296,6 +315,17 @@ dependencies = [
  "nix",
 ]
 
+[[package]]
+name = "syn"
+version = "1.0.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0df0eb663f387145cab623dea85b09c2c5b4b0aef44e945d928e682fce71bb03"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "unicode-xid",
+]
+
 [[package]]
 name = "sysfs_gpio"
 version = "0.5.4"
@@ -323,6 +353,26 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "thiserror"
+version = "1.0.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "54b3d3d2ff68104100ab257bb6bb0cb26c901abe4bd4ba15961f3bf867924012"
+dependencies = [
+ "thiserror-impl",
+]
+
+[[package]]
+name = "thiserror-impl"
+version = "1.0.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ca972988113b7715266f91250ddb98070d033c62a011fa0fcc57434a649310dd"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "thread_local"
 version = "1.0.1"
@@ -332,6 +382,12 @@ dependencies = [
  "lazy_static",
 ]
 
+[[package]]
+name = "unicode-xid"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "826e7639553986605ec5979c7dd957c7895e93eabed50ab2ffa7f6128a75097c"
+
 [[package]]
 name = "void"
 version = "1.0.2"

base-station/software/Cargo.toml

@@ -13,3 +13,4 @@ embedded-hal = "0.2.3"
 mqtt-protocol = "0.8"
 env_logger = "0.7"
 log = "0.4"
+thiserror = "1.0"

base-station/software/src/main.rs

@@ -1,7 +1,14 @@
 use std::env;
-use std::thread;
+use std::io;
+use std::thread::sleep;
 use std::time::Duration;
 
+use log::error;
+use ::mqtt::control::variable_header::ConnectReturnCode;
+
+use crate::mqtt::MQTT;
+
+mod mqtt;
 mod radio;
 mod spi;
 
@@ -14,8 +21,42 @@ fn main() {
 
     radio::start();
 
-    // TODO
+    // Whenever the MQTT connection returns an error, we wait for a couple of seconds and then try
+    // to reconnect.
+    loop {
+        // TODO: Pass events from ratio to mqtt connection.
+        if let Err(e) = run_mqtt_connection() {
+            error!("MQTT error: {:?}", e);
+        }
+        sleep(Duration::from_secs(3));
+    }
+}
+
+fn run_mqtt_connection() -> Result<(), Error> {
+    let mut m = MQTT::connect("127.0.0.1:1883")?;
+
+    let mut counter = 0;
     loop {
-        thread::sleep(Duration::from_secs(10));
+        // TODO: Real values
+        m.publish("gottschlag/livingroom/temperature", &format!("{}", counter))?;
+
+        sleep(Duration::from_secs(3));
+        counter += 1;
+    }
+}
+
+#[derive(thiserror::Error, Debug)]
+pub enum Error {
+    #[error("I/O error")]
+    Io(#[from] io::Error),
+    #[error("connection refused, return code {0:?}")]
+    ConnectionRefused(ConnectReturnCode),
+    #[error("radio error: {0:?}")]
+    Radio(embedded_nrf24l01::Error<std::io::Error>)
+}
+
+impl From<embedded_nrf24l01::Error<std::io::Error>> for Error {
+    fn from(e: embedded_nrf24l01::Error<std::io::Error>) -> Self {
+        Error::Radio(e)
     }
 }

base-station/software/src/mqtt.rs

@@ -0,0 +1,88 @@
+use std::io::Write;
+use std::net::TcpStream;
+use std::thread;
+
+use log::{info, error};
+use mqtt::control::variable_header::ConnectReturnCode;
+use mqtt::packet::*;
+use mqtt::{Decodable, Encodable, QualityOfService};
+use mqtt::{TopicFilter, TopicName};
+
+use super::Error;
+
+pub struct MQTT {
+    stream: TcpStream,
+}
+
+impl MQTT {
+    pub 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 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/radio.rs

@@ -1,5 +1,4 @@
 use std::convert::TryInto;
-use std::io;
 use std::thread;
 use std::time::{Duration, Instant};
 
@@ -10,6 +9,7 @@ use linux_embedded_hal::Pin;
 use log::{error, info};
 
 use crate::spi::EmbeddedHalSpidev;
+use crate::Error;
 
 pub fn start() {
     // TODO: Channels for sensor readings.
@@ -27,7 +27,7 @@ pub fn start() {
     });
 }
 
-fn radio_thread() -> Result<(), io::Error> {
+fn radio_thread() -> Result<(), Error> {
     info!("Initializing radio...");
 
     // The NRF module is connected as follows:
@@ -63,14 +63,14 @@ fn radio_thread() -> Result<(), io::Error> {
     irq.set_direction(Direction::In).unwrap();
 
     // Initialize the radio module.
-    let mut nrf24 = NRF24L01::new(ce, cs, spi).unwrap();
-    nrf24.set_frequency(0x32).unwrap();
-    nrf24.set_rf(DataRate::R2Mbps, 3).unwrap();
-    nrf24.set_crc(Some(CrcMode::OneByte)).unwrap();
-    nrf24.set_auto_retransmit(250, 3).unwrap();
+    let mut nrf24 = NRF24L01::new(ce, cs, spi)?;
+    nrf24.set_frequency(0x32)?;
+    nrf24.set_rf(DataRate::R2Mbps, 3)?;
+    nrf24.set_crc(Some(CrcMode::OneByte))?;
+    nrf24.set_auto_retransmit(250, 3)?;
     nrf24
         .set_pipes_rx_enable(&[true, false, false, false, false, false])
-        .unwrap();
+        ?;
     nrf24
         .set_rx_addr(0, &[0x56, 0x34, 0x12, 0x00, 0x00])
         .unwrap();