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

//! The incoming communication path

use core::marker::PhantomData;

use embedded_nal::UdpClientStack;
use embedded_timers::clock::Clock;
use heapless::Vec;

use crate::message::{
    codes::ResponseCode, encoded_message::EncodedMessage, options::CoapOption, Message, Type,
};

use super::{
    error::Error, Connection, MessageBuffer, MessageIdentification, RetransmissionState,
    RetransmissionTimeout, TransmissionParameters,
};

/// The state in which the incoming communication path is currently
#[derive(Debug, Clone, Copy)]
pub enum IncomingState {
    /// New requests can be handled
    //
    // In this state, it depends on the contained [`LastResponse`] what the internal message
    // buffer contains. Since it should not be required, access to it is forbidden.
    Idle(LastResponse),
    /// A request was received and needs to be handled by the user. The contained `bool` determines
    /// if the request was confirmable (`true`) or non-confirmable (`false`).
    ///
    /// In any case, the response may be sent with [`IncomingCommunication::schedule_response`]
    /// immediately. For CON requests, this will result in a piggybacked response. For NON
    /// requests, a NON response will be sent which is considered the default case for NON requests
    /// (see RFC 7252 5.2.3 Non-confirmable: "If the request message is Non-confirmable, then the
    /// response SHOULD be returned in a Non-confirmable message as well.")
    ///
    /// Furthermore, [`IncomingCommunication::schedule_rst`] may be called in both cases which will
    /// transmit a RST message.
    ///
    /// If a different behavior is desired, the following different options exist:
    /// - In `Received(true)`, [`IncomingCommunication::schedule_empty_ack`] allows to send an
    /// empty ACK which will trigger a state transition via [`IncomingState::SendingAck`]
    /// into [`IncomingState::AwaitingResponse`] where separate responses may be sent.
    /// Additionally, [`IncomingCommunication::schedule_piggybacked_response`] may be called which
    /// is identical to [`IncomingCommunication::schedule_response`] in this case.
    /// - In `Received(false)`, [`IncomingCommunication::schedule_con_response`] and
    /// [`IncomingCommunication::schedule_non_response`] are available, the second being
    /// identical to [`IncomingCommunication::schedule_response`] in this case.
    ///
    /// In this state, the message buffer contains the request message. Thus, the request
    /// message may be obtained by calling [`IncomingCommunication::request`].
    Received(bool),
    /// An ACK is being sent. Afterwards, we will go into the [`IncomingState::AwaitingResponse`]
    /// state.
    //
    // In this state, the message buffer still contains the request message and may be obtained
    // with [`IncomingCommunication::request`].
    SendingAck,
    /// A RST message is being sent. Afterwards, we will go back to the `Idle` state.
    SendingRst,
    /// A piggybacked response is being sent. Afterwards, we will go into the `Idle` state.
    //
    // In this state, the message buffer contains the piggybacked response message.
    SendingPiggybacked,
    /// The request has been acknowledged, i.e. a separate response is expected. In this state,
    /// [`IncomingCommunication::schedule_con_response`] and
    /// [`IncomingCommunication::schedule_non_response`] are available and either of those
    /// _must_ be called by the user.
    ///
    /// In this state, the message buffer still contains the request message and may be obtained
    /// with [`IncomingCommunication::request`].
    AwaitingResponse,
    /// A separate CON response is being sent. We will stay in this state and try retransmissions
    /// until an ACK to this message is received or until the whole communication attempt (with
    /// retries) times out.
    //
    // In this state, the message buffer contains the separate CON response message.
    SendingCon(RetransmissionState),
    /// A separate NON response is being sent.
    //
    // In this state, the message buffer contains the separate NON response message.
    SendingNon,
}

