From c238432b6d256ccd5660546e164bf728bb49335e Mon Sep 17 00:00:00 2001 From: Claas Date: Mon, 19 Aug 2024 13:49:03 +0200 Subject: [PATCH] Shift responsibility by introducing a generic parameter --- fan-controller/src/main.rs | 160 +++++++++++++++--------------- fan-controller/src/mqtt/client.rs | 81 +++++++++------ fan-controller/src/mqtt/packet.rs | 7 ++ 3 files changed, 141 insertions(+), 107 deletions(-) diff --git a/fan-controller/src/main.rs b/fan-controller/src/main.rs index eea4e18..e30bded 100644 --- a/fan-controller/src/main.rs +++ b/fan-controller/src/main.rs @@ -37,17 +37,18 @@ use static_cell::StaticCell; use {defmt_rtt as _, panic_probe as _}; +use self::mqtt::packet; +use self::mqtt::packet::Packet; use crate::configuration::*; use crate::mqtt::connect::Connect; use crate::mqtt::connect_acknowledgement::ConnectReasonCode; +use crate::mqtt::packet::FromPublish; use crate::mqtt::ping_request::PingRequest; use crate::mqtt::publish::Publish; use crate::mqtt::subscribe::{Subscribe, Subscription}; -use crate::mqtt::{connect, publish, subscribe}; use crate::mqtt::QualityOfService; +use crate::mqtt::{connect, publish, subscribe}; use fan::FanClient; -use self::mqtt::packet; -use self::mqtt::packet::Packet; mod configuration; mod fan; @@ -237,7 +238,7 @@ async fn mqtt_task( //TODO refactor this into a retry policy or of the sorts let mut client = loop { let result = client - .connect(MQTT_BROKER_USERNAME, MQTT_BROKER_PASSWORD, KEEP_ALIVE) + .connect::<()>(MQTT_BROKER_USERNAME, MQTT_BROKER_PASSWORD, KEEP_ALIVE) .await; match result { @@ -280,7 +281,7 @@ async fn mqtt_task( join( async { loop { - match receiver.receive().await { + match receiver.receive::<()>().await { Ok(packet) => { info!("Received packet: {}", packet); @@ -455,10 +456,13 @@ async fn mqtt_routine<'a>(mut socket: TcpSocket<'a>) { // Ok(()) } -async fn listen_for_publishes( +async fn listen_for_publishes( mut reader: TcpReader<'_>, mut buffer: [u8; 1024], -) -> Result<(), PublishReceiveError> { +) -> Result<(), PublishReceiveError> +where + T: FromPublish, +{ loop { // Should read at least one byte let bytes_read = reader @@ -467,7 +471,7 @@ async fn listen_for_publishes( .map_err(PublishReceiveError::ReadError)?; let packet = - Packet::read(&buffer[..bytes_read]).map_err(PublishReceiveError::ReadPacketError)?; + Packet::::read(&buffer[..bytes_read]).map_err(PublishReceiveError::ReadPacketError)?; match packet { //TODO handle disconnect packet packet @ Packet::ConnectAcknowledgement(_) @@ -475,75 +479,75 @@ async fn listen_for_publishes( warn!("Received unexpected packet type: {}", packet.get_type()); } Packet::Publish(publish) => { - info!("Received publish: {:?}", publish); + info!("Received publish"); // This part is not MQTT and application specific - match publish.topic_name { - "testfan/speed/percentage" => { - let payload = match core::str::from_utf8(&publish.payload) { - Ok(payload) => payload, - Err(error) => { - warn!("Expected percentage_command_topic payload (speed percentage) to be a valid UTF-8 string with a number"); - continue; - } - }; - - // And then to an integer... - let set_point = payload.parse::(); - let set_point = match set_point { - Ok(set_point) => set_point, - Err(error) => { - warn!("Expected speed percentage to be a number string. Payload is: {}", payload); - continue; - } - }; - - let Ok(setting) = FanSetting::new(set_point) else { - warn!( - "Setting fan speed out of bounds. Not accepting new setting: {}", - set_point - ); - continue; - }; - - //TODO confirm there was no error on the modbus side - // MODBUS_SIGNAL.signal(setting); - } - "testfan/on/set" => { - let payload = match core::str::from_utf8(publish.payload) { - Ok(payload) => payload, - Err(error) => { - warn!( - "Expected command_topic payload to be a valid UTF-8 string with either \"ON\" or \"OFF\". Payload is: {}", - publish.payload - ); - continue; - } - }; - - match payload { - "ON" => { - //TODO set to last known state before off - println!("Turning on"); - } - "OFF" => { - println!("Turning off"); - // MODBUS_SIGNAL.signal(FanSetting::ZERO); - } - unknown => { - warn!( - "Expected either \"ON\" or \"OFF\" but received: {}", - unknown - ); - } - } - } - unknown => { - warn!( - "Unexpected topic: {} with payload: {}", - publish.topic_name, publish.payload - ); - } - } + // match publish.topic_name { + // "testfan/speed/percentage" => { + // let payload = match core::str::from_utf8(&publish.payload) { + // Ok(payload) => payload, + // Err(error) => { + // warn!("Expected percentage_command_topic payload (speed percentage) to be a valid UTF-8 string with a number"); + // continue; + // } + // }; + // + // // And then to an integer... + // let set_point = payload.parse::(); + // let set_point = match set_point { + // Ok(set_point) => set_point, + // Err(error) => { + // warn!("Expected speed percentage to be a number string. Payload is: {}", payload); + // continue; + // } + // }; + // + // let Ok(setting) = FanSetting::new(set_point) else { + // warn!( + // "Setting fan speed out of bounds. Not accepting new setting: {}", + // set_point + // ); + // continue; + // }; + // + // //TODO confirm there was no error on the modbus side + // // MODBUS_SIGNAL.signal(setting); + // } + // "testfan/on/set" => { + // let payload = match core::str::from_utf8(publish.payload) { + // Ok(payload) => payload, + // Err(error) => { + // warn!( + // "Expected command_topic payload to be a valid UTF-8 string with either \"ON\" or \"OFF\". Payload is: {}", + // publish.payload + // ); + // continue; + // } + // }; + // + // match payload { + // "ON" => { + // //TODO set to last known state before off + // println!("Turning on"); + // } + // "OFF" => { + // println!("Turning off"); + // // MODBUS_SIGNAL.signal(FanSetting::ZERO); + // } + // unknown => { + // warn!( + // "Expected either \"ON\" or \"OFF\" but received: {}", + // unknown + // ); + // } + // } + // } + // unknown => { + // warn!( + // "Unexpected topic: {} with payload: {}", + // publish.topic_name, publish.payload + // ); + // } + // } } } } @@ -681,10 +685,8 @@ async fn input_task(pin_18: PIN_18) { warn!("No fan client found"); continue; }; - + fan.set_set_point(set_point).await; - - } } diff --git a/fan-controller/src/mqtt/client.rs b/fan-controller/src/mqtt/client.rs index 94c311c..2ef9424 100644 --- a/fan-controller/src/mqtt/client.rs +++ b/fan-controller/src/mqtt/client.rs @@ -7,14 +7,6 @@ use core::marker::PhantomData; use embassy_net::tcp::{self, TcpReader, TcpSocket, TcpWriter}; use embassy_time::{with_timeout, Duration, TimeoutError}; -pub(crate) struct MqttClient<'a, S> { - _state: PhantomData, - socket: TcpSocket<'a>, - // No experience how big this buffer should be - send_buffer: [u8; 256], - receive_buffer: [u8; 1024], - timeout: Duration, -} pub(crate) struct NotConnected; pub(crate) struct Connected; @@ -37,6 +29,14 @@ pub(crate) enum ReadError { Timeout(TimeoutError), } +pub(crate) struct MqttClient<'a, S> { + _state: PhantomData, + socket: TcpSocket<'a>, + // No experience how big this buffer should be + send_buffer: [u8; 256], + receive_buffer: [u8; 1024], + timeout: Duration, +} impl<'a, S> MqttClient<'a, S> { async fn read(&mut self) -> Result<&[u8], ReadError> { let read = self.socket.read(&mut self.receive_buffer); @@ -49,7 +49,7 @@ impl<'a, S> MqttClient<'a, S> { } } -impl<'a, T> MqttClient<'a, NotConnected> where T : FromPublish { +impl<'a> MqttClient<'a, NotConnected> { pub(crate) const fn new(socket: TcpSocket<'a>) -> Self { Self { socket, @@ -61,12 +61,15 @@ impl<'a, T> MqttClient<'a, NotConnected> where T : FromPublish { } /// The server must not send any data before send - pub(crate) async fn connect( + pub(crate) async fn connect( mut self, username: &str, password: &[u8], keep_alive: Duration, - ) -> Result, (Self, ConnectError)> { + ) -> Result, (Self, ConnectError)> + where + T: FromPublish, + { // Send MQTT connect packet let packet = Connect { client_identifier: "testfan", @@ -111,7 +114,7 @@ impl<'a, T> MqttClient<'a, NotConnected> where T : FromPublish { return Err((self, ConnectError::ErrorCode(error_code))); } - Ok(MqttClient { + Ok(MqttClient:: { socket: self.socket, _state: PhantomData, send_buffer: self.send_buffer, @@ -166,7 +169,10 @@ impl<'a> MqttReceiver<'a> { Ok(&self.receive_buffer[..bytes_read]) } - pub(crate) async fn receive(&mut self) -> Result { + pub(crate) async fn receive(&mut self) -> Result, ReceiveError> + where + T: FromPublish, + { let result = self.read().await; let bytes = match result { Ok(bytes) => bytes, @@ -179,7 +185,7 @@ impl<'a> MqttReceiver<'a> { mod runner { use crate::mqtt::connect::Connect; - use crate::mqtt::packet::Packet; + use crate::mqtt::packet::{FromPublish, Packet}; use crate::mqtt::{connect, packet, ConnectErrorReasonCode}; use embassy_net::tcp; use embassy_net::tcp::{TcpReader, TcpSocket, TcpWriter}; @@ -187,9 +193,12 @@ mod runner { use embassy_sync::pubsub::{PubSubChannel, Publisher, Subscriber}; use embassy_time::{with_timeout, Duration, TimeoutError}; - type ReceiveResult<'a> = Result, ReceiveError>; - struct State<'a> { - channel: PubSubChannel, 8, 1, 1>, + type ReceiveResult = Result, ReceiveError>; + struct State + where + T: FromPublish + Clone, + { + channel: PubSubChannel, 8, 1, 1>, } #[derive(Clone)] @@ -198,15 +207,21 @@ mod runner { DecodePacketError(packet::ReadError), } - struct MqttRunner<'a> { + struct MqttRunner<'a, T> + where + T: FromPublish + Clone, + { tcp_receiver: TcpReader<'a>, /// Publisher to send received packets to the client that waits for them - publisher: Publisher<'a, CriticalSectionRawMutex, ReceiveResult<'a>, 8, 1, 1>, + publisher: Publisher<'a, CriticalSectionRawMutex, ReceiveResult, 8, 1, 1>, receive_buffer: [u8; 1024], timeout: Duration, } - impl<'a> MqttRunner<'a> { + impl<'a, T> MqttRunner<'a, T> + where + T: FromPublish + Clone, + { pub(self) async fn read(&mut self) -> Result<&[u8], ReadError> { let read = self.tcp_receiver.read(&mut self.receive_buffer); let bytes_read = with_timeout(self.timeout, read) @@ -229,8 +244,7 @@ mod runner { // }; let read = self.tcp_receiver.read(&mut self.receive_buffer); - let bytes_read = with_timeout(self.timeout, read) - .await.unwrap().unwrap(); + let bytes_read = with_timeout(self.timeout, read).await.unwrap().unwrap(); let bytes = &self.receive_buffer[..bytes_read]; let result = Packet::read(bytes); @@ -238,7 +252,9 @@ mod runner { match result { Ok(packet) => self.publisher.publish(Ok(packet)).await, Err(error) => { - self.publisher.publish(Err(ReceiveError::DecodePacketError(error))).await; + self.publisher + .publish(Err(ReceiveError::DecodePacketError(error))) + .await; continue; } } @@ -265,17 +281,26 @@ mod runner { ErrorCode(ConnectErrorReasonCode), } - struct MqttClient<'a> { - subscriber: Subscriber<'a, CriticalSectionRawMutex, ReceiveResult<'a>, 8, 1, 1>, + struct MqttClient<'a, T> + where + T: FromPublish + Clone, + { + subscriber: Subscriber<'a, CriticalSectionRawMutex, ReceiveResult, 8, 1, 1>, tcp_writer: TcpWriter<'a>, send_buffer: [u8; 256], } - impl<'a> MqttClient<'a> { + impl<'a, T> MqttClient<'a, T> + where + T: FromPublish + Clone, + { pub fn new( - state: &'a State<'a>, + state: &'a State, mut socket: &'a mut TcpSocket<'a>, - ) -> (MqttRunner<'a>, Self) { + ) -> (MqttRunner<'a, T>, Self) + where + T: FromPublish + Clone, + { let (reader, writer) = socket.split(); //TODO handle out of subscribers/publishers error let publisher = state.channel.publisher().unwrap(); diff --git a/fan-controller/src/mqtt/packet.rs b/fan-controller/src/mqtt/packet.rs index 38954b2..5211d8b 100644 --- a/fan-controller/src/mqtt/packet.rs +++ b/fan-controller/src/mqtt/packet.rs @@ -97,3 +97,10 @@ where pub(crate) trait FromPublish { fn from_publish(publish: Publish) -> Self; } + +/// Temporary just for testing. TODO remove +impl FromPublish for () { + fn from_publish(publish: Publish) -> Self { + () + } +} \ No newline at end of file -- 2.51.2