gmqtt-client 0.3.1

use bytes::BytesMut;
use mqttbytes::v5::*;
use tokio_util::codec::{Decoder, Encoder};
use tracing::error;

use super::mqttbytes_error_to_anyhow;

pub(super) struct MqttV5Codec {
    max_incoming_size: usize,
}

impl MqttV5Codec {
    pub(super) fn new(max_incoming_size: usize) -> Self {
        Self { max_incoming_size }
    }
}

impl Decoder for MqttV5Codec {
    type Item = Packet;

    type Error = anyhow::Error;

    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
        // TODO: w razie problemów z odczytami pakietów dodać buforowanie

        if src.is_empty() {
            return Ok(None);
        }

        let packet = match mqttbytes::v5::read(src, self.max_incoming_size) {
            Ok(p) => Some(p),
            Err(mqttbytes::Error::InsufficientBytes(_)) => {
                // Nie wypisujemy, czasem przychodzi, ale nie zauważyłem aby powodował błędy w transmisji [ru]
                None
            }
            Err(err) => {
                error!("Error while decoding MQTT packet {:?}: {}", &src[..], err);
                None
            }
        };

        Ok(packet)
    }

    fn decode_eof(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
        match self.decode(buf)? {
            Some(frame) => Ok(Some(frame)),
            None => {
                if buf.is_empty() {
                    Ok(None)
                } else {
                    Err(
                        std::io::Error::new(std::io::ErrorKind::Other, "bytes remaining on stream")
                            .into(),
                    )
                }
            }
        }
    }
}

macro_rules! impl_encoder_for_packet {
    ($p:ty) => {
        impl Encoder<&$p> for MqttV5Codec {
            type Error = anyhow::Error;
            fn encode(&mut self, packet: &$p, dst: &mut BytesMut) -> Result<(), Self::Error> {
                let _encoded_bytes = packet.write(dst).map_err(mqttbytes_error_to_anyhow)?;

                Ok(())
            }
        }
    };
}

impl_encoder_for_packet!(Connect);
impl_encoder_for_packet!(ConnAck);
impl_encoder_for_packet!(Publish);
impl_encoder_for_packet!(PubAck);
impl_encoder_for_packet!(PubRec);
impl_encoder_for_packet!(PubRel);
impl_encoder_for_packet!(PubComp);
impl_encoder_for_packet!(Subscribe);
impl_encoder_for_packet!(SubAck);
impl_encoder_for_packet!(Unsubscribe);
impl_encoder_for_packet!(UnsubAck);
impl_encoder_for_packet!(PingReq);
impl_encoder_for_packet!(PingResp);
impl_encoder_for_packet!(Disconnect);