coap-zero 0.3.0

// Copyright Open Logistics Foundation
//
// Licensed under the Open Logistics Foundation License 1.3.
// For details on the licensing terms, see the LICENSE file.
// SPDX-License-Identifier: OLFL-1.3

//! A CoAP endpoint that can be used for 1-to-1 communication.
//!
//! This can act both as a "server" and "client", which means that requests can be sent in both
//! directions via a single connection.
//! As stated in [RFC 7252, Section 4.8](https://www.rfc-editor.org/rfc/rfc7252#section-4.8),
//! the Endpoint assumes `NSTART = 1`.
//! This means that at any given time, only one single outgoing request and
//! one single incoming request (including sending the corresponding response) can be handled.
//!
//! This Endpoint does not implement a listening server.
//! So it is not possible to listen to a port and handle any incoming connection.
//! Instead, only 1-to-1 connections can be established via ip, uri or socket.
//!
//! Currently, the [`Token`] size is fixed to 8 bytes (or 0 for Empty messages).

pub mod connect;
pub mod error;
pub mod incoming;
pub mod outgoing;

use core::time::Duration;

use byteorder::{ByteOrder, LittleEndian};
use embedded_hal::blocking::rng;
use embedded_nal::{SocketAddr, UdpClientStack};
use embedded_timers::clock::Clock;

use crate::message::{self, encoded_message::EncodedMessage, token::Token, Message};

use self::{error::Error, incoming::IncomingCommunication, outgoing::OutgoingCommunication};

/// Type alias for an [iri_string] URI
pub type Uri = iri_string::types::RiReferenceStr<iri_string::spec::UriSpec>;

/// The default port the CoAP via UDP Protocol communicates on
pub const DEFAULT_COAP_PORT: u16 = 5683;

/// Events that are generated on the endpoint-level
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum EndpointEvent<'a> {
    /// No _real_ event happened
    Nothing,
    /// A message was received which did not pass the [`EncodedMessage::check_msg_format`] test
    MsgFormatErr(message::Error),
    /// A "ping" message (empty CON) was received. It has been automatically answered with a RST
    /// message.
    Ping,
    /// We have received a message that could not be handled in any meaningful way. It has been
    /// automatically answered with a RST message.
    Unhandled(EncodedMessage<'a>),
}

/// Parameters of a CoAP Endpoint
#[derive(Debug, Clone, Copy)]
pub struct TransmissionParameters {
    /// Minimum spacing before retransmission
    pub ack_timeout: Duration,
    /// Initial Timeout must be between `ack_timeout` and `ack_timeout * ack_random_factor`
    pub ack_random_factor: f32,
    /// Maximum number of retransmissions
    pub max_retransmit: u32,
}

/// Default transmission parameters as defined in <https://datatracker.ietf.org/doc/html/rfc7252#section-4.8>
impl Default for TransmissionParameters {
    fn default() -> Self {
        Self {
            ack_timeout: Duration::from_secs(2),
            ack_random_factor: 1.5,
            max_retransmit: 4,
        }
    }
}

/// Stores the current (re)transmission state for the current message and implements the
/// exponential backoff strategy for retransmissions
#[derive(Debug, Clone, Copy)]
pub struct RetransmissionState {
    /// How often the message has already been retransmitted. If the message has not been sent at
    /// all, this will be None.
    retransmission_counter: Option<u32>,
    /// When the last (re)transmission was sent. If the message has never been sent at all, this
    /// will be None.
    last_transmission_instant: Option<embedded_timers::clock::Instant>,
    /// Initial timeout for the first transmission between
    /// ACK_TIMEOUT and ACK_RANDOM_FACTOR * ACK_TIMEOUT.
    initial_timeout: Duration,
    /// Maximum number of retransmissions, copied from the `TransmissionParameters`
    max_retransmit: u32,
}

/// Error type for [`RetransmissionState::retransmit_required`]
pub struct RetransmissionTimeout;