/// Distinguishes different cases on how we can respond to a request
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LastResponse {
    /// We have not yet answered any request
    //
    // The internal message_buffer should be `None` in this case.
    NoResponse,
    /// The last request was answered with a RST message
    //
    // The internal message buffer contains the last request because it has not been overwritten
    // with an actual response. It should not be required anyways.
    RespRst,
    /// The last request was answered with a CON message
    ///
    /// The contained `bool` stores if the CON message was successfully acked. If `false`,
    /// transmitting the CON timed out.
    //
    // The internal message buffer contains the CON response that was sent out.
    RespCon(bool),
    /// The last request was answered with a separate NON response or a piggybacked response (in
    /// both cases, we do not know if the requester has received our response)
    //
    // The internal message buffer contains the response we have sent out before.
    RespNon,
}

/// What has happened in the [`IncomingCommunication`] path when calling
/// [`CoapEndpoint::process`](super::CoapEndpoint::process)
///
/// This type is annotated with `must_use` because some events require user interaction to avoid
/// getting stuck in specific states. See the documentation for the specific events for more
/// information.
#[must_use]
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum IncomingEvent<'a> {
    /// No _real_ event happened. This pseudo-event is included as a variant here to allow concise
    /// event handling on the user side.
    Nothing,
    /// A new request was received. The user _must_ respond to this event. The event contains the
    /// `bool` flag if the request is confirmable (`true`) or non-confirmable (`false`) and the
    /// request message itself.
    ///
    /// In any case, the response may be sent with [`IncomingCommunication::schedule_response`]
    /// immediately. For CON requests, this will result in a piggybacked response. For NON
    /// requests, a NON response will be sent. For other options, see the
    /// [`IncomingState::Received`] state.
    ///
    /// The contained [`EncodedMessage`] keeps the borrow of the [`Endpoint`](super::CoapEndpoint)
    /// alive which might be undesirable in some situations. To resolve this, it may safely be `drop`ped, it can be accessed later via [`IncomingCommunication::request`].
    Request(bool, EncodedMessage<'a>),
    /// We have received the same request again. Depending on how we have already handled the
    /// request, appropriate action is taken automatically. For example, if the request has already
    /// been ACKed but a separate response is still missing, we send the ACK again. If we have sent
    /// a NON response before which may not have reached the other endpoint, we send it again.
    DuplicatedRequest,
    /// We have sent an ACK. With this event, a state transition into
    /// [`IncomingState::AwaitingResponse`] takes place in which more user interaction is
    /// _required_.
    SendAck,
    /// A separate CON response has been (re-)sent. NON and piggybacked responses directly result
    /// in a [`IncomingEvent::Success`] event.
    SendCon,
    /// We have successfully finished the communication (from our perspective). If our response was
    /// piggybacked or a separate NON message, we have no guarantee that the other endpoint has
    /// already received our response. If our response was a separate CON message, the
    /// corresponding ACK has been received and we know that the other endpoint has received our
    /// response.
    Success,
    /// We have sent a RST message. With this event, a state transition into
    /// [`IncomingState::Idle`] takes place.
    SendRst,
    /// Sending the separate CON response has timed out. Piggybacked or separate NON responses will
    /// not be ACKed so they can not time out.
    Timeout,
    /// We have received a RST when sending a separate CON response
    RecvRst,
}

/// Incoming communication path. This handles incoming requests and pings which will be
/// automatically answered with RST messages.
#[derive(Debug)]
pub struct IncomingCommunication<
    'a,
    UDP: UdpClientStack,
    CLOCK: Clock,
    const BUFFER_SIZE: usize = { crate::DEFAULT_COAP_MESSAGE_SIZE },
    const MAX_OPTION_COUNT: usize = { crate::DEFAULT_MAX_OPTION_COUNT },
    const MAX_OPTION_SIZE: usize = { crate::DEFAULT_MAX_OPTION_SIZE },
> {
    message_buffer: MessageBuffer<BUFFER_SIZE>,
    /// Depending on the state, contains the request we are currently handling or the last request
    /// we have handled (for de-duplication)
    last_request: Option<MessageIdentification>,
    state: IncomingState,
    clock: &'a CLOCK,
    transmission_parameters: TransmissionParameters,
    /// Next message ID to create/schedule a separate response. Since the endpoint is in charge of
    /// the message ID generation, the next message ID must be determined in advance.
    pub(super) next_message_id: Option<u16>,
    /// Random value between 0 and 1 to initialize the RetransmissionState required to schedule a
    /// CON message. Since the endpoint owns the RNG, the next random number must be determined in
    /// advance.
    pub(super) next_random: Option<f32>,
    _udp: PhantomData<UDP>,
}

