restructure and error logging

Cargo.lock

@@ -1,5 +1,20 @@
 # This file is automatically @generated by Cargo.
 # It is not intended for manual editing.
+[[package]]
+name = "addr2line"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7c0929d69e78dd9bf5408269919fcbcaeb2e35e5d43e5815517cdc6a8e11a423"
+dependencies = [
+ "gimli",
+]
+
+[[package]]
+name = "adler"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ee2a4ec343196209d6594e19543ae87a39f96d5534d7174822a3ad825dd6ed7e"
+
 [[package]]
 name = "ahash"
 version = "0.3.8"
@@ -9,6 +24,15 @@ dependencies = [
  "const-random",
 ]
 
+[[package]]
+name = "ansi_term"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ee49baf6cb617b853aa8d93bf420db2383fab46d314482ca2803b40d5fde979b"
+dependencies = [
+ "winapi 0.3.9",
+]
+
 [[package]]
 name = "async-channel"
 version = "1.5.1"
@@ -20,6 +44,27 @@ dependencies = [
  "futures-core",
 ]
 
+[[package]]
+name = "async-stream"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3670df70cbc01729f901f94c887814b3c68db038aad1329a418bae178bc5295c"
+dependencies = [
+ "async-stream-impl",
+ "futures-core",
+]
+
+[[package]]
+name = "async-stream-impl"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a3548b8efc9f8e8a5a0a2808c5bd8451a9031b9e5b879a79590304ae928b0a70"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "autocfg"
 version = "0.1.7"
@@ -32,6 +77,20 @@ version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cdb031dd78e28731d87d56cc8ffef4a8f36ca26c38fe2de700543e627f8a464a"
 
+[[package]]
+name = "backtrace"
+version = "0.3.55"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ef5140344c85b01f9bbb4d4b7288a8aa4b3287ccef913a14bcc78a1063623598"
+dependencies = [
+ "addr2line",
+ "cfg-if 1.0.0",
+ "libc",
+ "miniz_oxide",
+ "object",
+ "rustc-demangle",
+]
+
 [[package]]
 name = "base64"
 version = "0.12.3"
@@ -141,6 +200,32 @@ dependencies = [
  "bitflags",
 ]
 
+[[package]]
+name = "color-eyre"
+version = "0.5.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "86bc0bb03923141924d5b713a4acd7607c790f3fbc769abe63fe3f38bb268112"
+dependencies = [
+ "backtrace",
+ "color-spantrace",
+ "eyre",
+ "indenter",
+ "once_cell",
+ "owo-colors",
+ "tracing-error",
+]
+
+[[package]]
+name = "color-spantrace"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6a99aa4aa18448eef4c7d3f86d2720d2d8cad5c860fe9ff9b279293efdc8f5be"
+dependencies = [
+ "ansi_term",
+ "tracing-core",
+ "tracing-error",
+]
+
 [[package]]
 name = "concurrent-queue"
 version = "1.2.2"
@@ -233,6 +318,16 @@ version = "2.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f7531096570974c3a9dcf9e4b8e1cede1ec26cf5046219fb3b9d897503b9be59"
 
+[[package]]
+name = "eyre"
+version = "0.6.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5f29abf4740a4778632fe27a4f681ef5b7a6f659aeba3330ac66f48e20cfa3b7"
+dependencies = [
+ "indenter",
+ "once_cell",
+]
+
 [[package]]
 name = "fake-simd"
 version = "0.1.2"
@@ -342,6 +437,19 @@ dependencies = [
  "slab",
 ]
 
+[[package]]
+name = "generator"
+version = "0.6.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8cdc09201b2e8ca1b19290cf7e65de2246b8e91fb6874279722189c4de7b94dc"
+dependencies = [
+ "cc",
+ "libc",
+ "log",
+ "rustc_version",
+ "winapi 0.3.9",
+]
+
 [[package]]
 name = "generic-array"
 version = "0.12.3"
@@ -383,6 +491,12 @@ dependencies = [
  "wasi 0.9.0+wasi-snapshot-preview1",
 ]
 