impl RetransmissionState {
    /// Initializes this `RetransmissionState` with `retransmission_counter` set to None. This
    /// assumes that the state is initialized from one of the `schedule_` methods which can not
    /// actually transmit anything but only trigger an internal state change which requires that
    /// the first message is sent in the next `process_` invocation.
    ///
    /// `random` must be a random value between 0 and 1.
    pub fn new(transmission_parameters: TransmissionParameters, random: f32) -> Self {
        let min_timeout = transmission_parameters.ack_timeout.as_millis() as f32;
        let max_timeout = min_timeout * transmission_parameters.ack_random_factor;
        let initial_timeout_millis = max_timeout * random + min_timeout * (1.0 - random);
        let initial_timeout = Duration::from_millis(initial_timeout_millis as u64);
        Self {
            retransmission_counter: None,
            last_transmission_instant: None,
            initial_timeout,
            max_retransmit: transmission_parameters.max_retransmit,
        }
    }

    /// Returns if a retransmission is required, we should keep waiting or if the whole
    /// transmission has timed out. Returns:
    /// - `Some(true)` if a retransmission should happen
    /// - `Some(false)` if we should keep waiting
    /// - `Err(Error::Timeout)` if the transmission (with retransmissions) has timed out
    ///
    /// If a retransmission is required, this will increase the internal retransmission timer so
    /// this assumes that the retransmission will be triggered.
    pub fn retransmit_required(
        &mut self,
        now: embedded_timers::clock::Instant,
    ) -> Result<bool, RetransmissionTimeout> {
        match (self.retransmission_counter, self.last_transmission_instant) {
            (Some(retransmission_counter), Some(last_instant)) => {
                // From the CoAP RFC 4.2. Messages Transmitted Reliably:
                // "When the timeout is triggered and the retransmission counter is less than
                // MAX_RETRANSMIT, the message is retransmitted, the retransmission counter is
                // incremented, and the timeout is doubled."
                let waited = now - last_instant;
                let timeout = self.initial_timeout * 2u32.pow(retransmission_counter);
                if waited < timeout {
                    Ok(false)
                } else if retransmission_counter < self.max_retransmit {
                    self.retransmission_counter = Some(retransmission_counter + 1);
                    self.last_transmission_instant = Some(now);
                    Ok(true)
                } else {
                    Err(RetransmissionTimeout)
                }
            }
            _ => {
                self.retransmission_counter = Some(0);
                self.last_transmission_instant = Some(now);
                Ok(true)
            }
        }
    }
}

/// Gets a random number between 0 and 1
fn get_random<RNG: rng::Read>(rng: &mut RNG) -> Result<f32, RNG::Error> {
    let mut random_buf = [0_u8; 4];
    rng.read(&mut random_buf)?;
    // random number between 0 and 1
    Ok(u32::from_be_bytes(random_buf) as f32 / (u32::MAX - 1) as f32)
}

/// Message ID that cannot be 0 or u16::MAX
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
struct MessageId(u16);

impl MessageId {
    /// Tries to construct a new random Message ID
    ///
    /// ID = 0x0000 and ID = 0xFFFF won't get generated
    // TODO I read somewhere that all 0s and all 1s have special meaning, but I couldn't find it again.
    fn try_new<RNG: rng::Read>(rng: &mut RNG) -> Result<Self, RNG::Error> {
        let mut counter_bytes = [0_u8; 2];

        while counter_bytes == [0, 0] || counter_bytes == [0xFF, 0xFF] {
            rng.read(&mut counter_bytes)?;
        }

        Ok(Self(LittleEndian::read_u16(&counter_bytes)))
    }

    /// Calculates and returns the next Message ID
    fn next(&mut self) -> u16 {
        self.0 += 1;

        if self.0 == u16::MAX || self.0 == 0 {
            self.0 = 1;
        }

        self.0
    }
}

#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub(crate) struct MessageIdentification {
    pub(crate) id: u16,
    pub(crate) token: Token,
}

#[derive(Debug)]
pub(crate) struct MessageBuffer<const BUFFER_SIZE: usize = { crate::DEFAULT_COAP_MESSAGE_SIZE }> {
    buffer: [u8; BUFFER_SIZE],
    length: usize,
}

impl<const BUFFER_SIZE: usize> Default for MessageBuffer<BUFFER_SIZE> {
    fn default() -> Self {
        Self {
            buffer: [0_u8; BUFFER_SIZE],
            length: 0,
        }
    }
}

