rs_matter/transport/

exchange.rs

/*
 *
 *    Copyright (c) 2022-2026 Project CHIP Authors
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */

use core::fmt::{self, Display};
use core::num::NonZeroU8;
use core::pin::pin;

use either::Either as EitherIo;
use embassy_futures::select::{select, select3, Either, Either3};
use embassy_time::{Duration, Instant, Timer};

use crate::acl::Accessor;
use crate::bdx::{self, PROTO_ID_BDX};
use crate::crypto::Crypto;
use crate::dm::{AuxAclCheck, Metadata, NodeId};
use crate::error::{Error, ErrorCode};
use crate::im::{self, PROTO_ID_INTERACTION_MODEL};
use crate::sc::{self, PROTO_ID_SECURE_CHANNEL};
use crate::transport::session::Sessions;
use crate::transport::TxPayloadState;
use crate::utils::storage::pooled::{PooledBuffers, DEFAULT_BUFFER_POOL_SIZE};
use crate::utils::storage::WriteBuf;
use crate::{Matter, MatterState};

use super::mrp::{ReliableMessage, RetransEntry};
use super::network;
use super::packet::PacketHdr;
use super::plain_hdr::PlainHdr;
use super::proto_hdr::ProtoHdr;
use super::session::{Session, SessionMode};
use super::{PacketAccess, MAX_RX_BUF_SIZE, MAX_TX_BUF_SIZE};

/// Minimum buffer which should be allocated by user code that wants to pull RX messages via `Exchange::recv_into`
///
/// When the `large-buffers` feature is enabled, this tracks the larger TCP-capable packet
/// size, so that IM-level buffers (e.g. `IMBuffer`) can absorb a full Matter-over-TCP
/// message. Otherwise it stays at the UDP-sized default.
#[cfg(feature = "large-buffers")]
pub const MAX_EXCHANGE_RX_BUF_SIZE: usize = network::MAX_RX_LARGE_PACKET_SIZE;
#[cfg(not(feature = "large-buffers"))]
pub const MAX_EXCHANGE_RX_BUF_SIZE: usize = network::MAX_RX_PACKET_SIZE;

/// Maximum buffer which should be allocated and used by user code that wants to send messages via `Exchange::send`
///
/// Mirrors `MAX_EXCHANGE_RX_BUF_SIZE` with respect to the `large-buffers` feature.
#[cfg(feature = "large-buffers")]
pub const MAX_EXCHANGE_TX_BUF_SIZE: usize =
    network::MAX_TX_LARGE_PACKET_SIZE - PacketHdr::HDR_RESERVE - PacketHdr::TAIL_RESERVE;
#[cfg(not(feature = "large-buffers"))]
pub const MAX_EXCHANGE_TX_BUF_SIZE: usize =
    network::MAX_TX_PACKET_SIZE - PacketHdr::HDR_RESERVE - PacketHdr::TAIL_RESERVE;

/// A protocol payload buffer, sized to hold a full exchange RX message
/// ([`MAX_EXCHANGE_RX_BUF_SIZE`]).
///
/// This is the central buffer type that both [`IMBuffer`](crate::im::IMBuffer)
/// and [`BdxBuffer`](crate::bdx::BdxBuffer) alias, so a single
/// [`PooledBuffers`](crate::utils::storage::pooled::PooledBuffers) pool can be
/// shared across the data model and BDX.
pub type Buffer = crate::utils::storage::Vec<u8, MAX_EXCHANGE_RX_BUF_SIZE>;

/// A [`PooledBuffers`] pool pre-configured with Matter defaults: it holds
/// [`Buffer`]s (the central exchange-sized buffer) and [`DEFAULT_BUFFER_POOL_SIZE`]
/// of them, behind the default Matter raw mutex.
pub type MatterBuffers<const N: usize = DEFAULT_BUFFER_POOL_SIZE> = PooledBuffers<Buffer, N>;

/// An exchange identifier, uniquely identifying a session and an exchange within that session for a given Matter stack.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub struct ExchangeId(u32);

impl ExchangeId {
    pub(crate) fn new(session_id: u32, exchange_index: usize) -> Self {
        if session_id > 0x0fff_ffff {
            panic!("Session ID out of range");
        }

        if exchange_index >= 16 {
            panic!("Exchange index out of range");
        }

        Self(((exchange_index as u32) << 28) | session_id)
    }

    pub(crate) fn session_id(&self) -> u32 {
        self.0 & 0x0fff_ffff
    }

    pub(crate) fn exchange_index(&self) -> usize {
        (self.0 >> 28) as _
    }