+[[package]]
+name = "gimli"
+version = "0.23.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f6503fe142514ca4799d4c26297c4248239fe8838d827db6bd6065c6ed29a6ce"
+
 [[package]]
 name = "h2"
 version = "0.2.7"
@@ -511,6 +625,12 @@ dependencies = [
  "unicode-normalization",
 ]
 
+[[package]]
+name = "indenter"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e0bd112d44d9d870a6819eb505d04dd92b5e4d94bb8c304924a0872ae7016fb5"
+
 [[package]]
 name = "indexmap"
 version = "1.6.0"
@@ -591,6 +711,19 @@ dependencies = [
  "cfg-if 0.1.10",
 ]
 
+[[package]]
+name = "loom"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a0e8460f2f2121162705187214720353c517b97bdfb3494c0b1e33d83ebe4bed"
+dependencies = [
+ "cfg-if 0.1.10",
+ "generator",
+ "scoped-tls",
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "matches"
 version = "0.1.8"
@@ -619,6 +752,16 @@ dependencies = [
  "unicase",
 ]
 
+[[package]]
+name = "miniz_oxide"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0f2d26ec3309788e423cfbf68ad1800f061638098d76a83681af979dc4eda19d"
+dependencies = [
+ "adler",
+ "autocfg 1.0.1",
+]
+
 [[package]]
 name = "mio"
 version = "0.6.22"
@@ -743,6 +886,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "object"
+version = "0.22.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8d3b63360ec3cb337817c2dbd47ab4a0f170d285d8e5a2064600f3def1402397"
+
 [[package]]
 name = "once_cell"
 version = "1.5.2"
@@ -761,6 +910,12 @@ version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "624a8340c38c1b80fd549087862da4ba43e08858af025b236e509b6649fc13d5"
 
+[[package]]
+name = "owo-colors"
+version = "1.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7a1250cdd103eef6bd542b5ae82989f931fc00a41a27f60377338241594410f3"
+
 [[package]]
 name = "percent-encoding"
 version = "2.1.0"
@@ -1064,6 +1219,21 @@ dependencies = [
  "webpki",
 ]
 
+[[package]]
+name = "rustc-demangle"
+version = "0.1.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6e3bad0ee36814ca07d7968269dd4b7ec89ec2da10c4bb613928d3077083c232"
+
+[[package]]
+name = "rustc_version"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "138e3e0acb6c9fb258b19b67cb8abd63c00679d2851805ea151465464fe9030a"
+dependencies = [
+ "semver",
+]
+
 [[package]]
 name = "rustls"
 version = "0.18.1"
@@ -1105,11 +1275,40 @@ dependencies = [
  "untrusted",
 ]
 
+[[package]]
+name = "semver"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1d7eb9ef2c18661902cc47e535f9bc51b78acd254da71d375c2f6720d9a40403"
+dependencies = [
+ "semver-parser",
+]
+
+[[package]]
+name = "semver-parser"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "388a1df253eca08550bef6c72392cfe7c30914bf41df5269b68cbd6ff8f570a3"
+
 [[package]]
 name = "serde"
 version = "1.0.117"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b88fa983de7720629c9387e9f517353ed404164b1e482c970a90c1a4aaf7dc1a"
+dependencies = [
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_derive"
+version = "1.0.117"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cbd1ae72adb44aab48f325a02444a5fc079349a8d804c1fc922aed3f7454c74e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
 
 [[package]]
 name = "serde_json"
@@ -1159,6 +1358,16 @@ dependencies = [
  "opaque-debug 0.3.0",
 ]
 
+[[package]]
+name = "sharded-slab"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7b4921be914e16899a80adefb821f8ddb7974e3f1250223575a44ed994882127"
+dependencies = [
+ "lazy_static",
+ "loom",
+]
+
 [[package]]
 name = "signal-hook-registry"
 version = "1.2.2"
@@ -1207,10 +1416,13 @@ dependencies = [
 name = "taspromto"
 version = "0.1.0"
 dependencies = [
+ "async-stream",
+ "color-eyre",
  "ctrlc",
  "dashmap",
  "dotenv",
  "json",
+ "pin-utils",
  "rumqttc",
  "tokio",
  "warp",
@@ -1250,6 +1462,15 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "thread_local"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d40c6d1b69745a6ec6fb1ca717914848da4b44ae29d9b3080cbee91d72a69b14"
+dependencies = [
+ "lazy_static",
+]
+
 [[package]]
 name = "time"
 version = "0.1.44"
@@ -1365,9 +1586,21 @@ dependencies = [
  "cfg-if 0.1.10",
  "log",
  "pin-project-lite",
+ "tracing-attributes",
  "tracing-core",
 ]
 
+[[package]]
+name = "tracing-attributes"
+version = "0.1.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "80e0ccfc3378da0cce270c946b676a376943f5cd16aeba64568e7939806f4ada"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "tracing-core"
 version = "0.1.17"
@@ -1377,6 +1610,16 @@ dependencies = [
  "lazy_static",
 ]
 
+[[package]]
+name = "tracing-error"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b4d7c0b83d4a500748fa5879461652b361edf5c9d51ede2a2ac03875ca185e24"
+dependencies = [
+ "tracing",
+ "tracing-subscriber",
+]
+
 [[package]]
 name = "tracing-futures"
 version = "0.2.4"
@@ -1387,6 +1630,17 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "tracing-subscriber"
+version = "0.2.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a1fa8f0c8f4c594e4fc9debc1990deab13238077271ba84dd853d54902ee3401"
+dependencies = [
+ "sharded-slab",
+ "thread_local",
+ "tracing-core",
+]
+
 [[package]]
 name = "try-lock"
 version = "0.2.3"

Cargo.toml

@@ -14,3 +14,6 @@ json = "0.12.4"
 warp = "0.2.5"
 dotenv = "0.15.0"
 ctrlc = { version = "3.1.7", features = ["termination"] }
+color-eyre = "0.5.7"
+async-stream = "0.3.0"
+pin-utils = "0.1.0"

src/config.rs

@@ -0,0 +1,29 @@
+use color_eyre::{eyre::WrapErr, Result};
+use std::str::FromStr;
+
+#[derive(Default)]
+pub struct Config {
+    pub mqtt_host: String,
+    pub mqtt_port: u16,
+    pub host_port: u16,
+}
+
+impl Config {
+    pub fn from_env() -> Result<Self> {
+        let mqtt_host = dotenv::var("MQTT_HOSTNAME").wrap_err("MQTT_HOSTNAME not set")?;
+        let mqtt_port = dotenv::var("MQTT_PORT")
+            .ok()
+            .and_then(|port| u16::from_str(&port).ok())
+            .unwrap_or(1883);
+        let host_port = dotenv::var("PORT")
+            .ok()
+            .and_then(|port| u16::from_str(&port).ok())
+            .unwrap_or(80);
+
+        Ok(Config {
+            mqtt_host,
+            mqtt_port,
+            host_port,
+        })
+    }
+}

src/device.rs

@@ -0,0 +1,61 @@
+use std::fmt::Write;
+
+#[derive(Debug, Eq, PartialEq, Clone, Hash)]
+pub struct Device {
+    pub topic: String,
+    pub hostname: String,
+}
+
+impl Device {
+    pub fn get_topic(&self, prefix: &str, command: &str) -> String {
+        format!("{}/{}/{}/{}", prefix, self.topic, self.hostname, command)
+    }
+}
+
+#[derive(Debug, Default)]
+pub struct DeviceState {
+    pub state: bool,
+    pub name: String,
+    pub power_watts: Option<f32>,
+    pub power_yesterday: Option<f32>,
+    pub power_today: Option<f32>,
+}
+
+pub fn format_device_state<W: Write>(
+    mut writer: W,
+    device: &Device,
+    state: &DeviceState,
+) -> Result<(), std::fmt::Error> {
+    writeln!(
+        writer,
+        "switch_state{{tasmota_id=\"{}\", name=\"{}\"}} {}",
+        device.hostname,
+        state.name,
+        if state.state { 1 } else { 0 }
+    )?;
+
+    if let Some(power_watts) = state.power_watts {
+        writeln!(
+            writer,
+            "power_watts{{tasmota_id=\"{}\", name=\"{}\"}} {}",
+            device.hostname, state.name, power_watts
+        )?;
+    }
+
+    if let Some(power_yesterday) = state.power_yesterday {
+        writeln!(
+            writer,
+            "power_yesterday_kwh{{tasmota_id=\"{}\", name=\"{}\"}} {}",
+            device.hostname, state.name, power_yesterday
+        )?;
+    }
+
+    if let Some(power_today) = state.power_today {
+        writeln!(
+            writer,
+            "power_today_kwh{{tasmota_id=\"{}\", name=\"{}\"}} {}",
+            device.hostname, state.name, power_today
+        )?;
+    }
+    Ok(())
+}

src/main.rs

@@ -1,25 +1,28 @@
+mod config;
+mod device;
+mod mqtt;
+mod topic;
+
+use crate::config::Config;
+use crate::device::{format_device_state, Device, DeviceState};
+use crate::mqtt::mqtt_stream;
+use crate::topic::Topic;
+use color_eyre::{eyre::WrapErr, Result};
 use dashmap::DashMap;
-use rumqttc::{AsyncClient, Event, MqttOptions, Packet, QoS};
+use pin_utils::pin_mut;
+use rumqttc::{MqttOptions, QoS};
 use std::convert::TryFrom;
-use std::fmt::Write;
-use std::str::FromStr;
 use std::sync::Arc;
+use tokio::stream::StreamExt;
 use tokio::time::Duration;
 use warp::Filter;
 
 type DeviceStates = Arc<DashMap<Device, DeviceState>>;
 
 #[tokio::main]
-async fn main() {
+async fn main() -> Result<()> {
-    let mqtt_host = dotenv::var("MQTT_HOSTNAME").expect("MQTT_HOSTNAME not set");
+    let config = Config::from_env()?;
-    let mqtt_port = dotenv::var("MQTT_PORT")
+    let host_port = config.host_port;
-        .ok()
-        .and_then(|port| u16::from_str(&port).ok())
-        .unwrap_or(1883);
-    let host_port = dotenv::var("PORT")
-        .ok()
-        .and_then(|port| u16::from_str(&port).ok())
-        .unwrap_or(80);
 
     let device_states = DeviceStates::default();
 
@@ -31,8 +34,10 @@ async fn main() {
     let states = device_states.clone();
     tokio::task::spawn(async move {
         loop {
-            mqtt_client(&mqtt_host, mqtt_port, states.clone()).await;
+            if let Err(e) = mqtt_client(&config.mqtt_host, config.mqtt_port, states.clone()).await {
-            eprintln!("lost mqtt collection, reconnecting after 1s");
+                eprintln!("lost mqtt collection: {:#}", e);
+            }
+            eprintln!("reconnecting after 1s");
             tokio::time::delay_for(Duration::from_secs(1)).await;
         }
     });
@@ -43,220 +48,84 @@ async fn main() {
         .map(|state: DeviceStates| {
             let mut response = String::new();
             for device in state.iter() {
-                writeln!(
+                format_device_state(&mut response, &device.key(), &device.value()).unwrap();
-                    &mut response,
-                    "switch_state{{tasmota_id=\"{}\", name=\"{}\"}} {}",
-                    device.key().hostname,
-                    device.name,
-                    if device.state { 1 } else { 0 }
-                )
-                .unwrap();
-
-                if let Some(power_watts) = device.power_watts {
-                    writeln!(
-                        &mut response,
-                        "power_watts{{tasmota_id=\"{}\", name=\"{}\"}} {}",
-                        device.key().hostname,
-                        device.name,
-                        power_watts
-                    )
-                    .unwrap();
-                }
-
-                if let Some(power_yesterday) = device.power_yesterday {
-                    writeln!(
-                        &mut response,
-                        "power_yesterday_kwh{{tasmota_id=\"{}\", name=\"{}\"}} {}",
-                        device.key().hostname,
-                        device.name,
-                        power_yesterday
-                    )
-                    .unwrap();
-                }
-
-                if let Some(power_today) = device.power_today {
-                    writeln!(
-                        &mut response,
-                        "power_today_kwh{{tasmota_id=\"{}\", name=\"{}\"}} {}",
-                        device.key().hostname,
-                        device.name,
-                        power_today
-                    )
-                    .unwrap();
-                }
             }
             response
         });
 
     warp::serve(metrics).run(([0, 0, 0, 0], host_port)).await;
+    Ok(())
 }
 
-async fn mqtt_client(host: &str, port: u16, device_states: DeviceStates) {
+async fn mqtt_client(host: &str, port: u16, device_states: DeviceStates) -> Result<()> {
-    let mut mqttoptions = MqttOptions::new("rumqtt-async", host, port);
+    let mut mqtt_options = MqttOptions::new("taspromto", host, port);
-    mqttoptions.set_keep_alive(5);
+    mqtt_options.set_keep_alive(5);
 
-    let (client, mut event_loop) = AsyncClient::new(mqttoptions, 10);
+    let (client, stream) = mqtt_stream(mqtt_options)
-    client
-        .subscribe("tele/+/+/LWT", QoS::AtMostOnce)
         .await
-        .unwrap();
+        .wrap_err("Failed to setup mqtt listener")?;
-    client
-        .subscribe("stat/+/+/POWER", QoS::AtMostOnce)
-        .await
-        .unwrap();
-    client
-        .subscribe("tele/+/+/SENSOR", QoS::AtMostOnce)
-        .await
-        .unwrap();
-    client
-        .subscribe("stat/+/+/RESULT", QoS::AtMostOnce)
-        .await
-        .unwrap();
 
-    while let Ok(notification) = event_loop.poll().await {
+    pin_mut!(stream);
-        if let Event::Incoming(Packet::Publish(message)) = notification {
-            println!(
-                "{} {}",
-                message.topic,
-                std::str::from_utf8(message.payload.as_ref()).unwrap_or_default()
-            );
-            let topic = Topic::from(message.topic.as_str());
 
-            match topic {
+    while let Some(message) = stream.next().await {
-                Topic::LWT(device) => {
+        let message = message?;
-                    // on discovery, ask the device for it's power state and name
+        println!(
-                    client
+            "{} {}",
-                        .publish(
+            message.topic,
-                            device.get_topic("cmnd", "POWER"),
+            std::str::from_utf8(message.payload.as_ref()).unwrap_or_default()
-                            QoS::AtMostOnce,
+        );
-                            false,
+        let topic = Topic::from(message.topic.as_str());
-                            "",
+
-                        )
+        match topic {
-                        .await
+            Topic::LWT(device) => {
-                        .unwrap();
+                // on discovery, ask the device for it's power state and name
-                    client
+                client
-                        .publish(
+                    .publish(
-                            device.get_topic("cmnd", "DeviceName"),
+                        device.get_topic("cmnd", "POWER"),
-                            QoS::AtMostOnce,
+                        QoS::AtMostOnce,
-                            false,
+                        false,
-                            "",
+                        "",
-                        )
+                    )
-                        .await
+                    .await?;
-                        .unwrap();
+                client
-                }
+                    .publish(
-                Topic::Power(device) => {
+                        device.get_topic("cmnd", "DeviceName"),
-                    let state = message.payload.as_ref() == b"ON";
+                        QoS::AtMostOnce,
-                    device_states.entry(device).or_default().state = state;
+                        false,
-                }
+                        "",
-                Topic::Result(device) => {
+                    )
-                    let payload = std::str::from_utf8(message.payload.as_ref()).unwrap_or_default();
+                    .await?;
-                    if let Ok(json) = json::parse(payload) {
+            }
-                        let mut device_state = device_states.entry(device).or_default();
+            Topic::Power(device) => {
-                        if json["DeviceName"].is_string() {
+                let state = message.payload.as_ref() == b"ON";
-                            device_state.name = json["DeviceName"].to_string();
+                device_states.entry(device).or_default().state = state;
-                        }
+            }
+            Topic::Result(device) => {
+                let payload = std::str::from_utf8(message.payload.as_ref()).unwrap_or_default();
+                if let Ok(json) = json::parse(payload) {
+                    let mut device_state = device_states.entry(device).or_default();
+                    if json["DeviceName"].is_string() {
+                        device_state.name = json["DeviceName"].to_string();
                     }
                 }
-                Topic::Sensor(device) => {
+            }
-                    let payload = std::str::from_utf8(message.payload.as_ref()).unwrap_or_default();
+            Topic::Sensor(device) => {
-                    if let Ok(json) = json::parse(payload) {
+                let payload = std::str::from_utf8(message.payload.as_ref()).unwrap_or_default();
-                        let mut device_state = device_states.entry(device).or_default();
+                if let Ok(json) = json::parse(payload) {
-                        device_state.power_watts = json["ENERGY"]["Power"]
+                    let mut device_state = device_states.entry(device).or_default();
-                            .as_number()
+                    device_state.power_watts = json["ENERGY"]["Power"]
-                            .map(|num| f32::try_from(num).unwrap_or_default());
+                        .as_number()
-                        device_state.power_yesterday = json["ENERGY"]["Yesterday"]
+                        .map(|num| f32::try_from(num).unwrap_or_default());
-                            .as_number()
+                    device_state.power_yesterday = json["ENERGY"]["Yesterday"]
-                            .map(|num| f32::try_from(num).unwrap_or_default());
+                        .as_number()
-                        device_state.power_today = json["ENERGY"]["Today"]
+                        .map(|num| f32::try_from(num).unwrap_or_default());
-                            .as_number()
+                    device_state.power_today = json["ENERGY"]["Today"]
-                            .map(|num| f32::try_from(num).unwrap_or_default());
+                        .as_number()
-                    }
+                        .map(|num| f32::try_from(num).unwrap_or_default());
                 }
-                _ => {}
             }
+            _ => {}
         }
     }
-}
+    Ok(())
-
-#[derive(Debug, Eq, PartialEq, Clone, Hash)]
-struct Device {
-    topic: String,
-    hostname: String,
-}
-
-impl Device {
-    fn get_topic(&self, prefix: &str, command: &str) -> String {
-        format!("{}/{}/{}/{}", prefix, self.topic, self.hostname, command)
-    }
-}
-
-#[derive(Debug, Default)]
-struct DeviceState {
-    state: bool,
-    name: String,
-    power_watts: Option<f32>,
-    power_yesterday: Option<f32>,
-    power_today: Option<f32>,
-}
-
-#[derive(Debug, Eq, PartialEq)]
-enum Topic {
-    LWT(Device),
-    Power(Device),
-    State(Device),
-    Sensor(Device),
-    Result(Device),
-    Other(String),
-}
-
-impl From<&str> for Topic {
-    fn from(raw: &str) -> Self {
-        let mut parts = raw.split('/');
-        if let (Some(prefix), Some(topic), Some(hostname), Some(cmd)) =
-            (parts.next(), parts.next(), parts.next(), parts.next())
-        {
-            let device = Device {
-                topic: topic.to_string(),
-                hostname: hostname.to_string(),
-            };
-            match (prefix, cmd) {
-                ("tele", "LWT") => Topic::LWT(device),
-                ("tele", "STATE") => Topic::State(device),
-                ("stat", "POWER") => Topic::Power(device),
-                ("tele", "SENSOR") => Topic::Sensor(device),
-                ("stat", "RESULT") => Topic::Result(device),
-                _ => Topic::Other(raw.to_string()),
-            }
-        } else {
-            Topic::Other(raw.to_string())
-        }
-    }
-}
-
-#[test]
-fn parse_topic() {
-    let device = Device {
-        hostname: "hostname".to_string(),
-        topic: "foo".to_string(),
-    };
-    assert_eq!(
-        Topic::LWT(device.clone()),
-        Topic::from("tele/foo/hostname/LWT")
-    );
-    assert_eq!(
-        Topic::Power(device.clone()),
-        Topic::from("stat/foo/hostname/POWER")
-    );
-    assert_eq!(
-        Topic::State(device.clone()),
-        Topic::from("tele/foo/hostname/STATE")
-    );
-    assert_eq!(
-        Topic::Sensor(device.clone()),
-        Topic::from("tele/foo/hostname/SENSOR")
-    );
-    assert_eq!(
-        Topic::Result(device.clone()),
-        Topic::from("stat/foo/hostname/RESULT")
-    );
 }

src/mqtt.rs

@@ -0,0 +1,31 @@
+use async_stream::try_stream;
+use color_eyre::Result;
+use rumqttc::{AsyncClient, Event, EventLoop, MqttOptions, Packet, Publish, QoS};
+use tokio::stream::{Stream, StreamExt};
+
+pub async fn mqtt_stream(
+    mqtt_options: MqttOptions,
+) -> Result<(AsyncClient, impl Stream<Item = Result<Publish>>)> {
+    let (client, event_loop) = AsyncClient::new(mqtt_options, 10);
+    client.subscribe("tele/+/+/LWT", QoS::AtMostOnce).await?;
+    client.subscribe("stat/+/+/POWER", QoS::AtMostOnce).await?;
+    client.subscribe("tele/+/+/SENSOR", QoS::AtMostOnce).await?;
+    client.subscribe("stat/+/+/RESULT", QoS::AtMostOnce).await?;
+
+    let stream = event_loop_to_stream(event_loop).filter_map(|event| match event {
+        Ok(Event::Incoming(Packet::Publish(message))) => Some(Ok(message)),
+        Ok(_) => None,
+        Err(e) => Some(Err(e)),
+    });
+
+    Ok((client, stream))
+}
+
+fn event_loop_to_stream(mut event_loop: EventLoop) -> impl Stream<Item = Result<Event>> {
+    try_stream! {
+        loop {
+            let event = event_loop.poll().await?;
+            yield event;
+        }
+    }
+}

src/topic.rs

@@ -0,0 +1,63 @@
+use crate::device::Device;
+
+#[derive(Debug, Eq, PartialEq)]
+pub enum Topic {
+    LWT(Device),
+    Power(Device),
+    State(Device),
+    Sensor(Device),
+    Result(Device),
+    Other(String),
+}
+
+impl From<&str> for Topic {
+    fn from(raw: &str) -> Self {
+        let mut parts = raw.split('/');
+        if let (Some(prefix), Some(topic), Some(hostname), Some(cmd)) =
+            (parts.next(), parts.next(), parts.next(), parts.next())
+        {
+            let device = Device {
+                topic: topic.to_string(),
+                hostname: hostname.to_string(),
+            };
+            match (prefix, cmd) {
+                ("tele", "LWT") => Topic::LWT(device),
+                ("tele", "STATE") => Topic::State(device),
+                ("stat", "POWER") => Topic::Power(device),
+                ("tele", "SENSOR") => Topic::Sensor(device),
+                ("stat", "RESULT") => Topic::Result(device),
+                _ => Topic::Other(raw.to_string()),
+            }
+        } else {
+            Topic::Other(raw.to_string())
+        }
+    }
+}
+
+#[test]
+fn parse_topic() {
+    let device = Device {
+        hostname: "hostname".to_string(),
+        topic: "foo".to_string(),
+    };
+    assert_eq!(
+        Topic::LWT(device.clone()),
+        Topic::from("tele/foo/hostname/LWT")
+    );
+    assert_eq!(
+        Topic::Power(device.clone()),
+        Topic::from("stat/foo/hostname/POWER")
+    );
+    assert_eq!(
+        Topic::State(device.clone()),
+        Topic::from("tele/foo/hostname/STATE")
+    );
+    assert_eq!(
+        Topic::Sensor(device.clone()),
+        Topic::from("tele/foo/hostname/SENSOR")
+    );
+    assert_eq!(
+        Topic::Result(device.clone()),
+        Topic::from("stat/foo/hostname/RESULT")
+    );
+}