impl<
        'a,
        UDP,
        CLOCK,
        const BUFFER_SIZE: usize,
        const MAX_OPTION_COUNT: usize,
        const MAX_OPTION_SIZE: usize,
    > IncomingCommunication<'a, UDP, CLOCK, BUFFER_SIZE, MAX_OPTION_COUNT, MAX_OPTION_SIZE>
where
    UDP: UdpClientStack,
    CLOCK: Clock,
{
    /// Initializes the incoming communication path in the [`IncomingState::Idle`] state so it is
    /// ready to handle incoming requests.
    pub fn new(clock: &'a CLOCK, transmission_parameters: TransmissionParameters) -> Self {
        Self {
            message_buffer: MessageBuffer::default(),
            last_request: None,
            state: IncomingState::Idle(LastResponse::NoResponse),
            clock,
            transmission_parameters,
            next_message_id: None,
            next_random: None,
            _udp: PhantomData,
        }
    }

    /// Returns the current state of the `IncomingCommunication`. Depending on the state, the user
    /// _must_ take action to drive the state forward, see the documentation to the states in
    /// [`IncomingState`].
    pub fn state(&self) -> IncomingState {
        self.state
    }

    /// Resets the internal state machine
    pub fn reset(&mut self) {
        *self = Self::new(self.clock, self.transmission_parameters);
    }

    /// Returns the current request message which is the same that was returned in the `Request`
    /// event before. Therefore, usage of this method is only required if the decision between
    /// ACK + separate response and a piggybacked response shall be postponed (the message from the
    /// `Request` event can probably not be stored due to lifetime requirements).
    ///
    /// This method is only available in [`IncomingState::Received`], [`IncomingState::SendingAck`]
    /// and [`IncomingState::AwaitingResponse`] and will return [`Error::Forbidden`] otherwise.
    pub fn request(&self) -> Result<EncodedMessage, Error<<UDP as UdpClientStack>::Error>> {
        use IncomingState::*;
        match self.state {
            Received(_) | SendingAck | AwaitingResponse => {
                Ok(self.message_buffer.message().unwrap())
            }
            _ => Err(Error::Forbidden),
        }
    }

    /// Schedules an immediate response to be sent.
    ///
    /// If a confirmable request was received, a piggy-backed response will be scheduled.
    /// In case of a non-confirmable request, the response will be send as non-confirmable as well.
    ///
    /// For more control, the user can check if the request was confirmable or non-confirmable
    /// themselves and call [`IncomingCommunication::schedule_empty_ack`],
    /// [`IncomingCommunication::schedule_piggybacked_response`],
    /// [`IncomingCommunication::schedule_con_response`] or
    /// [`IncomingCommunication::schedule_non_response`] accordingly.
    pub fn schedule_response(
        &mut self,
        code: ResponseCode,
        options: Vec<CoapOption<'_>, MAX_OPTION_COUNT>,
        payload: Option<&[u8]>,
    ) -> Result<(), Error<<UDP as UdpClientStack>::Error>> {
        match self.state {
            IncomingState::Received(true) => {
                self.schedule_piggybacked_response(code, options, payload)
            }
            IncomingState::Received(false) => self.schedule_non_response(code, options, payload),
            _ => Err(Error::Forbidden),
        }
    }

    /// Schedules an Acknowledgement to be sent.
    ///
    /// The ACK will fit the `message_id` for the last received request
    pub fn schedule_empty_ack(&mut self) -> Result<(), Error<<UDP as UdpClientStack>::Error>> {
        if !matches!(self.state, IncomingState::Received(true)) {
            return Err(Error::Forbidden);
        }
        self.state = IncomingState::SendingAck;
        Ok(())
    }

    /// Schedules a RST message to be sent.
    pub fn schedule_rst(&mut self) -> Result<(), Error<<UDP as UdpClientStack>::Error>> {
        if !matches!(self.state, IncomingState::Received(_)) {
            return Err(Error::Forbidden);
        }
        self.state = IncomingState::SendingRst;
        Ok(())
    }

    /// Schedules an Acknowledgement with piggy-backed response
    ///
    /// The ACK will fit the `message_id` for the last received request.
    /// The response is defined by the given details (`code`, `payload` and `additional_options`).
    pub fn schedule_piggybacked_response(
        &mut self,
        code: ResponseCode,
        options: Vec<CoapOption<'_>, MAX_OPTION_COUNT>,
        payload: Option<&[u8]>,
    ) -> Result<(), Error<<UDP as UdpClientStack>::Error>> {
        if !matches!(self.state, IncomingState::Received(true)) {
            return Err(Error::Busy);
        }

        let ident = self.last_request.as_ref().unwrap();

        let message: Message<MAX_OPTION_COUNT> = Message::new(
            Type::Acknowledgement,
            code.into(),
            ident.id,
            ident.token,
            options,
            payload,
        );
        self.message_buffer.encode(message)?;

        self.state = IncomingState::SendingPiggybacked;

        Ok(())
    }

    /// Schedules a message to be sent as a separate CON response
    pub fn schedule_con_response(
        &mut self,
        code: ResponseCode,
        options: Vec<CoapOption<'_>, MAX_OPTION_COUNT>,
        payload: Option<&[u8]>,
    ) -> Result<(), Error<<UDP as UdpClientStack>::Error>> {
        self.schedule_separate_response(Type::Confirmable, code, options, payload)?;
        let retransmission_state = RetransmissionState::new(
            self.transmission_parameters,
            self.next_random.take().unwrap(),
        );
        self.state = IncomingState::SendingCon(retransmission_state);
        Ok(())
    }

    /// Schedules a message to be sent as a separate NON response
    pub fn schedule_non_response(
        &mut self,
        code: ResponseCode,
        options: Vec<CoapOption<'_>, MAX_OPTION_COUNT>,
        payload: Option<&[u8]>,
    ) -> Result<(), Error<<UDP as UdpClientStack>::Error>> {
        self.schedule_separate_response(Type::NonConfirmable, code, options, payload)?;
        self.state = IncomingState::SendingNon;
        Ok(())
    }

    /// Schedules a message to be sent as a response for an incoming request.
    ///
    /// Internal method to handle CON and NON cases
    fn schedule_separate_response(
        &mut self,
        response_type: Type,
        code: ResponseCode,
        options: Vec<CoapOption<'_>, MAX_OPTION_COUNT>,
        payload: Option<&[u8]>,
    ) -> Result<(), Error<<UDP as UdpClientStack>::Error>> {
        if !matches!(self.state, IncomingState::AwaitingResponse)
            && !matches!(self.state, IncomingState::Received(false))
        {
            return Err(Error::Busy);
        }

        let message: Message<MAX_OPTION_COUNT> = Message::new(
            response_type,
            code.into(),
            self.next_message_id.unwrap(),
            self.last_request.unwrap().token,
            options,
            payload,
        );

        self.message_buffer.encode(message)?;

        self.next_message_id = None;

        Ok(())
    }

    /// Drives the internal state machine
    ///
    /// If we receive a new request in a non-Idle state, it is ignored here. This seems to be the
    /// right thing to do if we imagine a setup with NSTART>1 because then, an other
    /// [`IncomingCommunication`] could handle this request. Currently, this will result in
    /// unhandled messages in the endpoint which is responsible to react accordingly. It is advised
    /// to silently ignore the request to trigger a retransmission on the requester's side.
    pub(crate) fn process_incoming(
        &mut self,
        connection: &mut Connection<UDP>,
        received: &mut Option<EncodedMessage<'_>>,
    ) -> Result<IncomingEvent, Error<<UDP as UdpClientStack>::Error>> {
        use IncomingEvent::*;
        use IncomingState::*;
        use LastResponse::*;

        // We can generate events when we handle the received message or when an internal
        // retransmission timeout occurs. So whenever either of those happens, we should skip the
        // other because we can not return two events at the same time. Since we can only handle a
        // received message _now_, we start by handling messages and skip checking timeouts. If the
        // user calls the process method often enough, the timeout will trigger anyways in the next
        // iteration.
        //
        // Message de-duplication happens in nearly all states. To not duplicate the de-duplication
        // code in every state, we handle de-duplication in a first match and the other cases in a
        // second match.

        if let (Some(message), Some(last_message_ident)) = (&received, self.last_request) {
            if message.is_request().unwrap()
                && last_message_ident.id == message.message_id()
                && last_message_ident.token == message.token().unwrap()
            {
                match self.state {
                    Idle(NoResponse) => panic!(
                        "In the Idle(NoResponse) state, the internal message buffer should be None"
                    ),
                    Idle(RespRst) => {
                        // Previously, we have responded with an ACK. We should do so now, too.
                        connection.send(&EncodedMessage::rst(last_message_ident.id))?;
                    }
                    Idle(RespCon(true)) => {
                        // The requester has already received and acked our response. So this
                        // must be a request which has been sent before their ACK. We should
                        // ignore it.
                    }
                    Idle(RespCon(false)) => {
                        // Our CON response (with retries!) timed out. We have already tried a lot
                        // to respond but the connection just seems to be bad. Who knows how this
                        // request reached us again, let's just not handle it again.
                    }
                    Idle(RespNon) => {
                        // We have responded with a NON message which may have not have reached the
                        // other endpoint. Let's just try again.
                        connection.send(self.message_buffer.message().unwrap().data)?;
                    }
                    Received(_) => {
                        // We do not know the response yet so we can only ignore it
                    }
                    SendingAck | SendingRst | SendingPiggybacked | SendingNon => {
                        // We can ignore it because a message will be sent now anyways
                    }
                    AwaitingResponse => {
                        // We have already ACKed the request but we do not know the separate
                        // response yet. It looks like the ACK did not reach the other endpoint, so
                        // let's send it again.
                        connection.send(&EncodedMessage::ack(last_message_ident.id))?;
                    }
                    SendingCon(_) => {
                        // We are in the process of (re-)transmitting a CON response anyways so
                        // we do not need to ACK again
                    }
                }
                received.take();
                return Ok(DuplicatedRequest);
            }
        }

        match self.state {
            Idle(_) => {
                // When we get passed some received data,
                // we need to check if we just received a request.
                if let Some(message) = received {
                    if message.is_request().unwrap() {
                        // New request, we save all metadata and return the message
                        self.last_request = Some(MessageIdentification {
                            id: message.message_id(),
                            token: message.token().unwrap(),
                        });
                        let confirmable = message.message_type() == Type::Confirmable;
                        self.message_buffer.replace_with(received.take().unwrap())?;
                        self.state = Received(confirmable);
                        return Ok(Request(confirmable, self.message_buffer.message().unwrap()));
                    }
                }
            }
            Received(_) | AwaitingResponse => {
                // We actually don't have anything to do
                // since we are just waiting for the user to ACK.
                // In case we are already in state AwaitingResponse,
                // we wait for the user to send the separate response.
            }
            SendingAck => {
                let ident = self.last_request.as_ref().unwrap();
                connection.send(&EncodedMessage::ack(ident.id))?;
                self.state = AwaitingResponse;
                return Ok(SendAck);
            }
            SendingRst => {
                let ident = self.last_request.as_ref().unwrap();
                connection.send(&EncodedMessage::rst(ident.id))?;
                self.state = Idle(RespRst);
                return Ok(SendRst);
            }
            SendingPiggybacked | SendingNon => {
                connection.send(self.message_buffer.message().unwrap().data)?;
                self.state = Idle(RespNon);
                return Ok(Success);
            }
            SendingCon(ref mut retransmission_state) => {
                if let Some(message) = received {
                    if message.message_type() == Type::Acknowledgement
                        && message.message_id()
                            == self.message_buffer.message().unwrap().message_id()
                    {
                        received.take();
                        self.state = Idle(RespCon(true));
                        return Ok(Success);
                    } else if message.message_type() == Type::Reset
                        && message.message_id()
                            == self.message_buffer.message().unwrap().message_id()
                    {
                        received.take();
                        self.state = Idle(RespCon(false));
                        return Ok(RecvRst);
                    }
                }
                match retransmission_state.retransmit_required(self.clock.try_now().unwrap()) {
                    Ok(true) => {
                        connection.send(self.message_buffer.message().unwrap().data)?;
                        return Ok(SendCon);
                    }
                    Ok(false) => {}
                    Err(RetransmissionTimeout) => {
                        self.state = Idle(RespCon(false));
                        // Timeout is an Event not an Error because it is specified in the protocol
                        return Ok(Timeout);
                    }
                }
            }
        }

        Ok(Nothing)
    }
}