    /// Get the session associated with this exchange from the given sessions store.
    ///
    /// ATTENTION: This method will panic if the session is not found in the store, so make sure to only call it when you are sure the session exists.
    pub(crate) fn session<'a>(&self, sessions: &'a mut Sessions) -> &'a mut Session {
        unwrap!(sessions.get(self.session_id()))
    }

    /// Get the exchange state associated with this exchange from the given sessions store.
    pub(crate) fn exch<'a>(&self, session: &'a mut Session) -> &'a mut ExchangeState {
        unwrap!(session.exchanges[self.exchange_index()].as_mut())
    }

    pub(crate) fn display<'a>(&'a self, session: &'a Session) -> ExchangeIdDisplay<'a> {
        ExchangeIdDisplay { id: self, session }
    }

    async fn recv<'a>(&self, matter: &'a Matter<'a>) -> Result<RxMessage<'a>, Error> {
        self.check_no_pending_retrans(matter)?;

        loop {
            let mut recv = pin!(matter.transport().get_if_rx(|packet| {
                if packet.buf.is_empty() {
                    false
                } else {
                    let for_us = self.with_state(matter, |state| {
                        let sess = self.session(&mut state.sessions);
                        if sess.is_for_rx(&packet.peer, &packet.header.plain) {
                            let exch = self.exch(sess);

                            return Ok(exch.is_for_rx(&packet.header.proto));
                        }

                        Ok(false)
                    });

                    for_us.unwrap_or(true)
                }
            }));

            let mut session_removed = pin!(matter.transport().wait_session_removed());

            let mut timeout = pin!(Timer::after(Duration::from_millis(
                RetransEntry::new(matter.dev_det().sai, 0).max_delay_ms() * 3 / 2
            )));

            match select3(&mut recv, &mut session_removed, &mut timeout).await {
                Either3::First(mut packet) => {
                    packet.clear_on_drop(true);

                    self.check_no_pending_retrans(matter)?;

                    break Ok(RxMessage(packet));
                }
                Either3::Second(_) => {
                    // Session removed

                    // Bail out if it was ours
                    self.with_state(matter, |_| Ok(()))?;

                    // If not, go back waiting for a packet
                    continue;
                }
                Either3::Third(_) => {
                    // Timeout waiting for an answer from the other peer
                    Err(ErrorCode::RxTimeout)?;
                }
            };
        }
    }

    /// Gets access to the TX buffer of the Matter stack for constructing a new TX message.
    /// If the TX buffer is not available, the method will wait indefinitely until it becomes available.
    ///
    /// NOTE:
    /// This is a low-level method that leaves the re-transmission logic on the shoulders of the user.
    /// Therefore, prefer using `Exchange::sender`, `Exchange::send` or `Exchange::send_with` instead.
    ///
    /// Note also that if the uderlying session or exchange tracked by the Matter stack is dropped
    /// (say, because of lack of resources or a hard networking error), the method will return an error.
    async fn init_send<'a>(&self, matter: &'a Matter<'a>) -> Result<TxMessage<'a>, Error> {
        self.with_state(matter, |_| Ok(()))?;

        let mut packet = matter
            .transport
            .get_if_tx(|packet| {
                packet.buf.is_empty() || self.with_state(matter, |_| Ok(())).is_err()
            })
            .await;

        // TODO: Resizing might be a bit expensive with large buffers
        unwrap!(packet.buf.resize_default(MAX_TX_BUF_SIZE));

        packet.clear_on_drop(true);

        let tx = TxMessage {
            exchange_id: *self,
            matter,
            packet,
        };

        self.with_state(matter, |_| Ok(()))?;

        Ok(tx)
    }

    /// Waits until the other side acknowledges the last message sent on this exchange,
    /// or until time for a re-transmission had come.
    ///
    /// If the last sent message was not using the MRP protocol, the method will return immediately with `TxOutcome::Done`.
    ///
    /// NOTE:
    /// This is a low-level method that leaves the re-transmission logic on the shoulders of the user.
    /// Therefore, prefer using `Exchange::sender`, `Exchange::send` or `Exchange::send_with` instead.
    ///
    /// Note also that if the uderlying session or exchange tracked by the Matter stack is dropped
    /// (say, because of lack of resources or a hard networking error), the method will return an error.
    async fn wait_tx<'a>(&self, matter: &'a Matter<'a>) -> Result<TxOutcome, Error> {
        if let Some(delay) = self.retrans_delay_ms(matter)? {
            let expired = unwrap!(Instant::now().checked_add(Duration::from_millis(delay)));

            loop {
                let mut notification = pin!(self.internal_wait_ack(matter));
                let mut session_removed = pin!(matter.transport().wait_session_removed());
                let mut timer = pin!(Timer::at(expired));

                if !matches!(
                    select3(&mut notification, &mut session_removed, &mut timer).await,
                    Either3::Second(_)
                ) {
                    break;
                }

                // Bail out if the removed session was ours
                self.with_state(matter, |_| Ok(()))?;
            }

            if self.retrans_delay_ms(matter)?.is_some() {
                Ok(TxOutcome::Retransmit)
            } else {
                Ok(TxOutcome::Done)
            }
        } else {
            Ok(TxOutcome::Done)
        }
    }

    fn accessor<'a>(
        &self,
        matter: &'a Matter<'a>,
        aux_acl_enabled: bool,
    ) -> Result<Accessor<'a>, Error> {
        self.with_state(matter, |state| {
            let sess = self.session(&mut state.sessions);

            Ok(Accessor::for_session(sess, matter, aux_acl_enabled))
        })
    }

    fn with_state<'a, F, T>(&self, matter: &'a Matter<'a>, f: F) -> Result<T, Error>
    where
        F: FnOnce(&mut MatterState) -> Result<T, Error>,
    {
        self.with_state_ex(matter, f)
    }

    fn with_state_ex<'a, F, T, E>(&self, matter: &'a Matter<'a>, f: F) -> Result<T, E>
    where
        F: FnOnce(&mut MatterState) -> Result<T, E>,
        E: From<Error>,
    {
        matter.with_state(|state| {
            if state.sessions.get(self.session_id()).is_some() {
                f(state)
            } else {
                warn!("Exchange {}: No session", self);
                Err(Error::from(ErrorCode::NoSession).into())
            }
        })
    }

    async fn internal_wait_ack<'a>(&self, matter: &'a Matter<'a>) -> Result<(), Error> {
        matter
            .transport
            .get_if_rx(|_| {
                self.retrans_delay_ms(matter)
                    .map(|retrans| retrans.is_none())
                    .unwrap_or(true)
            })
            .await;

        self.with_state(matter, |_| Ok(()))
    }

    fn retrans_delay_ms<'a>(&self, matter: &'a Matter<'a>) -> Result<Option<u64>, Error> {
        self.with_state(matter, |state| {
            let sess = self.session(&mut state.sessions);
            let exch = self.exch(sess);

            let mut jitter_rand = [0; 1];
            // TODO XXX FIXME matter.rand()(&mut jitter_rand);
            jitter_rand[0] = 100;

            Ok(exch.retrans_delay_ms(jitter_rand[0]))
        })
    }

    fn check_no_pending_retrans<'a>(&self, matter: &'a Matter<'a>) -> Result<(), Error> {
        self.with_state(matter, |state| {
            let sess = self.session(&mut state.sessions);
            let exch = self.exch(sess);

            if exch.mrp.is_retrans_pending() {
                error!("Exchange {}: Retransmission pending", self.display(sess));
                Err(ErrorCode::InvalidState)?;
            }

            Ok(())
        })
    }

    fn pending_retrans<'a>(&self, matter: &'a Matter<'a>) -> Result<bool, Error> {
        Ok(self.retrans_delay_ms(matter)?.is_some())
    }

    fn pending_ack<'a>(&self, matter: &'a Matter<'a>) -> Result<bool, Error> {
        self.with_state(matter, |state| {
            let sess = self.session(&mut state.sessions);
            let exch = self.exch(sess);

            Ok(exch.mrp.is_ack_pending())
        })
    }
}

impl Display for ExchangeId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}::{}", self.session_id(), self.exchange_index())
    }
}

#[cfg(feature = "defmt")]
impl defmt::Format for ExchangeId {
    fn format(&self, f: defmt::Formatter<'_>) {
        defmt::write!(f, "{}::{}", self.session_id(), self.exchange_index())
    }
}

/// A display wrapper for `ExchangeId` which also displays
/// the packet session ID, packet peer session ID and packet exchange ID.
pub struct ExchangeIdDisplay<'a> {
    id: &'a ExchangeId,
    session: &'a Session,
}

impl Display for ExchangeIdDisplay<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let state = self.session.exchanges[self.id.exchange_index()].as_ref();

        if let Some(state) = state {
            write!(
                f,
                "{} [SID:{:x},RSID:{:x},EID:{:x}]",
                self.id,
                self.session.get_local_sess_id(),
                self.session.get_peer_sess_id(),
                state.exch_id
            )
        } else {
            // This should never happen, as that would mean we have invalid exchange index
            // but let's not crash when displaying that
            write!(f, "{}???", self.id)
        }
    }
}

#[cfg(feature = "defmt")]
impl defmt::Format for ExchangeIdDisplay<'_> {
    fn format(&self, f: defmt::Formatter<'_>) {
        let state = self.session.exchanges[self.id.exchange_index()].as_ref();

        if let Some(state) = state {
            defmt::write!(
                f,
                "{} [SID:{:x},RSID:{:x},EID:{:x}]",
                self.id,
                self.session.get_local_sess_id(),
                self.session.get_peer_sess_id(),
                state.exch_id
            )
        } else {
            // This should never happen, as that would mean we have invalid exchange index
            // but let's not crash when displaying that
            defmt::write!(f, "{}???", self.id)
        }
    }
}

#[derive(Debug, PartialEq, Eq, Copy, Clone, Default)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub(crate) enum InitiatorState {
    #[default]
    Owned,
    Dropped,
}

#[derive(Debug, PartialEq, Eq, Copy, Clone, Default)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub(crate) enum ResponderState {
    #[default]
    AcceptPending,
    Owned,
    Dropped,
}

#[derive(Debug, PartialEq, Eq, Copy, Clone)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub(crate) enum Role {
    Initiator(InitiatorState),
    Responder(ResponderState),
}

impl Role {
    pub fn is_dropped_state(&self) -> bool {
        match self {
            Self::Initiator(state) => *state == InitiatorState::Dropped,
            Self::Responder(state) => *state == ResponderState::Dropped,
        }
    }

    pub fn set_dropped_state(&mut self) {
        match self {
            Self::Initiator(state) => *state = InitiatorState::Dropped,
            Self::Responder(state) => *state = ResponderState::Dropped,
        }
    }
}