impl<const BUFFER_SIZE: usize> MessageBuffer<BUFFER_SIZE> {
    /// Copies the given Encoded Message to the internal buffer
    fn replace_with<E>(&mut self, message: EncodedMessage<'_>) -> Result<(), Error<E>> {
        if message.data.len() > self.buffer.len() {
            return Err(Error::OutOfMemory);
        }

        self.length = message.data.len();

        self.buffer[..self.length].copy_from_slice(message.data);

        Ok(())
    }

    fn encode<const OPTION_COUNT: usize, E>(
        &mut self,
        message: Message<OPTION_COUNT>,
    ) -> Result<(), Error<E>> {
        match message.encode(&mut self.buffer) {
            Ok(encoded) => {
                self.length = encoded.message_length();
                Ok(())
            }
            Err(e) => {
                // We do not know when and how the encoding failed, so we need to invalidate the
                // internal buffer
                self.length = 0;
                Err(Error::MessageError(e))
            }
        }
    }

    /// Returns the stored bytes as [`EncodedMessage`]
    pub fn message(&self) -> Option<EncodedMessage<'_>> {
        if self.length > 0 {
            Some(EncodedMessage::try_new(&self.buffer[..self.length]).unwrap())
        } else {
            None
        }
    }
}

/// Helper type to conveniently handle sending and receiving on the `UdpClientStack` + `UdpSocket`
pub(crate) struct Connection<'a, UDP: UdpClientStack> {
    client_stack: &'a mut UDP,
    link: &'a mut ConnectionLink<UDP::UdpSocket>,
}

impl<'a, UDP: UdpClientStack> Connection<'a, UDP> {
    /// Helper method to block on a `UdpClientStack::send` and map the error appropriately
    fn send(&mut self, packet: &[u8]) -> Result<(), Error<<UDP as UdpClientStack>::Error>> {
        nb::block!(self.client_stack.send(&mut self.link.socket, packet)).map_err(Error::Network)
    }
    /// Helper method to map `WouldBlock` into `None` and `nb::Error::Other` to `endpoint::Error`
    fn receive(
        &mut self,
        buffer: &mut [u8],
    ) -> Result<Option<usize>, Error<<UDP as UdpClientStack>::Error>> {
        match self.client_stack.receive(&mut self.link.socket, buffer) {
            Ok((message_size, sender)) => {
                if sender == self.link.addr {
                    Ok(Some(message_size))
                } else {
                    // If the packet is not from our connection,
                    // we just ignore it
                    Ok(None)
                }
            }
            Err(nb::Error::WouldBlock) => Ok(None),
            Err(nb::Error::Other(e)) => Err(Error::Network(e)),
        }
    }
}

#[derive(Debug)]
pub(crate) struct ConnectionLink<Socket> {
    socket: Socket,
    addr: SocketAddr,
}

/// Endpoint for 1-to-1 connection to another CoAP Endpoint
#[derive(Debug)]
pub struct CoapEndpoint<
    'a,
    UDP,
    RNG,
    CLOCK,
    const MAX_OPTION_COUNT: usize = { crate::DEFAULT_MAX_OPTION_COUNT },
    const MAX_OPTION_SIZE: usize = { crate::DEFAULT_MAX_OPTION_SIZE },
    const INCOMING_BUFFER_SIZE: usize = { crate::DEFAULT_COAP_MESSAGE_SIZE },
    const OUTGOING_BUFFER_SIZE: usize = { crate::DEFAULT_COAP_MESSAGE_SIZE },
    const RECEIVE_BUFFER_SIZE: usize = { crate::DEFAULT_COAP_MESSAGE_SIZE },
> where
    UDP: UdpClientStack,
    RNG: rng::Read,
    CLOCK: Clock,
{
    connection_link: Option<ConnectionLink<UDP::UdpSocket>>,
    message_id_counter: MessageId,
    rng: RNG,
    incoming_communication: IncomingCommunication<
        'a,
        UDP,
        CLOCK,
        INCOMING_BUFFER_SIZE,
        MAX_OPTION_COUNT,
        MAX_OPTION_SIZE,
    >,
    outgoing_communication: OutgoingCommunication<
        'a,
        CLOCK,
        UDP,
        OUTGOING_BUFFER_SIZE,
        MAX_OPTION_COUNT,
        MAX_OPTION_SIZE,
    >,
    // TODO The receive buffer could be on the stack instead which is the same in worst-case
    // scenarios but better otherwise
    receive_buffer: &'a mut [u8],
}

