Moved the mqtt topic check up one leve into Devices

src/devices.rs

@@ -15,7 +15,7 @@ use std::collections::HashMap;
 use async_trait::async_trait;
 use google_home::{traits::OnOff, FullfillmentError, GoogleHome, GoogleHomeDevice};
 use pollster::FutureExt;
-use rumqttc::{AsyncClient, QoS};
+use rumqttc::{matches, AsyncClient, QoS};
 use thiserror::Error;
 use tokio::sync::{mpsc, oneshot};
 use tracing::{debug, error, trace};
@@ -192,8 +192,15 @@ impl OnMqtt for Devices {
         self.get::<dyn OnMqtt>()
             .iter_mut()
             .for_each(|(id, listener)| {
+                let subscribed = listener
+                    .topics()
+                    .iter()
+                    .any(|topic| matches(&message.topic, topic));
+
+                if subscribed {
                     trace!(id, "Handling");
                     listener.on_mqtt(message).block_on();
+                }
             })
     }
 }

src/devices/audio_setup.rs

@@ -1,6 +1,5 @@
 use async_trait::async_trait;
 use google_home::traits;
-use rumqttc::matches;
 use tracing::{debug, error, warn};
 
 use crate::config::MqttDeviceConfig;
@@ -56,10 +55,6 @@ impl OnMqtt for AudioSetup {
     }
 
     async fn on_mqtt(&mut self, message: &rumqttc::Publish) {
-        if !matches(&message.topic, &self.mqtt.topic) {
-            return;
-        }
-
         let action = match RemoteMessage::try_from(message) {
             Ok(message) => message.action(),
             Err(err) => {

src/devices/contact_sensor.rs

@@ -1,7 +1,7 @@
 use std::time::Duration;
 
 use async_trait::async_trait;
-use rumqttc::{matches, AsyncClient};
+use rumqttc::AsyncClient;
 use tokio::task::JoinHandle;
 use tracing::{debug, error, warn};
 
@@ -64,10 +64,6 @@ impl OnMqtt for ContactSensor {
     }
 
     async fn on_mqtt(&mut self, message: &rumqttc::Publish) {
-        if !matches(&message.topic, &self.mqtt.topic) {
-            return;
-        }
-
         let is_closed = match ContactMessage::try_from(message) {
             Ok(state) => state.is_closed(),
             Err(err) => {

src/devices/ikea_outlet.rs

@@ -7,7 +7,7 @@ use google_home::{
     GoogleHomeDevice,
 };
 use pollster::FutureExt as _;
-use rumqttc::{matches, AsyncClient, Publish};
+use rumqttc::{AsyncClient, Publish};
 use std::time::Duration;
 use tokio::task::JoinHandle;
 use tracing::{debug, error, warn};
@@ -83,10 +83,6 @@ impl OnMqtt for IkeaOutlet {
 
     async fn on_mqtt(&mut self, message: &Publish) {
         // Update the internal state based on what the device has reported
-        if !matches(&message.topic, &self.mqtt.topic) {
-            return;
-        }
-
         let state = match OnOffMessage::try_from(message) {
             Ok(state) => state.state(),
             Err(err) => {

src/devices/wake_on_lan.rs

@@ -9,7 +9,7 @@ use google_home::{
     types::Type,
     GoogleHomeDevice,
 };
-use rumqttc::{matches, Publish};
+use rumqttc::Publish;
 use tracing::{debug, error};
 
 use crate::{
@@ -59,10 +59,6 @@ impl OnMqtt for WakeOnLAN {
     }
 
     async fn on_mqtt(&mut self, message: &Publish) {
-        if !matches(&message.topic, &self.mqtt.topic) {
-            return;
-        }
-
         let activate = match ActivateMessage::try_from(message) {
             Ok(message) => message.activate(),
             Err(err) => {