#[derive(Debug)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub(crate) struct ExchangeState {
    pub(crate) exch_id: u16,
    pub(crate) role: Role,
    pub(crate) mrp: ReliableMessage,
    /// The Global Group Encrypted Data Message Counter value reserved for
    /// this exchange's (single) group data message - already covered by a
    /// durably stored boundary, see [`Exchange::initiate_group`].
    ///
    /// Per-exchange rather than per-session, so that two concurrent sends to
    /// the same group cannot consume each other's reservation. `None` for
    /// every other exchange kind.
    #[cfg(feature = "groups")]
    pub(crate) group_data_ctr: Option<u32>,
}

impl ExchangeState {
    pub fn is_for_rx(&self, rx_proto: &ProtoHdr) -> bool {
        self.exch_id == rx_proto.exch_id
            && rx_proto.is_initiator() == matches!(self.role, Role::Responder(_))
    }

    pub fn post_recv(&mut self, rx_plain: &PlainHdr, rx_proto: &ProtoHdr) -> Result<(), Error> {
        self.mrp.post_recv(rx_plain, rx_proto)?;

        Ok(())
    }

    pub fn pre_send(
        &mut self,
        tx_plain: &PlainHdr,
        tx_proto: &mut ProtoHdr,
        session_active_interval_ms: Option<u32>,
        session_idle_interval_ms: Option<u32>,
    ) -> Result<(), Error> {
        if matches!(self.role, Role::Initiator(_)) {
            tx_proto.set_initiator();
        } else {
            tx_proto.unset_initiator();
        }

        tx_proto.exch_id = self.exch_id;

        self.mrp.pre_send(
            tx_plain,
            tx_proto,
            session_active_interval_ms,
            session_idle_interval_ms,
        )
    }

    pub fn retrans_delay_ms(&mut self, jitter_rand: u8) -> Option<u64> {
        self.mrp
            .retrans
            .as_ref()
            .map(|retrans| retrans.delay_ms(jitter_rand))
    }
}

/// Meta-data when sending/receving messages via an Exchange.
/// Basically, the protocol ID, the protocol opcode and whether the message should be set in a reliable manner.
#[derive(Debug, Eq, PartialEq, Copy, Clone)]
pub struct MessageMeta {
    pub proto_id: u16,
    pub proto_opcode: u8,
    pub reliable: bool,
}

impl MessageMeta {
    // Create a new message meta-data instance
    pub const fn new(proto_id: u16, proto_opcode: u8, reliable: bool) -> Self {
        Self {
            proto_id,
            proto_opcode,
            reliable,
        }
    }

    /// Try to cast the protocol opcode to a specific type
    pub fn opcode<T: num::FromPrimitive>(&self) -> Result<T, Error> {
        num::FromPrimitive::from_u8(self.proto_opcode).ok_or(ErrorCode::InvalidOpcode.into())
    }

    /// Check if the protocol opcode is equal to a specific value
    pub fn check_opcode<T: num::FromPrimitive + PartialEq>(&self, opcode: T) -> Result<(), Error> {
        if self.opcode::<T>()? == opcode {
            Ok(())
        } else {
            Err(ErrorCode::Invalid.into())
        }
    }

    /// Create an instance from a ProtoHdr instance
    pub fn from(proto: &ProtoHdr) -> Self {
        Self {
            proto_id: proto.proto_id,
            proto_opcode: proto.proto_opcode,
            reliable: proto.is_reliable(),
        }
    }

    /// Set the protocol ID and opcode into a ProtoHdr instance
    pub fn set_into(&self, proto: &mut ProtoHdr) {
        proto.proto_id = self.proto_id;
        proto.proto_opcode = self.proto_opcode;
        proto.set_vendor(None);

        if self.reliable {
            proto.set_reliable();
        } else {
            proto.unset_reliable();
        }
    }

    pub fn reliable(self, reliable: bool) -> Self {
        Self { reliable, ..self }
    }

    /// Utility method to check if the specific proto opcode in the instance is expecting a TLV payload.
    pub(crate) fn is_tlv(&self) -> bool {
        match self.proto_id {
            PROTO_ID_SECURE_CHANNEL => self
                .opcode::<sc::OpCode>()
                .ok()
                .map(|op| op.is_tlv())
                .unwrap_or(false),
            PROTO_ID_INTERACTION_MODEL => self
                .opcode::<im::OpCode>()
                .ok()
                .map(|op| op.is_tlv())
                .unwrap_or(false),
            _ => false,
        }
    }

    /// Utility method to check if the protocol is Secure Channel, and the opcode is a standalone ACK (`MrpStandaloneAck`).
    pub(crate) fn is_standalone_ack(&self) -> bool {
        self.proto_id == PROTO_ID_SECURE_CHANNEL
            && self.proto_opcode == sc::OpCode::MRPStandAloneAck as u8
    }

    /// Utility method to check if the protocol is Secure Channel, and the opcode is Status.
    pub(crate) fn is_sc_status(&self) -> bool {
        self.proto_id == PROTO_ID_SECURE_CHANNEL
            && self.proto_opcode == sc::OpCode::StatusReport as u8
    }

    /// Utility method to check if the protocol is Secure Channel, and the opcode is a new session request.
    pub(crate) fn is_new_session(&self) -> bool {
        self.proto_id == PROTO_ID_SECURE_CHANNEL
            && (self.proto_opcode == sc::OpCode::PBKDFParamRequest as u8
                || self.proto_opcode == sc::OpCode::CASESigma1 as u8)
    }

    /// Utility method to check if the meta-data indicates a new exchange
    pub(crate) fn is_new_exchange(&self) -> bool {
        // Don't create new exchanges for standalone ACKs and for SC status codes
        !self.is_standalone_ack() && !self.is_sc_status()
    }

    /// Whether this message is a *control* message. Control messages
    /// travel on a separate counter space (the "control message
    /// counter") and set the `C` bit in the plain-header Security
    /// Flags. Today only the MCSP opcodes qualify; the transport
    /// layer flips the `C` bit on the outgoing packet based on this.
    pub fn is_control_msg(&self) -> bool {
        self.proto_id == PROTO_ID_SECURE_CHANNEL
            && (self.proto_opcode == sc::OpCode::MsgCounterSyncReq as u8
                || self.proto_opcode == sc::OpCode::MsgCounterSyncResp as u8)
    }
}

impl Display for MessageMeta {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self.proto_id {
            PROTO_ID_SECURE_CHANNEL => {
                if let Ok(opcode) = self.opcode::<sc::OpCode>() {
                    write!(f, "SC::{:?}", opcode)
                } else {
                    write!(f, "SC::{:02x}", self.proto_opcode)
                }
            }
            PROTO_ID_INTERACTION_MODEL => {
                if let Ok(opcode) = self.opcode::<im::OpCode>() {
                    write!(f, "IM::{:?}", opcode)
                } else {
                    write!(f, "IM::{:02x}", self.proto_opcode)
                }
            }
            PROTO_ID_BDX => {
                if let Ok(opcode) = self.opcode::<bdx::OpCode>() {
                    write!(f, "BDX::{:?}", opcode)
                } else {
                    write!(f, "BDX::{:02x}", self.proto_opcode)
                }
            }
            _ => write!(f, "{:02x}::{:02x}", self.proto_id, self.proto_opcode),
        }
    }
}

#[cfg(feature = "defmt")]
impl defmt::Format for MessageMeta {
    fn format(&self, f: defmt::Formatter<'_>) {
        match self.proto_id {
            PROTO_ID_SECURE_CHANNEL => {
                if let Ok(opcode) = self.opcode::<sc::OpCode>() {
                    defmt::write!(f, "SC::{:?}", opcode)
                } else {
                    defmt::write!(f, "SC::{:02x}", self.proto_opcode)
                }
            }
            PROTO_ID_INTERACTION_MODEL => {
                if let Ok(opcode) = self.opcode::<im::OpCode>() {
                    defmt::write!(f, "IM::{:?}", opcode)
                } else {
                    defmt::write!(f, "IM::{:02x}", self.proto_opcode)
                }
            }
            PROTO_ID_BDX => {
                if let Ok(opcode) = self.opcode::<bdx::OpCode>() {
                    defmt::write!(f, "BDX::{:?}", opcode)
                } else {
                    defmt::write!(f, "BDX::{:02x}", self.proto_opcode)
                }
            }
            _ => defmt::write!(f, "{:02x}::{:02x}", self.proto_id, self.proto_opcode),
        }
    }
}