impl<
        'a,
        UDP,
        RNG,
        CLOCK,
        const MAX_OPTION_COUNT: usize,
        const MAX_OPTION_SIZE: usize,
        const INCOMING_BUFFER_SIZE: usize,
        const OUTGOING_BUFFER_SIZE: usize,
        const RECEIVE_BUFFER_SIZE: usize,
    >
    CoapEndpoint<
        'a,
        UDP,
        RNG,
        CLOCK,
        MAX_OPTION_COUNT,
        MAX_OPTION_SIZE,
        INCOMING_BUFFER_SIZE,
        OUTGOING_BUFFER_SIZE,
        RECEIVE_BUFFER_SIZE,
    >
where
    UDP: UdpClientStack,
    RNG: rng::Read,
    CLOCK: Clock,
{
    /// Constructs a new CoAP Endpoint.
    ///
    /// Initializes the message_id_counter randomly.
    pub fn try_new(
        transmission_parameters: TransmissionParameters,
        mut rng: RNG,
        clock: &'a CLOCK,
        receive_buffer: &'a mut [u8],
    ) -> Result<Self, Error<<UDP as UdpClientStack>::Error>> {
        let message_id_counter = MessageId::try_new(&mut rng).map_err(|_| Error::Rng)?;

        let mut slf = Self {
            connection_link: None,
            message_id_counter,
            rng,
            incoming_communication: IncomingCommunication::new(clock, transmission_parameters),
            outgoing_communication: OutgoingCommunication::new(clock, transmission_parameters),
            receive_buffer,
        };

        if slf.incoming_communication.next_message_id.is_none() {
            slf.incoming_communication.next_message_id = Some(slf.message_id_counter.next());
        }
        let random = get_random(&mut slf.rng).map_err(|_| Error::Rng)?;
        slf.incoming_communication.next_random = Some(random);

        if slf.outgoing_communication.next_message_id.is_none() {
            slf.outgoing_communication.next_message_id = Some(slf.message_id_counter.next());
        }
        if slf.outgoing_communication.next_token.is_none() {
            slf.outgoing_communication.next_token = Some(
                Token::try_new(crate::message::token::TokenLength::Eight, &mut slf.rng)
                    .map_err(|_| Error::Rng)?,
            );
        }
        let random = get_random(&mut slf.rng).map_err(|_| Error::Rng)?;
        slf.outgoing_communication.next_random = Some(random);

        Ok(slf)
    }

    /// Access the [`IncomingCommunication`]
    pub fn incoming(
        &mut self,
    ) -> &mut IncomingCommunication<
        'a,
        UDP,
        CLOCK,
        INCOMING_BUFFER_SIZE,
        MAX_OPTION_COUNT,
        MAX_OPTION_SIZE,
    > {
        &mut self.incoming_communication
    }

    /// Access the [`OutgoingCommunication`]
    pub fn outgoing(
        &mut self,
    ) -> &mut OutgoingCommunication<
        'a,
        CLOCK,
        UDP,
        OUTGOING_BUFFER_SIZE,
        MAX_OPTION_COUNT,
        MAX_OPTION_SIZE,
    > {
        &mut self.outgoing_communication
    }

    /// Handles internal state machine.
    ///
    /// Has to be called periodically.
    /// If any ongoing process is present, this function has to be called at an interval
    /// less than CoapEndpoint::transmission_parameter.ack_timeout.
    ///
    /// This method receives a possibly pending RX message and offers this received message to
    /// [`IncomingCommunication`] and [`OutgoingCommunication`]. Incoming pings are automatically
    /// answered with a RST message and a corresponding [`EndpointEvent::Ping`] is generated. If
    /// neither the incoming nor the outgoing communication path have consumed the RX message, an
    /// [`EndpointEvent::Unhandled`] is generated.
    ///
    /// Unhandled requests are silently ignored because we may be able to handle a possible future
    /// retransmission. Unhandled NON and CON responses are automatically rejected with a
    /// RST message because there is no reason to believe that a response which was not consumed
    /// now will be handled in the future. Unhandled ACKs and RSTs are silently ignored because
    /// there must never be a response to ACKs and RSTs, see section 4.2. (Messages Transmitted
    /// Reliably) from RFC 7252: "More generally, recipients of Acknowledgement and Reset messages
    /// MUST NOT respond with either Acknowledgement or Reset messages."
    pub fn process(
        &mut self,
        client_stack: &mut UDP,
    ) -> Result<
        (
            Result<incoming::IncomingEvent, Error<<UDP as UdpClientStack>::Error>>,
            Result<outgoing::OutgoingEvent, Error<<UDP as UdpClientStack>::Error>>,
            EndpointEvent,
        ),
        Error<<UDP as UdpClientStack>::Error>,
    > {
        if self.incoming_communication.next_message_id.is_none() {
            self.incoming_communication.next_message_id = Some(self.message_id_counter.next());
        }
        if self.incoming_communication.next_random.is_none() {
            let random = get_random(&mut self.rng).map_err(|_| Error::Rng)?;
            self.incoming_communication.next_random = Some(random);
        }

        if self.outgoing_communication.next_message_id.is_none() {
            self.outgoing_communication.next_message_id = Some(self.message_id_counter.next());
        }
        if self.outgoing_communication.next_token.is_none() {
            self.outgoing_communication.next_token = Some(
                Token::try_new(crate::message::token::TokenLength::Eight, &mut self.rng)
                    .map_err(|_| Error::Rng)?,
            );
        }
        if self.outgoing_communication.next_random.is_none() {
            let random = get_random(&mut self.rng).map_err(|_| Error::Rng)?;
            self.outgoing_communication.next_random = Some(random);
        }

        let Some(ref mut link) = self.connection_link else {
            return Err(Error::NotConnected);
        };
        let mut connection = Connection { client_stack, link };

        // It may feel strange to have a mutable event variable here because a later event could
        // theoretically overwrite an earlier event. But in the endpoint, events are only generated
        // when the received message is handled and consumed (or not generated at all). Therefore,
        // we will never overwrite anything else than the `Nothing` event.
        let mut endpoint_event = EndpointEvent::Nothing;

        let mut received_message = None;
        if let Some(message_len) = connection.receive(self.receive_buffer)? {
            match EncodedMessage::try_new(&self.receive_buffer[..message_len]) {
                Ok(message) => {
                    match message.check_msg_format::<MAX_OPTION_SIZE>() {
                        Ok(()) => received_message = Some(message),
                        Err(e) => {
                            // Malformed message received, send a RST if it was a NON or CON (never
                            // respond to ACKs or RSTs)
                            use message::Type::{Confirmable, NonConfirmable};
                            if matches!(message.message_type(), Confirmable | NonConfirmable) {
                                connection.send(&EncodedMessage::rst(message.message_id()))?;
                            }
                            endpoint_event = EndpointEvent::MsgFormatErr(e);
                        }
                    }
                }
                Err(e) => {
                    // We could not event construct the encoded message, i.e. message header
                    // incomplete -> no message ID to send a RST
                    endpoint_event = EndpointEvent::MsgFormatErr(e);
                }
            }
        }

        if let Some(message) = received_message.as_mut() {
            if message.is_ping().unwrap() {
                connection.send(&EncodedMessage::rst(message.message_id()))?;
                received_message = None;
                endpoint_event = EndpointEvent::Ping;
            }
        }

        let outgoing_result = self
            .outgoing_communication
            .process_outgoing(&mut connection, &mut received_message);

        let incoming_result = self
            .incoming_communication
            .process_incoming(&mut connection, &mut received_message);

        if let Some(message) = received_message {
            use message::Type::{Confirmable, NonConfirmable};
            if matches!(message.message_type(), Confirmable | NonConfirmable)
                && message.is_response().unwrap()
            {
                connection.send(&EncodedMessage::rst(message.message_id()))?;
            }
            endpoint_event = EndpointEvent::Unhandled(message);
        }

        Ok((incoming_result, outgoing_result, endpoint_event))
    }
}

