diff --git a/mqtt/src/main.rs b/mqtt/src/main.rs index 6df1808..6957c1c 100644 --- a/mqtt/src/main.rs +++ b/mqtt/src/main.rs @@ -4,6 +4,7 @@ mod variable_byte_integer; use crate::packet::{QualityOfService, RetainHandling, Subscription, SubscriptionOptions}; use bytes::BytesMut; use packet::{ConnectErrorReasonCode, ConnectReasonCode, PacketType, UnknownConnectReasonCode}; +use std::collections::HashMap; use std::env; use std::net::{AddrParseError, ToSocketAddrs}; use std::sync::Arc; @@ -58,20 +59,6 @@ async fn set_up_tls_connection( Ok(stream) } -enum MqttActorMessage { - Connect { - client_identifier: Arc, - username: Arc, - password: Arc<[u8]>, - responder: oneshot::Sender>, - }, -} - -struct MqttActor { - receiver: mpsc::Receiver, - stream: TcpStream, -} - /// Errors while reading the Connect Acknowledgement packet (CONNACK) #[derive(Debug, thiserror::Error)] enum ConnectAcknowledgementError { @@ -91,9 +78,57 @@ enum ConnectAcknowledgementError { #[derive(Debug)] enum ConnectError { SendError(std::io::Error), + FlushError(std::io::Error), AcknowledgementError(ConnectAcknowledgementError), } +/// Errors while reading the Subscribe Acknowledgement packet (SUBACK) +#[derive(Debug, thiserror::Error)] +enum SubscribeAcknowledgementError { + #[error("Error while reading from the stream: {0}")] + ReadError(#[from] std::io::Error), + /// The server sent a different packet type than connect acknowledgement (2) + #[error("Unexpected packet type: {0:#04x}")] + UnexpectedPacketType(u8), + #[error("Invalid remaining length: {0}")] + InvalidRemainingLength(DecodeVariableByteIntegerError), + #[error("Invalid properties length: {0}")] + InvalidPropertiesLength(DecodeVariableByteIntegerError), + #[error("Invalid reason code length: expected {expected} topic(s), got {actual}")] + InvalidReasonCodeLength { expected: usize, actual: usize }, +} + +#[derive(Debug)] +enum SubscribeError { + SendError(std::io::Error), + FlushError(std::io::Error), + AcknowledgementError(SubscribeAcknowledgementError), +} + +enum MqttActorMessage { + Connect { + client_identifier: Arc, + username: Arc, + password: Arc<[u8]>, + responder: oneshot::Sender>, + }, + Publish { + topic: Arc, + payload: Arc<[u8]>, + responder: oneshot::Sender>, + }, + Subscribe { + subscriptions: Arc<[Subscription]>, + responder: oneshot::Sender>, + }, +} + +struct MqttActor { + receiver: mpsc::Receiver, + stream: TcpStream, + send_publish: mpsc::Sender<()>, +} + async fn read_connect_acknowledgement( stream: &mut TcpStream, ) -> Result<(), ConnectAcknowledgementError> { @@ -149,6 +184,8 @@ async fn connect( .await .map_err(ConnectError::SendError)?; + stream.flush().await.map_err(ConnectError::FlushError)?; + // We read the acknowledgemetn "synchronous" as we can't receive or send other packets before the connection is established // Response for connection has to come in next @@ -157,6 +194,100 @@ async fn connect( .map_err(ConnectError::AcknowledgementError) } +async fn read_subscribe_acknowledgement( + stream: &mut TcpStream, + subscriptions: Arc<[Subscription]>, +) -> Result<(), SubscribeAcknowledgementError> { + // Fixed header + let packet_type_and_flags = stream.read_u8().await?; + if (packet_type_and_flags >> 4) != 9 { + return Err(SubscribeAcknowledgementError::UnexpectedPacketType( + packet_type_and_flags >> 4, + )); + } + + let (mut remaining_length, _remaining_length_length) = VariableByteInteger::decode(stream) + .await + .map_err(SubscribeAcknowledgementError::InvalidRemainingLength)?; + + let mut buffer = Vec::with_capacity(remaining_length); + // Could warn if we read less or more than remaining_length + let _bytes_read = stream.read_buf(&mut buffer).await?; + + // Variable header + let packet_identifier: u16 = ((buffer[0] as u16) << 8) | buffer[1] as u16; + println!("Packet identifier {packet_identifier}"); + remaining_length -= 2; + println!("Remaining length {remaining_length}"); + + // Properties + let (properties_length, properties_length_length) = + VariableByteInteger::decode_2(buffer.as_ref()) + .map_err(SubscribeAcknowledgementError::InvalidPropertiesLength)?; + println!("Properties length {properties_length}"); + remaining_length -= properties_length_length; + println!("Remaining length {remaining_length}"); + //TODO read properties + + // Payload + // Reason code for each subscribed topic in the same order + let mut topic_index = 0; + if remaining_length != subscriptions.len() { + return Err(SubscribeAcknowledgementError::InvalidReasonCodeLength { + expected: subscriptions.len(), + actual: remaining_length, + }); + } + + // Index should be aligned as we checked length before + while remaining_length > 0 { + let reason_code = buffer[2 + properties_length_length + topic_index]; + remaining_length -= 1; + + let topic = &subscriptions[topic_index].topic_filter; + topic_index += 1; + println!("Reason code for topic {topic}: {reason_code:#04x}"); + println!("Remaining length {remaining_length}"); + } + + Ok(()) +} + +async fn publish( + stream: &mut TcpStream, + topic: Arc, + payload: Arc<[u8]>, +) -> Result<(), std::io::Error> { + let packet = packet::create_publish(topic.as_ref(), payload.as_ref()); + + // Fire and forget with quality of service 0 but we can at least confirm locally if the packet was sent + stream.write_all(&packet).await?; + stream.flush().await?; + + // We don't wait for the PUBACK packet if there is no Quality of Service (QoS) configured + // If there was QoS it, then we would need to assign a packet identifier and store the channel for as long as we wait for the PUBACK + Ok(()) +} + +async fn subscribe( + stream: &mut TcpStream, + subscriptions: Arc<[Subscription]>, +) -> Result<(), SubscribeError> { + //TODO manage available identifiers + let packet = packet::create_subscribe(9, subscriptions.as_ref()); + + stream + .write_all(&packet) + .await + .map_err(SubscribeError::SendError)?; + stream.flush().await.map_err(SubscribeError::FlushError)?; + + // We wait for the subscribe acknowledgement but we should probably do that "asynchronously" + read_subscribe_acknowledgement(stream, subscriptions) + .await + .map_err(SubscribeError::AcknowledgementError) +} + async fn process_message(message: MqttActorMessage, stream: &mut TcpStream) { match message { MqttActorMessage::Connect { @@ -167,28 +298,47 @@ async fn process_message(message: MqttActorMessage, stream: &mut TcpStream) { } => { let result = connect(stream, client_identifier, username, password).await; + // Ignore error if they cancelled waiting for the response + let _ = responder.send(result); + } + MqttActorMessage::Publish { + topic, + payload, + responder, + } => { + let result = publish(stream, topic, payload).await; + + // Ignore error if they cancelled waiting for the response + let _ = responder.send(result); + } + MqttActorMessage::Subscribe { + subscriptions, + responder, + } => { + let result = subscribe(stream, subscriptions).await; + // Ignore error if they cancelled waiting for the response let _ = responder.send(result); } } } -fn process_mqtt_message(buffer: &Vec) { - todo!("Process MQTT message") +async fn process_packet(first_byte: u8, actor: &mut MqttActor) { + println!("Received packet: {}", first_byte >> 4); + actor.send_publish.send(()).await.unwrap(); } async fn run_actor(mut actor: MqttActor) { - let mut buffer = Vec::new(); loop { // let a = actor.stream.read_to_end(&mut buffer).await; tokio::select! { Some(message) = actor.receiver.recv() => process_message(message, &mut actor.stream).await, - Ok(_bytes_read) = actor.stream.read_to_end(&mut buffer) => process_mqtt_message(&buffer), + Ok(first_byte) = actor.stream.read_u8() => process_packet(first_byte, &mut actor).await, else => break, } - - buffer.clear(); } + + println!("Actor done"); } struct MqttActorHandle { @@ -203,9 +353,13 @@ enum HandleError { } impl MqttActorHandle { - fn new(stream: TcpStream) -> Self { + fn new(stream: TcpStream, send_publish: mpsc::Sender<()>) -> Self { let (sender, receiver) = mpsc::channel(8); - let actor = MqttActor { stream, receiver }; + let actor = MqttActor { + stream, + receiver, + send_publish, + }; tokio::spawn(run_actor(actor)); @@ -241,6 +395,60 @@ impl MqttActorHandle { .map_err(Error::ReceiveError)? .map_err(Error::ActorError) } + + async fn publish( + &self, + topic: Arc, + payload: Arc<[u8]>, + ) -> Result<(), HandleError> + { + type Error = HandleError< + mpsc::error::SendError, + oneshot::error::RecvError, + std::io::Error, + >; + let (sender, receiver) = oneshot::channel(); + self.sender + .send(MqttActorMessage::Publish { + topic, + payload, + responder: sender, + }) + .await + .map_err(Error::SendError)?; + + receiver + .await + .map_err(Error::ReceiveError)? + .map_err(Error::ActorError) + } + + async fn subscribe( + &self, + subscriptions: Arc<[Subscription]>, + ) -> Result<(), HandleError> + { + let (sender, receiver) = oneshot::channel(); + + type Error = HandleError< + mpsc::error::SendError, + oneshot::error::RecvError, + SubscribeError, + >; + + self.sender + .send(MqttActorMessage::Subscribe { + subscriptions, + responder: sender, + }) + .await + .map_err(Error::SendError)?; + + receiver + .await + .map_err(Error::ReceiveError)? + .map_err(Error::ActorError) + } } async fn set_up_tcp_connection( @@ -267,9 +475,12 @@ async fn main() -> Result<(), AppError> { // The password might need to be surrounded by single quotes (e.g. 'password') to be read correctly let password = env::var("MQTT_BROKER_PASSWORD")?; + // Channel to receive publish messages + let (sender, mut receiver) = mpsc::channel(8); + // Could directly set it up in the handle new but eh let stream = set_up_tcp_connection(broker_address, broker_port).await?; - let actor = MqttActorHandle::new(stream); + let actor = MqttActorHandle::new(stream, sender); actor .connect( @@ -279,8 +490,72 @@ async fn main() -> Result<(), AppError> { ) .await .unwrap(); + // homeassistant/{domain}/{object_id}/config + + // Prefix is "homeassistant", but it can be changed in home assistant configuration + const DISCOVERY_TOPIC: &str = "homeassistant/fan/testfan/config"; + + // Configuration is like the YAML configuration that would be added in Home Assistant but as JSON + // Command topic: The MQTT topic to publish commands to change the state of the fan + //TODO set firmware version from Cargo.toml package version + //TODO think about setting hardware version, support url, and manufacturer + //TODO create single home assistant device with multiple entities for sensors in fan and the bypass + //TODO add diagnostic entity like IP address + //TODO availability topic + const DISCOVERY_PAYLOAD: &[u8] = br#"{ + "name": "Fan", + "unique_id": "testfan", + "state_topic": "testfan/on/state", + "command_topic": "testfan/on/set", + "percentage_state_topic": "testfan/speed/percentage_state", + "percentage_command_topic": "testfan/speed/percentage", + + "speed_range_min": 1, + "speed_range_max": 64000, + "qos": 0, + "optimistic": true + }"#; println!("Connected"); + + // Subscribing to the state topics before telling home assistant to use them to control the fan + + // Send discover publish packet + actor + .publish(DISCOVERY_TOPIC.into(), DISCOVERY_PAYLOAD.into()) + .await + .unwrap(); + + println!("Published discovery"); + let subscriptions = Arc::new([ + // Listen to when the fan should be turned on or off + // Payload will be "ON" or "OFF" + Subscription::new( + "testfan/on/set".into(), + SubscriptionOptions::new( + QualityOfService::AtMostOnceDelivery, + false, + false, + RetainHandling::DoNotSend, + ), + ), + // Listen to speed changes from home assistant + Subscription::new( + "testfan/speed/percentage".into(), + SubscriptionOptions::new( + QualityOfService::AtMostOnceDelivery, + false, + false, + RetainHandling::DoNotSend, + ), + ), + ]); + actor.subscribe(subscriptions).await.unwrap(); + + while let Some(_) = receiver.recv().await { + println!("Received message") + } + Ok(()) } @@ -408,7 +683,7 @@ async fn old_code() -> Result<(), AppError> { // Listen to when the fan should be turned on or off // Payload will be "ON" or "OFF" Subscription::new( - "testfan/on/set", + "testfan/on/set".into(), SubscriptionOptions::new( QualityOfService::AtMostOnceDelivery, false, @@ -418,7 +693,7 @@ async fn old_code() -> Result<(), AppError> { ), // Listen to speed changes from home assistant Subscription::new( - "testfan/speed/percentage", + "testfan/speed/percentage".into(), SubscriptionOptions::new( QualityOfService::AtMostOnceDelivery, false, @@ -462,7 +737,7 @@ async fn old_code() -> Result<(), AppError> { remaining_length -= 1; println!("Remaining length: {}", remaining_length); - //TODO figure out what this trailing byte in the subscribe acknolwedgement is + //TODO this is the other reason codes for each topic let whut = stream.read_u8().await?; println!("what: {}", whut); // Packet type of next packet diff --git a/mqtt/src/packet.rs b/mqtt/src/packet.rs index 1fdf1a7..8842805 100644 --- a/mqtt/src/packet.rs +++ b/mqtt/src/packet.rs @@ -1,4 +1,4 @@ -use std::convert::TryFrom; +use std::{convert::TryFrom, sync::Arc}; pub(super) fn create_connect(client_identifier: &str, username: &str, password: &[u8]) -> Vec { let identifier_length = client_identifier.len() as u16; @@ -194,20 +194,20 @@ impl SubscriptionOptions { } } -pub(super) struct Subscription<'a> { - topic_filter: &'a str, +pub(super) struct Subscription { + pub(super) topic_filter: Arc, options: SubscriptionOptions, } -impl<'a> Subscription<'a> { - pub(super) const fn new(topic_filter: &'a str, options: SubscriptionOptions) -> Self { +impl Subscription { + pub(super) fn new(topic_filter: Arc, options: SubscriptionOptions) -> Self { Self { options, topic_filter, } } - const fn length(&self) -> usize { + fn length(&self) -> usize { self.topic_filter.len() + size_of::() } } diff --git a/mqtt/src/variable_byte_integer.rs b/mqtt/src/variable_byte_integer.rs index 5475afd..894e93b 100644 --- a/mqtt/src/variable_byte_integer.rs +++ b/mqtt/src/variable_byte_integer.rs @@ -145,6 +145,7 @@ impl VariableByteInteger { Ok(Self(output)) } + //TODO clean up having all these decode and encode versions pub(super) async fn decode( stream: &mut TcpStream, ) -> Result<(usize, usize), DecodeVariableByteIntegerError> { @@ -180,6 +181,39 @@ impl VariableByteInteger { Ok((value, index)) } + pub(super) fn decode_2(bytes: &[u8]) -> Result<(usize, usize), DecodeVariableByteIntegerError> { + let mut multiplier = 1; + let mut value = 0; + let mut index = 0; + + loop { + let encoded_byte = bytes[index]; + index += 1; + + value += (encoded_byte & 127) as usize * multiplier; + + if multiplier > 128 * 128 * 128 { + return Err(DecodeVariableByteIntegerError::MalformedVariableByteIntegerError); + } + + multiplier *= 128; + + // The last byte has the most significant bit set to 0 indicating that there are no more bytes to follow + if (encoded_byte & 128) == 0 { + break; + } + + // The current byte indicates the next byte is part of the integer but we would be past 4 bytes + if index > 4 { + return Err(DecodeVariableByteIntegerError::InvalidLength); + } + } + + // As index is immediately increased it reflects the length at this point + // index is usize but can't pass the value 4 + Ok((value, index)) + } + /// Returns the length of the encoded integer in bytes which is either 1, 2, 3 or 4. /// This helps to determine how many bytes are needed on the wire to represent this integer. const fn length(&self) -> NonZero {