diff --git a/fan-controller/src/main.rs b/fan-controller/src/main.rs index e30bded..7d85613 100644 --- a/fan-controller/src/main.rs +++ b/fan-controller/src/main.rs @@ -321,8 +321,8 @@ async fn mqtt_task( }, ]; - const SUBSCRIBE_PACKET: Subscribe<2> = Subscribe { - subscriptions: SUBSCRIPTIONS, + const SUBSCRIBE_PACKET: Subscribe = Subscribe { + subscriptions: &SUBSCRIPTIONS, packet_identifier: 42, }; } @@ -470,8 +470,8 @@ where .await .map_err(PublishReceiveError::ReadError)?; - let packet = - Packet::::read(&buffer[..bytes_read]).map_err(PublishReceiveError::ReadPacketError)?; + let packet = Packet::::read(&buffer[..bytes_read]) + .map_err(PublishReceiveError::ReadPacketError)?; match packet { //TODO handle disconnect packet packet @ Packet::ConnectAcknowledgement(_) @@ -490,7 +490,7 @@ where // continue; // } // }; - // + // // // And then to an integer... // let set_point = payload.parse::(); // let set_point = match set_point { @@ -500,7 +500,7 @@ where // continue; // } // }; - // + // // let Ok(setting) = FanSetting::new(set_point) else { // warn!( // "Setting fan speed out of bounds. Not accepting new setting: {}", @@ -508,7 +508,7 @@ where // ); // continue; // }; - // + // // //TODO confirm there was no error on the modbus side // // MODBUS_SIGNAL.signal(setting); // } @@ -523,7 +523,7 @@ where // continue; // } // }; - // + // // match payload { // "ON" => { // //TODO set to last known state before off diff --git a/fan-controller/src/mqtt/client.rs b/fan-controller/src/mqtt/client.rs index 0ca5fe5..f2be223 100644 --- a/fan-controller/src/mqtt/client.rs +++ b/fan-controller/src/mqtt/client.rs @@ -187,13 +187,15 @@ mod runner { use crate::mqtt::connect::Connect; use crate::mqtt::connect_acknowledgement::{ConnectAcknowledgement, ConnectReasonCode}; use crate::mqtt::packet::{FromPublish, Packet}; - use crate::mqtt::{connect, packet, ConnectErrorReasonCode}; - use defmt::warn; + use crate::mqtt::subscribe::{Subscribe, Subscription}; + use crate::mqtt::subscribe_acknowledgement::SubscribeAcknowledgement; + use crate::mqtt::{connect, packet, subscribe, ConnectErrorReasonCode}; + use defmt::{warn, Format}; use embassy_net::tcp; use embassy_net::tcp::{TcpReader, TcpSocket, TcpWriter}; use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex; use embassy_sync::pubsub::{PubSubChannel, Publisher, Subscriber, WaitResult}; - use embassy_time::{with_timeout, Duration, TimeoutError}; + use embassy_time::{with_deadline, with_timeout, Duration, Instant, TimeoutError}; type ReceiveResult = Result, ReceiveError>; struct State @@ -203,7 +205,7 @@ mod runner { channel: PubSubChannel, 8, 1, 1>, } - #[derive(Clone, Debug)] + #[derive(Clone, Debug, Format)] enum ReceiveError { ReadError(ReadError), DecodePacketError(packet::ReadError), @@ -261,7 +263,7 @@ mod runner { } } - #[derive(Debug, Clone)] + #[derive(Debug, Clone, Format)] pub(crate) enum ReadError { TcpReadError(tcp::Error), Timeout(TimeoutError), @@ -275,6 +277,14 @@ mod runner { ReceiveError(ReceiveError), /// Received a connect acknowledgement with an error code ErrorCode(ConnectErrorReasonCode), + Timeout(TimeoutError), + } + + pub(crate) enum SubscribeError { + WriteSubscribeError(subscribe::WriteError), + TcpWriteError(tcp::Error), + TcpFlushError(tcp::Error), + Timeout(TimeoutError), } struct MqttClient<'a, T> @@ -284,6 +294,7 @@ mod runner { subscriber: Subscriber<'a, CriticalSectionRawMutex, ReceiveResult, 8, 1, 1>, tcp_writer: TcpWriter<'a>, send_buffer: [u8; 256], + timeout: Duration, } impl<'a, T> MqttClient<'a, T> @@ -301,18 +312,20 @@ mod runner { //TODO handle out of subscribers/publishers error let publisher = state.channel.publisher().unwrap(); let subscriber = state.channel.subscriber().unwrap(); + let timeout = Duration::from_secs(5); ( MqttRunner { publisher, tcp_receiver: reader, receive_buffer: [0; 1024], - timeout: Duration::from_secs(5), + timeout: timeout.clone(), }, Self { subscriber, tcp_writer: writer, send_buffer: [0; 256], + timeout, }, ) } @@ -331,6 +344,7 @@ mod runner { keep_alive_seconds: keep_alive.as_secs() as u16, }; + // Send let mut offset = 0; if let Err(error) = packet.write(&mut self.send_buffer, &mut offset) { return Err((ConnectError::WriteConnectError(error))); @@ -344,12 +358,21 @@ mod runner { return Err((ConnectError::TcpFlushError(error))); } + // Receive + // Wait for connect acknowledgement // Discard all messages before the connect acknowledgement // The server has to send a connect acknowledgement before sending any other packet - // TCP should ensure the order of packets, so they should not arrive out of order + // TCP should ensure the order of packets (afaik), so they should not arrive out of order + // Using a deadline because the loop could be run multiple times + let deadline = Instant::now() + self.timeout; loop { - let result = self.subscriber.next_message().await; + let result = with_deadline(deadline, self.subscriber.next_message()).await; + + let result = match result { + Ok(result) => result, + Err(error) => return Err(ConnectError::Timeout(error)), + }; match result { WaitResult::Lagged(missed_messages) => { @@ -376,5 +399,74 @@ mod runner { } } } + + pub(crate) async fn subscribe<'s>( + &mut self, + subscriptions: &[Subscription<'s>], + ) -> Result<(), SubscribeError> { + //TODO add packet identifier management when there can be more than one subscribe in flight + // That requires a non-mutable reference on self though. Which means putting the tcp writer behind a channel or mutex(?) + let packet = Subscribe { + subscriptions, + packet_identifier: 42, + }; + + // Send + let mut offset = 0; + if let Err(error) = packet.write(&mut self.send_buffer, &mut offset) { + return Err((SubscribeError::WriteSubscribeError(error))); + }; + + if let Err(error) = self.tcp_writer.write(&self.send_buffer[..offset]).await { + return Err((SubscribeError::TcpWriteError(error))); + } + + if let Err(error) = self.tcp_writer.flush().await { + return Err((SubscribeError::TcpFlushError(error))); + } + + // Receive + let deadline = Instant::now() + self.timeout; + loop { + let result = with_deadline(deadline, self.subscriber.next_message()).await; + + let result = match result { + Ok(result) => result, + Err(error) => return Err(SubscribeError::Timeout(error)), + }; + + match result { + WaitResult::Lagged(missed_messages) => { + warn!( + "Missed {} messages while waiting for subscribe acknowledgement: ", + missed_messages + ); + } + WaitResult::Message(result) => { + // Ignoring all errors because we don't know if they were as a result of + // sending the subscribe packet + let packet = match result { + Err(error) => { + warn!( + "Error while waiting for subscribe acknowledgement: {:?}", + error + ); + continue; + } + Ok(packet) => packet, + }; + + // Ignore all messages not intended for us + let Packet::SubscribeAcknowledgement(SubscribeAcknowledgement { + //TODO how to pass arbitrary amount of subscribe topics back? + }) = packet else { + continue; + }; + } + } + } + + defmt::todo!() + } } } diff --git a/fan-controller/src/mqtt/mod.rs b/fan-controller/src/mqtt/mod.rs index fe6b24e..7418702 100644 --- a/fan-controller/src/mqtt/mod.rs +++ b/fan-controller/src/mqtt/mod.rs @@ -42,7 +42,7 @@ pub(super) enum ConnectErrorReasonCode { ConnectionRateExceeded = 0x9F, } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Format)] pub struct UnknownConnectErrorReasonCode(u8); pub(super) enum QualityOfService { @@ -64,7 +64,7 @@ impl QualityOfService { } } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Format)] enum ReadConnectAcknowledgementError { InvalidReasonCode(UnknownConnectErrorReasonCode), InvalidPropertiesLength(VariableByteIntegerDecodeError), diff --git a/fan-controller/src/mqtt/packet.rs b/fan-controller/src/mqtt/packet.rs index 5211d8b..2573bdd 100644 --- a/fan-controller/src/mqtt/packet.rs +++ b/fan-controller/src/mqtt/packet.rs @@ -9,7 +9,7 @@ use crate::mqtt::variable_byte_integer::VariableByteIntegerDecodeError; use crate::mqtt::{publish, variable_byte_integer, ReadConnectAcknowledgementError}; use defmt::Format; -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Format)] pub(crate) enum ReadError { /// The packet type is not supported. This can happen if there is a packet received that is /// only intended for the broker and not the client. Or the packet type is not yet implemented. @@ -71,8 +71,7 @@ where let packet = T::from_publish(publish); Ok(Packet::Publish(packet)) } - // N doesn't matter here - Subscribe::<2>::TYPE => Err(ReadError::UnsupportedPacketType(packet_type)), + Subscribe::TYPE => Err(ReadError::UnsupportedPacketType(packet_type)), SubscribeAcknowledgement::TYPE => { let subscribe_acknowledgement = SubscribeAcknowledgement::read(variable_header_and_payload) @@ -103,4 +102,4 @@ impl FromPublish for () { fn from_publish(publish: Publish) -> Self { () } -} \ No newline at end of file +} diff --git a/fan-controller/src/mqtt/publish.rs b/fan-controller/src/mqtt/publish.rs index 0d3dfec..fe6f890 100644 --- a/fan-controller/src/mqtt/publish.rs +++ b/fan-controller/src/mqtt/publish.rs @@ -1,6 +1,6 @@ use core::str::Utf8Error; -use defmt::Format; +use defmt::{write, Debug2Format, Format, Formatter}; use crate::mqtt::variable_byte_integer; use crate::mqtt::variable_byte_integer::{ @@ -26,6 +26,25 @@ pub(crate) enum ReadError { InvalidTopicName(Utf8Error), } +impl Format for ReadError { + fn format(&self, fmt: Formatter) { + match self { + ReadError::UnsupportedQualityOfServiceLevel(error) => { + write!(fmt, "Unsupported quality of service level: {}", error) + } + ReadError::VariableByteIntegerError(error) => { + write!(fmt, "Variable byte integer error: {}", error) + } + ReadError::ZeroLengthTopicName => { + write!(fmt, "Zero length topic name") + } + ReadError::InvalidTopicName(error) => { + write!(fmt, "Invalid topic name: {:?}", Debug2Format(error)) + } + } + } +} + impl<'a> Publish<'a> { pub(crate) const TYPE: u8 = 3; diff --git a/fan-controller/src/mqtt/subscribe.rs b/fan-controller/src/mqtt/subscribe.rs index 1b64d52..48d4edb 100644 --- a/fan-controller/src/mqtt/subscribe.rs +++ b/fan-controller/src/mqtt/subscribe.rs @@ -72,12 +72,12 @@ pub(crate) enum WriteError { RemainingLengthError(variable_byte_integer::VariableByteIntegerEncodeError), } -pub(crate) struct Subscribe<'a, const N: usize> { - pub(crate) subscriptions: [Subscription<'a>; N], +pub(crate) struct Subscribe<'a> { + pub(crate) subscriptions: &'a [Subscription<'a>], pub(crate) packet_identifier: u16, } -impl<'a, const N: usize> Subscribe<'a, N> { +impl<'a> Subscribe<'a> { pub(crate) const TYPE: u8 = 8; pub(crate) fn write(self, buffer: &mut [u8], offset: &mut usize) -> Result<(), WriteError> { let variable_header_length = size_of::() + size_of::(); diff --git a/fan-controller/src/mqtt/subscribe_acknowledgement.rs b/fan-controller/src/mqtt/subscribe_acknowledgement.rs index aeb8a9d..3f0cf05 100644 --- a/fan-controller/src/mqtt/subscribe_acknowledgement.rs +++ b/fan-controller/src/mqtt/subscribe_acknowledgement.rs @@ -2,7 +2,7 @@ use crate::mqtt::variable_byte_integer; use crate::mqtt::variable_byte_integer::VariableByteIntegerDecodeError; use defmt::Format; -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Format)] pub(crate) enum SubscribeAcknowledgementError { InvalidPropertiesLength(VariableByteIntegerDecodeError), } @@ -22,17 +22,17 @@ impl SubscribeAcknowledgement { //TODO stop ignoring properties //TODO check if topics are acknowledged - // offset += properties_length; - // - // // Payload - // // Reason code for each subscribed topic in the same order - // let mut topic_index = 0; - // - // loop { - // let reason_code = buffer[offset]; - // offset += 1; - // - // } + offset += properties_length; + + // Payload + // Reason code for each subscribed topic in the same order + let mut topic_index = 0; + + loop { + let reason_code = buffer[offset]; + offset += 1; + + } Ok(SubscribeAcknowledgement) }