#[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::incoming::{IncomingEvent, IncomingState},
        message::{
            codes::{RequestCode, SuccessCode},
            token::{Token, TokenLength},
            Message, Type,
        },
    };

    use super::{super::CoapEndpoint, 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())
        }
    }

    impl MyClock {
        fn advance(&self, step: Duration) {
            *self.now.borrow_mut() = *self.last_time.borrow() + step;
        }
    }

    #[test]
    fn receive_con_get_piggybacked() {
        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 (incoming, _, _) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::Nothing));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Idle(_)
        ));

        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"),
        );
        let response: Message<'_> = Message::new(
            Type::Acknowledgement,
            SuccessCode::Content.into(),
            next_message_id,
            token,
            Vec::new(),
            Some(b"world"),
        );
        let mut response_buf = [0_u8; 32];
        let response_length = response.encode(&mut response_buf).unwrap().message_length();

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

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

        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::Request(true, _)));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Received(true)
        ));
        // TODO Test Message

        endpoint
            .incoming()
            .schedule_piggybacked_response(SuccessCode::Content.into(), Vec::new(), Some(b"world"))
            .unwrap();

        stack
            .expect_receive()
            .once()
            .return_once(|_, _| Err(nb::Error::WouldBlock));
        stack.expect_send().once().return_once(move |_, buffer| {
            assert_eq!(buffer, &response_buf[..response_length]);

            Ok(())
        });

        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::Success));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Idle(_)
        ));
    }

    #[test]
    fn receive_non_get_non() {
        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 (incoming, _, _) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::Nothing));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Idle(_)
        ));

        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::NonConfirmable,
            RequestCode::Get.into(),
            next_message_id,
            token,
            Vec::new(),
            Some(b"/hello"),
        );
        let response: Message<'_> = Message::new(
            Type::NonConfirmable,
            SuccessCode::Content.into(),
            endpoint.incoming_communication.next_message_id.unwrap(),
            token,
            Vec::new(),
            Some(b"world"),
        );
        let mut response_buf = [0_u8; 32];
        let response_length = response.encode(&mut response_buf).unwrap().message_length();

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

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

        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(
            incoming.unwrap(),
            IncomingEvent::Request(false, _)
        ));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Received(false)
        ));
        // TODO Test Message

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

        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::Nothing));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Received(false)
        ));

        endpoint
            .incoming()
            .schedule_response(SuccessCode::Content.into(), Vec::new(), Some(b"world"))
            .unwrap();

        stack
            .expect_receive()
            .once()
            .return_once(|_, _| Err(nb::Error::WouldBlock));
        stack.expect_send().once().return_once(move |_, buffer| {
            assert_eq!(buffer, &response_buf[..response_length]);

            Ok(())
        });

        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::Success));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Idle(_)
        ));
    }

    #[test]
    fn receive_non_get_con() {
        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();

        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::NonConfirmable,
            RequestCode::Get.into(),
            next_message_id,
            token,
            Vec::new(),
            Some(b"/hello"),
        );

        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(move |_, buffer| {
            let encoded_message = request.encode(buffer).unwrap();

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

        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Idle(_)
        ));
        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(
            incoming.unwrap(),
            IncomingEvent::Request(false, _)
        ));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Received(false)
        ));

        endpoint
            .incoming()
            .schedule_con_response(SuccessCode::Content.into(), Vec::new(), Some(b"world"))
            .unwrap();

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

        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::SendingCon(_)
        ));
        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::SendCon));
        // TODO We have sent the message and waiting for the ack now. Staying in SendingCon feels a little unintuitive.
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::SendingCon(_)
        ));

        let ack_id = endpoint
            .incoming_communication
            .message_buffer
            .message()
            .unwrap()
            .message_id();
        let ack = Message::<0>::new_ack(ack_id);

        stack.expect_receive().once().return_once(move |_, buffer| {
            let encoded_ack = ack.encode(buffer).unwrap();
            Ok((
                encoded_ack.message_length(),
                SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 5683),
            ))
        });

        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::Success));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Idle(_)
        ));
    }

    #[test]
    fn receive_non_get_con_timeout() {
        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();

        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::NonConfirmable,
            RequestCode::Get.into(),
            next_message_id,
            token,
            Vec::new(),
            Some(b"/hello"),
        );

        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(move |_, buffer| {
            let encoded_message = request.encode(buffer).unwrap();

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

        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Idle(_)
        ));
        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(
            incoming.unwrap(),
            IncomingEvent::Request(false, _)
        ));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Received(false)
        ));

        endpoint
            .incoming()
            .schedule_con_response(SuccessCode::Content.into(), Vec::new(), Some(b"world"))
            .unwrap();

        stack
            .expect_receive()
            .returning(move |_, _| Err(nb::Error::WouldBlock));
        stack.expect_send().returning(|_, _| Ok(()));

        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::SendCon));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::SendingCon(_)
        ));

        // 4 Retries
        clock.advance(Duration::from_secs(3));
        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::SendCon));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::SendingCon(_)
        ));

        clock.advance(Duration::from_secs(5));
        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::SendCon));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::SendingCon(_)
        ));

        clock.advance(Duration::from_secs(9));
        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::SendCon));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::SendingCon(_)
        ));

        clock.advance(Duration::from_secs(17));
        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::SendCon));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::SendingCon(_)
        ));

        // Timeout
        clock.advance(Duration::from_secs(33));
        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::Timeout));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Idle(_)
        ));
    }

    #[test]
    fn receive_piggybacked_response() {
        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();

        let message_id = endpoint.message_id_counter.next();
        let mut token = Token::default();
        token.length = TokenLength::Eight;
        endpoint.rng.read(&mut token.bytes).unwrap();
        let message: Message<'_> = Message::new(
            Type::Acknowledgement,
            SuccessCode::Content.into(),
            message_id,
            token,
            Vec::new(),
            Some(b"world"),
        );

        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(move |_, buffer| {
            let encoded_message = message.encode(buffer).unwrap();

            Ok((
                encoded_message.message_length(),
                SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 5683),
            ))
        });
        // The Response should be handled by outgoing or be unhandled, so incoming should do nothing
        // at all.

        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Idle(_)
        ));
        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::Nothing));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Idle(_)
        ));
    }

    #[test]
    fn receive_ack() {
        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();

        let message_id = endpoint.message_id_counter.next();

        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(move |_, buffer| {
            let encoded_message = Message::<0>::new_ack(message_id).encode(buffer).unwrap();

            Ok((
                encoded_message.message_length(),
                SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 5683),
            ))
        });
        // The ACK is not handled at all because it is not expected anywhere, so incoming should do
        // nothing at all.

        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Idle(_)
        ));
        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::Nothing));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Idle(_)
        ));
    }

    #[test]
    fn receive_reset() {
        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();

        let message_id = endpoint.message_id_counter.next();
        let message: Message<'_> = Message::new_rst(message_id);

        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(move |_, buffer| {
            let encoded_message = message.encode(buffer).unwrap();

            Ok((
                encoded_message.message_length(),
                SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 5683),
            ))
        });
        // The Response should be handled by outgoing, so incoming should do nothing at all.

        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Idle(_)
        ));
        let (incoming, _outgoing, _endpoint) = endpoint.process(&mut stack).unwrap();
        assert!(matches!(incoming.unwrap(), IncomingEvent::Nothing));
        assert!(matches!(
            endpoint.incoming_communication.state,
            IncomingState::Idle(_)
        ));
    }
}