#[cfg(test)]
mod tests {
    use core::{cell::RefCell, time::Duration};

    use embedded_hal::prelude::_embedded_hal_blocking_rng_Read;
    use embedded_nal::{IpAddr, Ipv4Addr, SocketAddr, UdpClientStack};
    use heapless::Vec;
    use mockall::predicate::*;
    use mockall::*;

    use crate::{
        endpoint::outgoing::OutgoingEvent,
        message::{
            codes::RequestCode,
            encoded_message::EncodedMessage,
            token::{Token, TokenLength},
            Message, Type,
        },
    };

    use super::{incoming::IncomingEvent, CoapEndpoint, EndpointEvent, TransmissionParameters};

    #[derive(Debug)]
    struct Random {
        value: u128,
    }

    impl embedded_hal::blocking::rng::Read for Random {
        type Error = std::io::Error;

        fn read(&mut self, buf: &mut [u8]) -> Result<(), Self::Error> {
            self.value += 1;

            let buf_len = buf.len();

            buf[..buf_len].copy_from_slice(&self.value.to_le_bytes()[..buf_len]);

            Ok(())
        }
    }

    #[derive(Debug)]
    struct StackError;

    struct Socket;

    mock! {
        Stack {}

        impl UdpClientStack for Stack {
            type UdpSocket = Socket;
            type Error = StackError;

            fn socket(&mut self) -> Result<Socket, StackError>;
            fn connect(
                &mut self,
                socket: &mut Socket,
                remote: SocketAddr
            ) -> Result<(), StackError>;
            fn send(
                &mut self,
                socket: &mut Socket,
                buffer: &[u8]
            ) -> Result<(), nb::Error<StackError>>;
            fn receive(
                &mut self,
                socket: &mut Socket,
                buffer: &mut [u8]
            ) -> Result<(usize, SocketAddr), nb::Error<StackError>>;
            fn close(&mut self, socket: Socket) -> Result<(), StackError>;
        }
    }