/// An RX message pending on an `Exchange` instance.
pub struct RxMessage<'a>(PacketAccess<'a, MAX_RX_BUF_SIZE>);

impl RxMessage<'_> {
    /// Get the meta-data of the pending message
    pub fn meta(&self) -> MessageMeta {
        MessageMeta::from(&self.0.header.proto)
    }

    /// Get the payload of the pending message
    pub fn payload(&self) -> &[u8] {
        &self.0.buf[self.0.payload_start..]
    }
}

/// Accessor to the TX message buffer of the underlying Matter transport stack.
///
/// This is used to construct a new TX message to be sent on an `Exchange` instance.
///
/// NOTE: It is strongly advised to use the `TxMessage` accessor in combination with the `Sender` utility,
/// which takes care of all message retransmission logic. Alternatively, one can use the
/// `Exchange::send` or `Exchange::send_with` which also take care of re-transmissions.
pub struct TxMessage<'a> {
    exchange_id: ExchangeId,
    matter: &'a Matter<'a>,
    packet: PacketAccess<'a, MAX_TX_BUF_SIZE>,
}

impl TxMessage<'_> {
    /// Get a reference to the payload buffer of the TX message being built
    pub fn payload(&mut self) -> &mut [u8] {
        &mut self.packet.buf[PacketHdr::HDR_RESERVE..MAX_TX_BUF_SIZE - PacketHdr::TAIL_RESERVE]
    }

    /// Complete and send a TX message by providing:
    /// - The payload size that was filled-in by user code in the payload buffer returned by `TxMessage::payload`
    /// - The TX message meta-data
    pub fn complete<M>(
        mut self,
        payload_start: usize,
        payload_end: usize,
        meta: M,
    ) -> Result<(), Error>
    where
        M: Into<MessageMeta>,
    {
        if payload_start > payload_end
            || payload_end - payload_start
                > MAX_TX_BUF_SIZE - PacketHdr::HDR_RESERVE - PacketHdr::TAIL_RESERVE
        {
            Err(ErrorCode::Invalid)?;
        }

        let meta: MessageMeta = meta.into();

        self.packet.header.reset();

        meta.set_into(&mut self.packet.header.proto);

        self.matter.with_state(|state| {
            let session = state
                .sessions
                .get(self.exchange_id.session_id())
                .ok_or(ErrorCode::NoSession)?;

            // The session's `peer_active_interval_ms` / `peer_idle_interval_ms`
            // are seeded from our own `BasicInfoConfig` at session-creation
            // time and overwritten once Sigma1 / Sigma2 / PBKDFParamRequest /
            // PBKDFParamResponse delivers a real peer value, so MRP
            // retransmission backoff to this peer reflects whichever is
            // more accurate (Matter Core spec).
            let (peer, retransmission) = session.pre_send(
                Some(self.exchange_id.exchange_index()),
                &mut self.packet.header,
                Some(session.get_peer_active_interval_ms()),
                Some(session.get_peer_idle_interval_ms()),
            )?;

            self.packet.peer = peer;
            self.packet.payload_start = PacketHdr::HDR_RESERVE + payload_start;
            self.packet
                .buf
                .truncate(PacketHdr::HDR_RESERVE + payload_end);
            self.packet.tx_info.payload_state = TxPayloadState::NotEncoded {
                session_id: session.id,
            };
            self.packet.tx_info.retransmission = retransmission;
            self.packet.clear_on_drop(false);

            Ok(())
        })
    }
}

/// Outcome from calling `Exchange::wait_tx`
#[derive(Copy, Clone, Eq, PartialEq, Debug)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub enum TxOutcome {
    /// The other side has acknowledged the last message or the last message was not using the MRP protocol
    /// Stop re-sending.
    Done,
    /// Need to re-send the last message.
    Retransmit,
}

impl TxOutcome {
    /// Check if the outcome is `Done`
    pub const fn is_done(&self) -> bool {
        matches!(self, Self::Done)
    }
}

pub struct SenderTx<'a, 'b> {
    sender: &'b mut Sender<'a>,
    message: TxMessage<'a>,
}

impl SenderTx<'_, '_> {
    pub fn split(&mut self) -> (&Exchange<'_>, &mut [u8]) {
        (self.sender.exchange, self.message.payload())
    }

    pub fn payload(&mut self) -> &mut [u8] {
        self.message.payload()
    }

    pub fn complete(
        self,
        payload_start: usize,
        payload_end: usize,
        meta: MessageMeta,
    ) -> Result<(), Error> {
        self.message.complete(payload_start, payload_end, meta)?;

        self.sender.initial = false;

        Ok(())
    }
}

/// Utility struct for sending a message with potential retransmissions.
pub struct Sender<'a> {
    exchange: &'a Exchange<'a>,
    initial: bool,
    complete: bool,
}

impl<'a> Sender<'a> {
    fn new(exchange: &'a Exchange<'a>) -> Result<Self, Error> {
        exchange.id.check_no_pending_retrans(exchange.matter)?;

        Ok(Self {
            exchange,
            initial: true,
            complete: false,
        })
    }

    /// Get the TX buffer of the underlying Matter stack for (re)constructing a new TX message,
    /// waiting for the TX buffer to become available, if it is not.
    ///
    /// If the method returns `None`, it means that the message was already acknowledged by the other side,
    /// or that the message does not need acknowledgement and re-transmissions.
    ///
    /// When called for the first time, the method will always return a `Some` value, as the message has not been sent even once yet.
    /// Once the method returns `None`, it will always return `None` on subsequent calls, as the message has been acknowledged by the other side.
    ///
    /// Example:
    /// ```ignore
    /// let exchange = ...;
    ///
    /// let sender = exchange.sender()?;
    ///
    /// while let Some(mut tx) = sender.tx().await? {
    ///     let (exchange, payload) = tx.split()?;
    ///
    ///     // Write the message payload in the `payload` `&mut [u8]` slice
    ///     // On every iteration of the loop, write the _same_ payload (as message re-transmission is idempotent w.r.t. the message)
    ///     ...
    ///
    ///     // Complete the payload by providing `MessageMeta`, payload start and payload end
    ///     // On every iteration of the loop, proide the _same_ meta-data (as message re-transmission is idempotent w.r.t. the message)
    ///     let meta = ...;
    ///     let payload_start = ...;
    ///     let payload_end = ...;
    ///
    ///     tx.complete(payload_start, payload_end, meta)?;
    /// }
    /// ```
    pub async fn tx(&mut self) -> Result<Option<SenderTx<'a, '_>>, Error> {
        trace!(
            "Sender::tx called, initial={}, complete={}",
            self.initial,
            self.complete
        );
        if self.complete {
            trace!("Sender::tx - already complete, returning None");
            return Ok(None);
        }

        if !self.initial {
            trace!("Sender::tx - not initial, calling wait_tx");
            let outcome = self.exchange.id.wait_tx(self.exchange.matter).await?;
            trace!("Sender::tx - wait_tx returned {:?}", outcome);
            if outcome.is_done() {
                // No need to re-transmit
                self.complete = true;
                trace!("Sender::tx - ACK received, returning None");
                return Ok(None);
            }
            trace!("Sender::tx - need to retransmit");
        }

        let id = self.exchange.id;
        let matter = self.exchange.matter;

        trace!("Sender::tx - calling init_send");
        let tx = id.init_send(matter).await?;
        trace!("Sender::tx - init_send returned");

        if self.initial || id.pending_retrans(matter)? {
            trace!("Sender::tx - returning Some(SenderTx)");
            Ok(Some(SenderTx {
                sender: self,
                message: tx,
            }))
        } else {
            self.complete = true;
            trace!("Sender::tx - no pending retrans, returning None");
            Ok(None)
        }
    }
}

