smart-home/base-station/software/src/mqtt.rs

use std::io::Write;
use std::net::TcpStream;
use std::thread;

use log::{error, info};
use mqtt::control::variable_header::ConnectReturnCode;
use mqtt::packet::*;
use mqtt::{Decodable, Encodable, QualityOfService};
use mqtt::{TopicFilter, TopicName};

use super::Error;

pub struct MQTT {
    stream: TcpStream,
}

impl MQTT {
    pub fn connect(server_addr: &str) -> Result<MQTT, Error> {
        info!("Connecting to MQTT broker at {}...", server_addr);
        let mut stream = TcpStream::connect(server_addr)?;

        let mut conn = ConnectPacket::new("MQTT", "72711cd2-faaf-44f0-8490-b92eb063ff4b");
        conn.set_clean_session(true);
        let mut buf = Vec::new();
        conn.encode(&mut buf).unwrap();
        stream.write_all(&buf[..])?;

        let connack = ConnackPacket::decode(&mut stream).unwrap();
        if connack.connect_return_code() != ConnectReturnCode::ConnectionAccepted {
            panic!(
                "Failed to connect to server, return code {:?}",
                connack.connect_return_code()
            );
        }
        info!("Connected to MQTT broker.");

        // We need to listen for packets from the server to reply to ping requests.
        // TODO: More protocol handling.
        let mut cloned_stream = stream.try_clone().unwrap();
        thread::spawn(move || {
            loop {
                let packet = match VariablePacket::decode(&mut cloned_stream) {
                    Ok(pk) => pk,
                    Err(err) => {
                        error!("could not decode received packet: {:?}", err);
                        continue;
                    }
                };

                match packet {
                    VariablePacket::PingreqPacket(..) => {
                        let pingresp = PingrespPacket::new();
                        info!("Sending Ping response {:?}", pingresp);
                        pingresp.encode(&mut cloned_stream).unwrap();
                    }
                    VariablePacket::DisconnectPacket(..) => {
                        // Nothing to do here, the other end closes the connection.
                        break;
                    }
                    _ => {}
                }
            }
        });

        Ok(MQTT { stream })
    }

    pub fn publish(&mut self, topic: &str, value: &str) -> Result<(), Error> {
        // Publish the value.
        let topic = TopicName::new(topic).unwrap();
        let publish_packet = PublishPacket::new(topic, QoSWithPacketIdentifier::Level0, value);
        let mut buf = Vec::new();
        publish_packet.encode(&mut buf).unwrap();
        self.stream.write_all(&buf[..])?;

        // Check for any packets from the server to see whether we were disconnected.
        // TODO
        Ok(())
    }
}

impl Drop for MQTT {
    fn drop(&mut self) {
        // TODO: Stop the receive thread.
    }
}