    #[derive(Debug)]
    struct MyClock {
        last_time: RefCell<Duration>,
        now: RefCell<Duration>,
    }

    impl embedded_timers::clock::Clock for MyClock {
        fn try_now(
            &self,
        ) -> Result<embedded_timers::clock::Instant, embedded_timers::clock::ClockError> {
            *self.last_time.borrow_mut() = *self.now.borrow();
            Ok(*self.now.borrow())
        }
    }

    #[test]
    fn nothing() {
        let mut stack = MockStack::default();

        let clock = MyClock {
            last_time: RefCell::new(Duration::from_secs(0)),
            now: RefCell::new(Duration::from_secs(1)),
        };

        let mut receive_buffer = [0_u8; crate::DEFAULT_COAP_MESSAGE_SIZE];

        let mut endpoint: CoapEndpoint<
            '_,
            MockStack,
            Random,
            MyClock,
            8,
            32,
            128,
            //{ coap_zero::DEFAULT_COAP_MESSAGE_SIZE },
        > = CoapEndpoint::try_new(
            TransmissionParameters::default(),
            Random { value: 0 },
            &clock,
            &mut receive_buffer,
        )
        .unwrap();

        stack.expect_socket().once().return_once(|| Ok(Socket));
        stack.expect_connect().once().return_once(|_, _| Ok(()));

        endpoint
            .connect_to_addr(&mut stack, "127.0.0.1:5683".parse().unwrap())
            .unwrap();

        stack
            .expect_receive()
            .once()
            .return_once(|_, _| Err(nb::Error::WouldBlock));

        let (_, _, endpoint_event) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(endpoint_event, EndpointEvent::Nothing));
    }

    #[test]
    fn message_format_error() {
        let mut stack = MockStack::default();

        let clock = MyClock {
            last_time: RefCell::new(Duration::from_secs(0)),
            now: RefCell::new(Duration::from_secs(1)),
        };

        let mut receive_buffer = [0_u8; crate::DEFAULT_COAP_MESSAGE_SIZE];

        let mut endpoint: CoapEndpoint<
            '_,
            MockStack,
            Random,
            MyClock,
            8,
            32,
            128,
            //{ coap_zero::DEFAULT_COAP_MESSAGE_SIZE },
        > = CoapEndpoint::try_new(
            TransmissionParameters::default(),
            Random { value: 0 },
            &clock,
            &mut receive_buffer,
        )
        .unwrap();

        stack.expect_socket().once().return_once(|| Ok(Socket));
        stack.expect_connect().once().return_once(|_, _| Ok(()));

        endpoint
            .connect_to_addr(&mut stack, "127.0.0.1:5683".parse().unwrap())
            .unwrap();

        stack.expect_receive().once().return_once(|_, buffer| {
            let malformed_message = [
                // This is the CON GET /hello message from the other tests
                // only changed the payload marker.   v Here 255 would be correct.
                72, 1, 0, 4, 5, 0, 0, 0, 0, 0, 0, 0, 254, 47, 104, 101, 108, 108, 111,
            ];

            buffer[..malformed_message.len()].copy_from_slice(&malformed_message);

            Ok((
                malformed_message.len(),
                SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 5683),
            ))
        });

        let (incoming_event, _, _) = endpoint.process(&mut stack).unwrap();
        match incoming_event.unwrap() {
            IncomingEvent::Request(_con, req) => {
                let msg: Result<Message<'_, 8>, crate::message::Error> = req.try_into();
                if !matches!(msg, Err(crate::message::Error::InvalidOption(_))) {
                    panic!("Expected Err(InvalidOption), got: {msg:?}");
                }
            }
            event => {
                panic!("Expected IncomingEvent::Request, got: {event:?}");
            }
        }

        stack.expect_receive().once().return_once(|_, buffer| {
            let malformed_message = [
                // This message is invalid because the Code is 0.05 which does not exist
                72, 5, 0, 4, 5, 0, 0, 0, 0, 0, 0, 0, 255, 47, 104, 101, 108, 108, 111,
            ];

            buffer[..malformed_message.len()].copy_from_slice(&malformed_message);

            Ok((
                malformed_message.len(),
                SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 5683),
            ))
        });
        stack.expect_send().once().return_once(move |_, buf| {
            // The MsgFormatError will be responded automatically with a RST message
            assert_eq!(buf, crate::message::encoded_message::EncodedMessage::rst(4));
            Ok(())
        });

        let (_, _, endpoint_event) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(endpoint_event, EndpointEvent::MsgFormatErr(_)));
    }

    #[test]
    fn ping_pong() {
        let mut stack = MockStack::default();

        let clock = MyClock {
            last_time: RefCell::new(Duration::from_secs(0)),
            now: RefCell::new(Duration::from_secs(1)),
        };

        let mut receive_buffer = [0_u8; crate::DEFAULT_COAP_MESSAGE_SIZE];

        let mut endpoint: CoapEndpoint<
            '_,
            MockStack,
            Random,
            MyClock,
            8,
            32,
            128,
            //{ coap_zero::DEFAULT_COAP_MESSAGE_SIZE },
        > = CoapEndpoint::try_new(
            TransmissionParameters::default(),
            Random { value: 0 },
            &clock,
            &mut receive_buffer,
        )
        .unwrap();

        stack.expect_socket().once().return_once(|| Ok(Socket));
        stack.expect_connect().once().return_once(|_, _| Ok(()));

        endpoint
            .connect_to_addr(&mut stack, "127.0.0.1:5683".parse().unwrap())
            .unwrap();

        let message_id = 1;
        let ping: Message<0> = Message::new_ping(message_id);
        let mut pong_buf = [0_u8; 4];
        let _encoded_pong = EncodedMessage::new_rst(message_id, &mut pong_buf);

        stack.expect_receive().once().return_once(move |_, buffer| {
            let encoded_message = ping.encode(buffer).unwrap();

            Ok((
                encoded_message.message_length(),
                SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 5683),
            ))
        });
        stack.expect_send().once().return_once(move |_, buf| {
            assert_eq!(buf, pong_buf);
            Ok(())
        });

        let (_, _, endpoint_event) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(endpoint_event, EndpointEvent::Ping));

        stack
            .expect_receive()
            .once()
            .return_once(|_, _| Err(nb::Error::WouldBlock));

        let (_, _, endpoint_event) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(endpoint_event, EndpointEvent::Nothing));
    }

    #[test]
    fn unhandled() {
        let mut stack = MockStack::default();

        let clock = MyClock {
            last_time: RefCell::new(Duration::from_secs(0)),
            now: RefCell::new(Duration::from_secs(1)),
        };

        let mut receive_buffer = [0_u8; crate::DEFAULT_COAP_MESSAGE_SIZE];

        let mut endpoint: CoapEndpoint<
            '_,
            MockStack,
            Random,
            MyClock,
            8,
            32,
            128,
            //{ coap_zero::DEFAULT_COAP_MESSAGE_SIZE },
        > = CoapEndpoint::try_new(
            TransmissionParameters::default(),
            Random { value: 0 },
            &clock,
            &mut receive_buffer,
        )
        .unwrap();

        stack.expect_socket().once().return_once(|| Ok(Socket));
        stack.expect_connect().once().return_once(|_, _| Ok(()));

        endpoint
            .connect_to_addr(&mut stack, "127.0.0.1:5683".parse().unwrap())
            .unwrap();

        let next_message_id_1 = endpoint.message_id_counter.next();
        let mut token_1 = Token::default();
        token_1.length = TokenLength::Eight;
        endpoint.rng.read(&mut token_1.bytes).unwrap();
        let request_1: Message<'_> = Message::new(
            Type::Confirmable,
            RequestCode::Get.into(),
            next_message_id_1,
            token_1,
            Vec::new(),
            Some(b"/hello"),
        );

        let next_message_id_2 = endpoint.message_id_counter.next();
        let mut token_2 = Token::default();
        token_2.length = TokenLength::Eight;
        endpoint.rng.read(&mut token_2.bytes).unwrap();
        let request_2: Message<'_> = Message::new(
            Type::Confirmable,
            RequestCode::Get.into(),
            next_message_id_2,
            token_2,
            Vec::new(),
            Some(b"/hello"),
        );
        let request_test = request_2.clone();

        stack.expect_receive().once().return_once(move |_, buffer| {
            let encoded_message = request_1.encode(buffer).unwrap();

            Ok((
                encoded_message.message_length(),
                SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 5683),
            ))
        });

        let (incoming, _, endpoint_event) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::Request(true, _)));
        assert!(matches!(endpoint_event, EndpointEvent::Nothing));

        stack.expect_receive().once().return_once(move |_, buffer| {
            let encoded_message = request_2.encode(buffer).unwrap();

            Ok((
                encoded_message.message_length(),
                SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 5683),
            ))
        });

        let (incoming, _, endpoint_event) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::Nothing));
        assert!(matches!(endpoint_event, EndpointEvent::Unhandled(_)));

        match endpoint_event {
            EndpointEvent::Unhandled(message) => {
                assert_eq!(Message::try_from(message).unwrap(), request_test)
            }
            _ => (),
        }
    }

    #[test]
    fn ignore_message() {
        let mut stack = MockStack::default();

        let clock = MyClock {
            last_time: RefCell::new(Duration::from_secs(0)),
            now: RefCell::new(Duration::from_secs(1)),
        };

        let mut receive_buffer = [0_u8; crate::DEFAULT_COAP_MESSAGE_SIZE];

        let mut endpoint: CoapEndpoint<
            '_,
            MockStack,
            Random,
            MyClock,
            8,
            32,
            128,
            //{ coap_zero::DEFAULT_COAP_MESSAGE_SIZE },
        > = CoapEndpoint::try_new(
            TransmissionParameters::default(),
            Random { value: 0 },
            &clock,
            &mut receive_buffer,
        )
        .unwrap();

        stack.expect_socket().once().return_once(|| Ok(Socket));
        stack.expect_connect().once().return_once(|_, _| Ok(()));

        endpoint
            .connect_to_addr(&mut stack, "127.0.0.1:5683".parse().unwrap())
            .unwrap();

        let next_message_id = endpoint.message_id_counter.next();
        let mut token = Token::default();
        token.length = TokenLength::Eight;
        endpoint.rng.read(&mut token.bytes).unwrap();
        let request: Message<'_> = Message::new(
            Type::Confirmable,
            RequestCode::Get.into(),
            next_message_id,
            token,
            Vec::new(),
            Some(b"/hello"),
        );

        stack.expect_receive().once().return_once(move |_, buffer| {
            let encoded_message = request.encode(buffer).unwrap();

            Ok((
                encoded_message.message_length(),
                // We expect only messages from 127.0.0.1                        v
                SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 2)), 5683),
            ))
        });

        let (incoming, outgoing, endpoint_event) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::Nothing));
        assert!(matches!(outgoing.unwrap(), OutgoingEvent::Nothing));
        assert!(matches!(endpoint_event, EndpointEvent::Nothing));
    }
}