/// Owned-`Self` counterpart to [`SenderTx`].
///
/// Holds the [`Exchange`] by value (rather than via a `&mut` through
/// the parent [`Sender`]), so consumers that consume their exchange
/// can drive the retransmit loop without a self-referential struct.
/// Returned by [`OwnedSender::tx`] when the framework needs the
/// message bytes (re-)built into a fresh TX slot.
///
/// Pair with [`SenderTx`] for the borrowed-self mirror.
pub struct OwnedSenderTx<'a> {
    exchange: Exchange<'a>,
    message: TxMessage<'a>,
}

impl<'a> OwnedSenderTx<'a> {
    /// Get a `(borrowed-exchange, payload-slice)` pair for inspecting
    /// the exchange and writing the message bytes.
    pub fn split(&mut self) -> (&Exchange<'_>, &mut [u8]) {
        (&self.exchange, self.message.payload())
    }

    /// Get a mutable reference to the payload slice the bytes go into.
    pub fn payload(&mut self) -> &mut [u8] {
        self.message.payload()
    }

    /// Commit the bytes currently in the slot. The framework dispatches
    /// the wire-level send asynchronously; this call returns the
    /// [`OwnedSender`] so the caller can ask for the next event
    /// (retransmit, ACK).
    pub fn complete(
        self,
        payload_start: usize,
        payload_end: usize,
        meta: MessageMeta,
    ) -> Result<OwnedSender<'a>, Error> {
        self.message.complete(payload_start, payload_end, meta)?;

        Ok(OwnedSender {
            exchange: self.exchange,
            initial: false,
            complete: false,
        })
    }
}

/// Owned-`Self` counterpart to [`Sender`].
///
/// Consumes the [`Exchange`] on construction (via
/// [`Exchange::into_sender`]) and returns it back to the caller when
/// the retransmit loop is done (i.e. when [`OwnedSender::tx`]
/// completes with `Either::Right`, meaning the message has been
/// ACK-ed by the peer or did not require ACK).
///
/// Use cases:
/// - The Interaction Model client (`im::client::ImClient`) consumes
///   an exchange to drive a single IM transaction end-to-end, and
///   wants to walk the retransmit loop while still owning the
///   `Exchange` for the subsequent receive phase. The owned-`Self`
///   shape avoids a self-referential struct that would otherwise hold
///   both an owned `Exchange` and a `&mut`-Sender on it.
///
/// User loop pattern (mirrors [`Sender`]):
///
/// ```ignore
/// let mut sender = exchange.into_sender()?;
/// let exchange = loop {
///     match sender.tx().await? {
///         Either::Left(mut slot) => {
///             let (_, payload) = slot.split();
///             // Write the message bytes (same every iteration —
///             // retransmission is idempotent w.r.t. the message).
///             let (start, end, meta) = /* … */;
///             sender = slot.complete(start, end, meta)?;
///         }
///         Either::Right(exchange) => break exchange,
///     }
/// };
/// // `exchange` is yours again.
/// ```
pub struct OwnedSender<'a> {
    exchange: Exchange<'a>,
    initial: bool,
    complete: bool,
}

impl<'a> OwnedSender<'a> {
    fn new(exchange: Exchange<'a>) -> Result<Self, Error> {
        exchange.id.check_no_pending_retrans(exchange.matter)?;

        Ok(Self {
            exchange,
            initial: true,
            complete: false,
        })
    }

    /// Next event from the framework. Owned-`Self` mirror of
    /// [`Sender::tx`]:
    /// - `Either::Left(slot)` — a TX slot is ready; (re-)build the
    ///   message bytes into it and call `slot.complete(...)` to get
    ///   the `OwnedSender` back for the next iteration.
    /// - `Either::Right(exchange)` — the message has been ACK-ed (or
    ///   did not need an ACK); the loop is done and the exchange is
    ///   returned to the caller.
    pub async fn tx(mut self) -> Result<EitherIo<OwnedSenderTx<'a>, Exchange<'a>>, Error> {
        trace!(
            "OwnedSender::tx called, initial={}, complete={}",
            self.initial,
            self.complete
        );
        if self.complete {
            trace!("OwnedSender::tx - already complete, returning exchange");
            return Ok(EitherIo::Right(self.exchange));
        }

        if !self.initial {
            trace!("OwnedSender::tx - not initial, calling wait_tx");
            let outcome = self.exchange.id.wait_tx(self.exchange.matter).await?;
            trace!("OwnedSender::tx - wait_tx returned {:?}", outcome);
            if outcome.is_done() {
                // No need to re-transmit
                self.complete = true;
                trace!("OwnedSender::tx - ACK received, returning exchange");
                return Ok(EitherIo::Right(self.exchange));
            }
            trace!("OwnedSender::tx - need to retransmit");
        }

        let id = self.exchange.id;
        let matter = self.exchange.matter;

        trace!("OwnedSender::tx - calling init_send");
        let tx = id.init_send(matter).await?;
        trace!("OwnedSender::tx - init_send returned");

        if self.initial || id.pending_retrans(matter)? {
            trace!("OwnedSender::tx - returning Left(OwnedSenderTx)");
            Ok(EitherIo::Left(OwnedSenderTx {
                exchange: self.exchange,
                message: tx,
            }))
        } else {
            trace!("OwnedSender::tx - no pending retrans, returning exchange");
            Ok(EitherIo::Right(self.exchange))
        }
    }
}

/// An exchange within a Matter stack, representing a session and an exchange within that session.
///
/// This is the main API for sending and receiving messages within the Matter stack.
/// Used by upper-level layers like the Secure Channel and Interaction Model.
pub struct Exchange<'a> {
    id: ExchangeId,
    matter: &'a Matter<'a>,
    rx: Option<RxMessage<'a>>,
}

impl<'a> Exchange<'a> {
    pub(crate) const fn new(id: ExchangeId, matter: &'a Matter<'a>) -> Self {
        Self {
            id,
            matter,
            rx: None,
        }
    }

    /// Get the Id of the exchange
    pub fn id(&self) -> ExchangeId {
        self.id
    }

    /// Get the Matter stack instance associated with this exchange
    pub fn matter(&self) -> &'a Matter<'a> {
        self.matter
    }

    /// Open an exchange over a CASE session to an already-commissioned node.
    ///
    /// If a CASE session for `(fabric_idx, peer_node_id)` already exists, an
    /// exchange is opened on it directly via [`Exchange::initiate_for_session`]
    /// (the common case - session and peer address reused, no mDNS). Otherwise
    /// the peer's operational address is resolved over mDNS and a fresh CASE
    /// session is established (driving [`crate::sc::case::CaseInitiator`]) before
    /// the exchange is opened on it; the peer's MRP/session parameters advertised
    /// in the mDNS TXT records seed the session.
    ///
    /// Establishing requires a running mDNS responder (e.g.
    /// `BuiltinMdns::run`) to service the resolve; without one the
    /// resolve times out and this returns [`ErrorCode::NotFound`].
    ///
    /// With the `case-responder-only` feature the establish path is compiled
    /// out: only an existing session is reused, and if none exists this returns
    /// [`ErrorCode::NoSession`] instead of opening a new one. This lets the linker
    /// drop the CASE initiator and mDNS resolver from a node that never needs to
    /// initiate CASE (a pure accessory that only reports over sessions its peers
    /// established).
    #[inline(always)]
    pub async fn initiate<C: Crypto>(
        matter: &'a Matter<'a>,
        crypto: C,
        fabric_idx: NonZeroU8,
        peer_node_id: NodeId,
    ) -> Result<Self, Error> {
        matter
            .transport
            .initiate(matter, crypto, fabric_idx, peer_node_id)
            .await
    }

    /// Open an exchange over a **PASE** session to a not-yet-commissioned node
    /// at the given peer address (use-case 2).
    ///
    /// If a PASE session **to that peer** already exists, an exchange is opened on
    /// it directly via [`Exchange::initiate_for_session`]. Otherwise a new PASE
    /// session is established: a plaintext session is opened and the PASE protocol
    /// ([`crate::sc::pase::PaseInitiator`]) is run with `passcode`, then an
    /// exchange is opened on the resulting PASE session.
    ///
    /// Reuse is keyed by peer address (not a single global PASE session), so a
    /// commissioner can drive several concurrent commissionings.
    ///
    /// The peer's MRP/session parameters are negotiated by PASE itself
    /// (PBKDFParamRequest/Response).
    ///
    /// Discovery of the address is out of scope here (it may come from mDNS - see
    /// [`crate::transport::Transport::browse_commissionable`] - or from a BLE/BTP
    /// advertisement, etc.); this method is transport-agnostic and takes the
    /// already-known address.
    #[inline(always)]
    pub async fn initiate_pase<C: Crypto>(
        matter: &'a Matter<'a>,
        crypto: C,
        peer_addr: network::Address,
        passcode: u32,
    ) -> Result<Self, Error> {
        matter
            .transport
            .initiate_pase(matter, crypto, peer_addr, passcode)
            .await
    }

    /// Create a new initiator exchange on the provided Matter stack for the provided session ID.
    #[inline(always)]
    pub fn initiate_for_session<C: Crypto>(
        matter: &'a Matter<'a>,
        crypto: C,
        session_id: u32,
    ) -> Result<Self, Error> {
        matter
            .transport()
            .initiate_for_session(matter, crypto, session_id)
    }

    /// Create a new initiator exchange for sending group DATA messages to
    /// `(fab_idx, group_id)` — encrypted with the group's active operational
    /// key and addressed at the group's multicast address.
    ///
    /// Group messages are fire-and-forget: no MRP, no acknowledgements and
    /// no responses — so the exchange is only good for *sending* (e.g. an
    /// Interaction Model invoke with `SuppressResponse`); do not wait for
    /// replies on it. The group's security material (a `GroupKeyMap` entry
    /// plus its key set) must be provisioned on the fabric, else this fails
    /// with [`ErrorCode::NotFound`].
    #[cfg(feature = "groups")]
    pub fn initiate_group<C: Crypto, K: crate::persist::KvBlobStoreAccess>(
        matter: &'a Matter<'a>,
        crypto: C,
        kv: K,
        fab_idx: NonZeroU8,
        group_id: u16,
    ) -> Result<Self, Error> {
        let (session_id, group_data_ctr, boundary) = matter.with_state(|state| {
            let session = state.sessions.get_or_create_for_group_tx(
                &crypto,
                &state.fabrics,
                fab_idx,
                group_id,
                matter.dev_det(),
            )?;
            let session_id = session.id;

            // One group data message per exchange, so exactly one counter
            // value is reserved here.
            let (ctr, boundary) = state.sessions.reserve_global_group_data_ctr(&crypto)?;

            Ok::<_, Error>((session_id, ctr, boundary))
        })?;

        // Store the moved boundary BEFORE the reserved value can reach the
        // wire: a restart then resumes past it, so receivers never see a
        // counter value replayed. Writes happen once per
        // `GROUP_DATA_CTR_EPOCH` messages, not per message.
        if let Some(boundary) = boundary {
            kv.access(|store, buf| {
                store.store(
                    crate::persist::GROUP_DATA_COUNTER_KEY,
                    &boundary.to_le_bytes(),
                    buf,
                )
            })?;

            debug!(
                "Group data message counter boundary persisted: {}",
                boundary
            );
        }

        let exchange = Self::initiate_for_session(matter, crypto, session_id)?;

        // Stash the reservation in the per-exchange state, which is where
        // `Session::pre_send` picks it up when stamping the message.
        matter.with_state(|state| {
            let session = state
                .sessions
                .get(exchange.id().session_id())
                .ok_or(ErrorCode::NoSession)?;

            let exch = session.exchanges[exchange.id().exchange_index()]
                .as_mut()
                .ok_or(ErrorCode::NoExchange)?;

            exch.group_data_ctr = Some(group_data_ctr);

            Ok::<_, Error>(())
        })?;

        Ok(exchange)
    }

    /// Create a new initiator exchange on a new plaintext session to
    /// the given peer address.
    ///
    /// Low-level primitive below the three high-level entry points
    /// ([`Exchange::initiate`], [`Exchange::initiate_pase`],
    /// [`Exchange::initiate_for_session`]): the returned exchange carries the
    /// first handshake message of PASE ([`crate::sc::pase::PaseInitiator`]) or
    /// CASE ([`crate::sc::case::CaseInitiator`]). Use this only when driving a handshake protocol
    /// directly; otherwise prefer `initiate_pase` (which runs PASE for you) or
    /// `initiate` (CASE). If there is no space for a new session, an existing
    /// session is evicted and the operation retried.
    #[inline(always)]
    pub async fn initiate_plaintext<C: Crypto>(
        matter: &'a Matter<'a>,
        crypto: C,
        peer_addr: network::Address,
    ) -> Result<Self, Error> {
        matter
            .transport
            .initiate_plaintext(matter, crypto, peer_addr)
            .await
    }

    /// Open an exchange over a fresh plaintext session to an already-
    /// commissioned node, resolving its operational address over mDNS.
    ///
    /// Like [`initiate_plaintext`](Self::initiate_plaintext) but it discovers the
    /// peer address itself (as [`initiate`](Self::initiate) does for CASE), rather
    /// than taking a known one. For sessionless protocols that carry their own
    /// security and must not establish a CASE session — e.g. sending an ICD
    /// Check-In message to a registered client.
    ///
    /// Requires a running mDNS responder to service the resolve; without one the
    /// resolve times out and this returns [`ErrorCode::NotFound`].
    #[inline(always)]
    pub async fn initiate_plaintext_operational<C: Crypto>(
        matter: &'a Matter<'a>,
        crypto: C,
        fabric_idx: NonZeroU8,
        peer_node_id: NodeId,
    ) -> Result<Self, Error> {
        matter
            .transport
            .initiate_plaintext_operational(matter, crypto, fabric_idx, peer_node_id)
            .await
    }

    /// Accepts a new responder exchange pending on the provided Matter stack.
    ///
    /// If there is no new pending responder exchange, the method will wait indefinitely until one appears.
    #[inline(always)]
    pub async fn accept(matter: &'a Matter<'a>) -> Result<Self, Error> {
        Self::accept_after(matter, 0).await
    }

    /// Accepts a new responder exchange pending on the provided Matter stack, but only if the
    /// pending exchange was pending for longer than `received_timeout_ms`.
    ///
    /// If there is no new pending responder exchange, the method will wait indefinitely until one appears.
    pub async fn accept_after(
        matter: &'a Matter<'a>,
        received_timeout_ms: u32,
    ) -> Result<Self, Error> {
        if received_timeout_ms > 0 {
            loop {
                let mut accept = pin!(matter.transport().accept_if(matter, |_, exch, _| {
                    exch.mrp.has_rx_timed_out(received_timeout_ms as _)
                }));

                let mut timer = pin!(Timer::after(embassy_time::Duration::from_millis(
                    received_timeout_ms as u64
                )));

                if let Either::First(exchange) = select(&mut accept, &mut timer).await {
                    break exchange;
                }
            }
        } else {
            matter.transport().accept_if(matter, |_, _, _| true).await
        }
    }

    /// Get access to the pending RX message on this exchange, and consume it when the returned `RxMessage` instance is dropped.
    ///
    /// If there is no pending RX message, the method will wait indefinitely until one appears.
    ///
    /// Note that if the uderlying session or exchange tracked by the Matter stack is dropped
    /// (say, because of lack of resources or a hard networking error), the method will return an error.
    #[inline(always)]
    pub async fn recv(&mut self) -> Result<RxMessage<'_>, Error> {
        self.recv_fetch().await?;

        self.rx.take().ok_or(ErrorCode::InvalidState.into())
    }

    /// Get access to the pending RX message on this exchange, and consume it
    /// by copying the payload into the provided `WriteBuf` instance.
    ///
    /// A syntax sugar for calling ```self.recv().await?``` and then copying the payload.
    ///
    /// Returns the exchange message meta-data.
    ///
    /// If there is no pending RX message, the method will wait indefinitely until one appears.
    ///
    /// If there is already a pending RX message, which was already fetched using `Exchange::recv_fetch` and that
    /// message is not cleared yet using `Exchange::rx_done` or via some of the `Exchange::send*` methods,
    /// the method will return that message.
    ///
    /// Note that if the uderlying session or exchange tracked by the Matter stack is dropped
    /// (say, because of lack of resources or a hard networking error), the method will return an error.
    #[inline(always)]
    pub async fn recv_into(&mut self, wb: &mut WriteBuf<'_>) -> Result<MessageMeta, Error> {
        let rx = self.recv().await?;

        wb.reset();
        wb.append(rx.payload())?;

        Ok(rx.meta())
    }

    /// Return a _reference_ to the pending RX message on this exchange.
    ///
    /// If there is no pending RX message, the method will wait indefinitely until one appears.
    ///
    /// Unlike `recv` which returns the actual message object which - when dropped - allows the transport to
    /// fetch the _next_ RX message for this or other exchanges, `recv_fetch` keeps the received message around,
    /// which is convenient when the message needs to be examined / processed by multiple layers of application code.
    ///
    /// Note however that this does not come for free - keeping the RX message around means that the transport cannot receive
    /// _other_ RX messages which blocks the whole transport layer, as the transport layer uses a single RX message buffer.
    ///
    /// Therefore, calling `recv_fetch` should be done with care and the message should be marked as processed (and thus dropped) -
    /// via `rx_done` as soon as possible, ideally without `await`-ing between `recv_fetch` and `rx_done`
    ///
    /// Note that if the uderlying session or exchange tracked by the Matter stack is dropped
    /// (say, because of lack of resources or a hard networking error), the method will return an error.
    #[inline(always)]
    pub async fn recv_fetch(&mut self) -> Result<&RxMessage<'a>, Error> {
        if self.rx.is_none() {
            let rx = self.id.recv(self.matter).await?;

            self.rx = Some(rx);
        }

        self.rx()
    }

    /// Returns the RX message which was already fetched using a previous call to `recv_fetch`.
    /// If there is no fetched RX message, the method will fail with `ErrorCode::InvalidState`.
    ///
    /// This method only exists as a slight optimization for the cases where the user is sure, that there is
    /// an RX message already fetched with `recv_fetch`, as - unlike `recv_fetch` - this method does not `await` and hence
    /// variables used after calling `rx` do not have to be stored in the generated future.
    ///
    /// But in general and putting optimizations aside, it is always safe to replace calls to `rx` with calls to `recv_fetch`.
    #[inline(always)]
    pub fn rx(&self) -> Result<&RxMessage<'a>, Error> {
        self.rx.as_ref().ok_or(ErrorCode::InvalidState.into())
    }

    /// Clears the RX message which was already fetched using a previous call to `recv_fetch`.
    /// If there is no fetched RX message, the method will do nothing.
    #[inline(always)]
    pub fn rx_done(&mut self) -> Result<(), Error> {
        self.rx = None;

        Ok(())
    }

    /// Gets access to the TX buffer of the Matter stack for constructing a new TX message.
    /// If the TX buffer is not available, the method will wait indefinitely until it becomes available.
    ///
    /// NOTE:
    /// This is a low-level method that leaves the re-transmission logic on the shoulders of the user.
    /// Therefore, prefer using `Exchange::sender`, `Exchange::send` or `Exchange::send_with` instead.
    ///
    /// Note also that if the uderlying session or exchange tracked by the Matter stack is dropped
    /// (say, because of lack of resources or a hard networking error), the method will return an error.
    #[inline(always)]
    pub async fn init_send(&mut self) -> Result<TxMessage<'_>, Error> {
        self.rx = None;

        self.id.init_send(self.matter).await
    }

    /// Waits until the other side acknowledges the last message sent on this exchange,
    /// or until time for a re-transmission had come.
    ///
    /// If the last sent message was not using the MRP protocol, the method will return immediately with `TxOutcome::Done`.
    ///
    /// NOTE:
    /// This is a low-level method that leaves the re-transmission logic on the shoulders of the user.
    /// Therefore, prefer using `Exchange::sender`, `Exchange::send` or `Exchange::send_with` instead.
    ///
    /// Note also that if the uderlying session or exchange tracked by the Matter stack is dropped
    /// (say, because of lack of resources or a hard networking error), the method will return an error.
    #[inline(always)]
    pub async fn wait_tx(&mut self) -> Result<TxOutcome, Error> {
        self.rx = None;

        self.id.wait_tx(self.matter).await
    }

    /// Returns `true` if there is a pending message re-transmission.
    /// A re-transmission will be pending if the last sent message was using the MRP protocol, and
    /// an acknowledgement for the other side is still pending.
    ///
    /// NOTE:
    /// This is a low-level method that leaves the re-transmission logic on the shoulders of the user.
    /// Therefore, prefer using `Exchange::sender`, `Exchange::send` or `Exchange::send_with` instead.
    ///
    /// Note also that if the uderlying session or exchange tracked by the Matter stack is dropped
    /// (say, because of lack of resources or a hard networking error), the method will return an error.
    pub fn pending_retrans(&self) -> Result<bool, Error> {
        self.id.pending_retrans(self.matter)
    }

    /// Returns `true` if there is a pending message acknowledgement.
    /// An acknowledgement be pending if the last received message was using the MRP protocol, and we have to acknowledge it.
    ///
    /// NOTE:
    /// This is a low-level method that leaves the re-transmission logic on the shoulders of the user.
    /// Therefore, prefer using `Exchange::sender`, `Exchange::send` or `Exchange::send_with` instead.
    ///
    /// Note also that if the uderlying session or exchange tracked by the Matter stack is dropped
    /// (say, because of lack of resources or a hard networking error), the method will return an error.
    pub fn pending_ack(&self) -> Result<bool, Error> {
        self.id.pending_ack(self.matter)
    }

    /// Acknowledge the last message received on this exchange (by sending a `MrpStandaloneAck`).
    ///
    /// If the last message was already acknowledged
    /// (either by a previous call to this method, by piggy-backing on a reliable message, or by the Matter stack itself),
    /// this method does nothing.
    ///
    /// Note that if the uderlying session or exchange tracked by the Matter stack is dropped
    /// (say, because of lack of resources or a hard networking error), the method will return an error.
    #[inline(always)]
    pub async fn acknowledge(&mut self) -> Result<(), Error> {
        if self.pending_ack()? {
            let tx = self.id.init_send(self.matter).await?;

            if self.pending_ack()? {
                // Check whether we still need to send an ACK.
                // Necessary because we `.await` above, and while we are awaiting, the transport
                // might automatically send an ACK for us.
                // (That is, if the global RX transport buffer happens to be already empty and if the other peer re-sends the message.)
                tx.complete::<MessageMeta>(0, 0, sc::OpCode::MRPStandAloneAck.into())?;
            }
        }

        Ok(())
    }

    /// Utility for sending a message on this exchange that automatically handles all re-transmission logic
    /// in case the constructed message needs to be send reliably.
    ///
    /// Note that if the uderlying session or exchange tracked by the Matter stack is dropped
    /// (say, because of lack of resources or a hard networking error), the method will return an error.
    pub fn sender(&mut self) -> Result<Sender<'_>, Error> {
        self.rx = None;

        Sender::new(self)
    }

    /// Owned-`Self` counterpart to [`Exchange::sender`].
    ///
    /// Consumes the exchange and returns an [`OwnedSender`] that
    /// owns it for the lifetime of the retransmit loop. Once the
    /// loop completes ([`OwnedSender::tx`] returns
    /// `Either::Right(exchange)`), the exchange is handed back to
    /// the caller.
    ///
    /// Used by consumers — such as the IM client — that take an
    /// `Exchange` by value and need to drive a send + receive cycle
    /// in sequence without giving up ownership.
    pub fn into_sender(mut self) -> Result<OwnedSender<'a>, Error> {
        self.rx = None;

        OwnedSender::new(self)
    }

    /// Utility for sending a message on this exchange that automatically handles all re-transmission logic
    /// in case the constructed message needs to be send reliably.
    ///
    /// The message is constructed by the provided closure, which is given a `WriteBuf` instance to write the message payload into.
    ///
    /// Note that the closure is expected to construct the exact same message when called multiple times.
    ///
    /// Note also that if the uderlying session or exchange tracked by the Matter stack is dropped
    /// (say, because of lack of resources or a hard networking error), the method will return an error.
    pub async fn send_with<F>(&mut self, mut f: F) -> Result<(), Error>
    where
        F: FnMut(&Exchange, &mut WriteBuf) -> Result<Option<MessageMeta>, Error>,
    {
        let mut sender = self.sender()?;

        while let Some(mut tx) = sender.tx().await? {
            let (exchange, payload) = tx.split();

            let mut wb = WriteBuf::new(payload);

            if let Some(meta) = f(exchange, &mut wb)? {
                let payload_start = wb.get_start();
                let payload_end = wb.get_tail();
                tx.complete(payload_start, payload_end, meta)?;
            } else {
                // Closure aborted sending
                break;
            }
        }

        Ok(())
    }

    /// Send the provided exchange meta-data and payload as part of this exchange.
    ///
    /// If the provided exchange meta-data indicates a reliable message, the message will be automatically re-transmitted until
    /// the other side acknowledges it.
    ///
    /// Note that if the uderlying session or exchange tracked by the Matter stack is dropped
    /// (say, because of lack of resources or a hard networking error), the method will return an error.
    pub async fn send<M>(&mut self, meta: M, payload: &[u8]) -> Result<(), Error>
    where
        M: Into<MessageMeta>,
    {
        let meta = meta.into();

        self.send_with(|_, wb| {
            wb.append(payload)?;

            Ok(Some(meta))
        })
        .await
    }

    pub(crate) fn accessor<M>(&self, metadata: M) -> Result<Accessor<'a>, Error>
    where
        M: Metadata,
    {
        self.id.accessor(self.matter, metadata.aux_acl_enabled())
    }

    pub fn is_groupcast(&self) -> Result<bool, Error> {
        self.with_state(|state| {
            Ok(matches!(
                self.id().session(&mut state.sessions).get_session_mode(),
                SessionMode::Group { .. }
            ))
        })
    }

    pub(crate) fn with_state<F, T>(&self, f: F) -> Result<T, Error>
    where
        F: FnOnce(&mut MatterState) -> Result<T, Error>,
    {
        self.id.with_state(self.matter, f)
    }

    pub(crate) fn with_state_ex<F, T, E>(&self, f: F) -> Result<T, E>
    where
        F: FnOnce(&mut MatterState) -> Result<T, E>,
        E: From<Error>,
    {
        self.id.with_state_ex(self.matter, f)
    }
}

impl Drop for Exchange<'_> {
    fn drop(&mut self) {
        let closed = self.with_state(|state| {
            let sess = self.id().session(&mut state.sessions);
            let exch_index = self.id.exchange_index();

            let closed = sess.remove_exch(exch_index);
            if closed {
                // RX group sessions (unicast peer = the sender's address) are
                // ephemeral, one per received message — remove them with their
                // last exchange. TX group sessions (multicast peer) stay: the
                // just-queued outgoing packet still needs the session for
                // encoding, and subsequent sends to the same group reuse it
                // (`get_or_create_for_group_tx`); LRU eviction reclaims them.
                if matches!(sess.get_session_mode(), SessionMode::Group { .. })
                    && sess.exchanges.iter().all(Option::is_none)
                    && !sess.is_peer_multicast()
                {
                    // Group session with no remaining exchanges — remove it
                    state.sessions.remove(self.id.session_id());
                    self.matter.transport().notify_session_removed();
                }

                Ok(true)
            } else {
                Ok(false)
            }
        });

        if !matches!(closed, Ok(true)) {
            self.matter.transport().exchange_dropped.notify();
        }
    }
}

impl Display for Exchange<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.id)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::crypto::test_only_crypto;
    use crate::dm::devices::test::{TEST_DEV_ATT, TEST_DEV_COMM, TEST_DEV_DET};
    use crate::error::ErrorCode;
    use crate::transport::session::SessionMode;
    use crate::Matter;
    use futures_lite::future::block_on;

    fn test_matter() -> Matter<'static> {
        Matter::new(&TEST_DEV_DET, TEST_DEV_COMM, &TEST_DEV_ATT, 0)
    }

    fn fill_sessions(matter: &Matter<'_>, reserved: bool) {
        let dev_det = matter.dev_det();
        matter.with_state(|state| loop {
            if state
                .sessions
                .add(0, reserved, network::Address::new(), None, dev_det)
                .is_err()
            {
                break;
            }
        });
    }

    #[test]
    fn test_initiate_plaintext_creates_initiator_exchange() {
        let matter = test_matter();
        let crypto = test_only_crypto();
        let peer = network::Address::new();

        let exchange = block_on(Exchange::initiate_plaintext(&matter, &crypto, peer)).unwrap();

        exchange
            .with_state(|state| {
                let sess = exchange.id().session(&mut state.sessions);
                let exch = exchange.id().exch(sess);

                assert!(matches!(exch.role, Role::Initiator(_)));
                assert_eq!(sess.id, exchange.id().session_id());
                assert!(!sess.is_encrypted());
                assert_eq!(*sess.get_session_mode(), SessionMode::PlainText);

                Ok(())
            })
            .unwrap();
    }

    #[test]
    fn test_initiate_plaintext_retries_after_eviction() {
        let matter = test_matter();
        let crypto = test_only_crypto();
        let peer = network::Address::new();

        fill_sessions(&matter, false);

        let exchange = block_on(Exchange::initiate_plaintext(&matter, &crypto, peer)).unwrap();

        exchange
            .with_state(|state| {
                let sess = exchange.id().session(&mut state.sessions);
                let exch = exchange.id().exch(sess);

                assert!(matches!(exch.role, Role::Initiator(_)));
                assert_eq!(sess.id, exchange.id().session_id());
                assert!(!sess.is_encrypted());
                assert_eq!(*sess.get_session_mode(), SessionMode::PlainText);

                Ok(())
            })
            .unwrap();
    }

    #[test]
    fn test_initiate_plaintext_fails_when_no_session_can_be_evicted() {
        let matter = test_matter();
        let crypto = test_only_crypto();
        let peer = network::Address::new();

        fill_sessions(&matter, true);

        let result = block_on(Exchange::initiate_plaintext(&matter, &crypto, peer));

        match result {
            Err(err) => assert!(matches!(err.code(), ErrorCode::NoSpaceSessions)),
            Ok(_) => panic!("expected NoSpaceSessions error"),
        }
    }
}