korri-n2k 0.7.1

//! Address supervisor built on top of [`AddressManager`] for Tokio runtime.
//!
//! It keeps the claiming state-machine alive and optionally offers:
//!
//! * a transmission handle (`AddressHandle`) to queue frames/PGNs;
//! * a frame receiver (`AddressFrames`) to pull incoming traffic, unfiltered.
//!
//! This implementation uses `tokio::sync::mpsc` channels.

use core::fmt::Debug;
use core::future::pending;
use futures_util::{future::select, future::Either, pin_mut};
use std::sync::Arc;
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::mpsc::{channel, Receiver, Sender};

use crate::error::{ClaimFault, SendPgnError};
use crate::infra::codec::traits::PgnData;
use crate::protocol::management::address_claiming::engine::ClaimStatus;
use crate::protocol::management::address_supervisor::{
    handle_command, AddressHandleError, AddressSupervisorRunError, ClaimedAddress,
    SupervisorCommand,
};
use crate::protocol::management::{address_manager::AddressManager, iso_name::IsoName};
use crate::protocol::transport::can_frame::CanFrame;
use crate::protocol::transport::fast_packet::MAX_FAST_PACKET_PAYLOAD;
use crate::protocol::transport::traits::can_bus::CanBus;
use crate::protocol::transport::traits::korri_timer::KorriTimer;

/// Service assembling the supervisor components.
pub struct AddressService<'a, C: CanBus, T: KorriTimer>
where
    C::Error: Debug,
{
    manager: AddressManager<'a, C, T>,
    command_rx: Option<Receiver<SupervisorCommand>>,
    frame_tx: Option<Sender<CanFrame>>,
    handle: Option<AddressHandle>,
    frames: Option<AddressFrames>,
    claimed: Arc<ClaimedAddress>,
}

impl<'a, C, T> AddressService<'a, C, T>
where
    C: CanBus,
    C::Error: Debug,
    T: KorriTimer,
{
    /// Wrap an already-initialised [`AddressManager`].
    pub fn new(manager: AddressManager<'a, C, T>, cmd_cap: usize, frame_cap: usize) -> Self {
        let (cmd_tx, cmd_rx) = if cmd_cap > 0 {
            let (tx, rx) = channel(cmd_cap);
            (Some(tx), Some(rx))
        } else {
            (None, None)
        };

        let (frame_tx, frame_rx) = if frame_cap > 0 {
            let (tx, rx) = channel(frame_cap);
            (Some(tx), Some(rx))
        } else {
            (None, None)
        };

        let claimed = Arc::new(ClaimedAddress::new());

        Self {
            manager,
            command_rx: cmd_rx,
            frame_tx,
            handle: cmd_tx.map(|tx| AddressHandle {
                sender: tx,
                claimed: Arc::clone(&claimed),
            }),
            frames: frame_rx.map(|rx| AddressFrames { receiver: rx }),
            claimed,
        }
    }

    /// Convenience helper: build the manager and the service in one call.
    /// The claim itself happens under [`AddressRunner::drive`].
    pub fn with_name(
        can_bus: C,
        timer: T,
        my_name: IsoName,
        strategy: crate::protocol::management::address_claiming::AddressClaimStrategy<'a>,
        cmd_cap: usize,
        frame_cap: usize,
    ) -> Result<Self, ClaimFault> {
        let manager = AddressManager::new(can_bus, timer, my_name, strategy)?;
        Ok(Self::new(manager, cmd_cap, frame_cap))
    }

    /// Split into handle/receiver/runner components.
    pub fn into_parts(self) -> AddressServiceParts<'a, C, T> {
        AddressServiceParts {
            handle: self.handle,
            frames: self.frames,
            runner: AddressRunner {
                manager: self.manager,
                command_rx: self.command_rx,
                frame_tx: self.frame_tx,
                claimed: self.claimed,
            },
        }
    }
}

/// Bundle returned by [`AddressService::into_parts`].
pub struct AddressServiceParts<'a, C, T>
where
    C: CanBus,
    C::Error: Debug,
    T: KorriTimer,
{
    pub handle: Option<AddressHandle>,
    pub frames: Option<AddressFrames>,
    pub runner: AddressRunner<'a, C, T>,
}

/// Runner that drives the supervisor loop.
pub struct AddressRunner<'a, C, T>
where
    C: CanBus,
    C::Error: Debug,
    T: KorriTimer,
{
    manager: AddressManager<'a, C, T>,
    command_rx: Option<Receiver<SupervisorCommand>>,
    frame_tx: Option<Sender<CanFrame>>,
    claimed: Arc<ClaimedAddress>,
}

impl<'a, C, T> AddressRunner<'a, C, T>
where
    C: CanBus,
    C::Error: Debug,
    T: KorriTimer,
{
    /// Own the single `select` of the supervisor: wait, poll the engine
    /// synchronously, execute, come back. No transition is ever cancelled.
    pub async fn drive(mut self) -> Result<(), AddressSupervisorRunError<C::Error>> {
        let mut rx: Option<CanFrame> = None;

        loop {
            let mut output = self.manager.poll(rx.as_ref());

            // Published before the emission below: a lost address must stop
            // producers now, not after an `await` on a possibly slow bus.
            self.claimed.set(Some(output.status));

            // The engine saw it first; the application gets it too, unfiltered.
            // This `take` is also the "rx = None once consumed" the engine expects.
            if let Some(frame) = rx.take() {
                self.forward_frame(frame);
            }

            // Always emitted, whatever the status: a `Claimed` returned together
            // with a defence frame still owes that frame to the bus.
            if let Some(frame) = output.tx {
                self.manager
                    .emit_claim(&frame)
                    .await
                    .map_err(AddressSupervisorRunError::Send)?;
                output = self.manager.tx_sent();
            }

            let mut command = None;
            let mut channel_closed = false;

            // Scoped: both futures borrow `self` and must die before
            // `command_rx` is cleared or a command is executed.
            {
                let command_rx = &mut self.command_rx;
                let command_future = async {
                    match command_rx {
                        Some(rx) => rx.recv().await,
                        None => pending::<Option<SupervisorCommand>>().await,
                    }
                };
                pin_mut!(command_future);

                // `None` waits on the bus and on commands alone. It never means
                // "done": a conflict must still be able to wake the engine.
                let recv = self.manager.recv_until(output.wake_at_ms);
                pin_mut!(recv);

                // `select` polls its first argument first: a frame ready at
                // the same instant as the deadline wins, as the engine wants.
                match select(recv, command_future).await {
                    Either::Left((frame, _)) => {
                        rx = frame.map_err(AddressSupervisorRunError::Receive)?;
                    }
                    Either::Right((received, _)) => match received {
                        Some(cmd) => command = Some(cmd),
                        None => channel_closed = true,
                    },
                }
            }

            // A closed channel returns `None` instantly and forever:
            // drop it so the next round falls back on `pending()`.
            if channel_closed {
                self.command_rx = None;
            }

            if let Some(cmd) = command {
                self.run_command(cmd).await?;
            }
        }
    }

    /// Run one command. Only a bus error stops the runner: a rejected command is
    /// the caller's mistake and must not take address management down with it.
    ///
    /// Any `Err` from `CanBus` is terminal, transient or not. The driver is
    /// responsible for recovering what it can: see [`CanBus`].
    async fn run_command(
        &mut self,
        command: SupervisorCommand,
    ) -> Result<(), AddressSupervisorRunError<C::Error>> {
        match handle_command(&mut self.manager, command).await {
            Ok(()) => Ok(()),
            Err(SendPgnError::Send(err)) => Err(AddressSupervisorRunError::Send(err)),
            Err(_rejected) => {
                #[cfg(feature = "defmt")]
                defmt::warn!("supervisor command rejected");
                // dropped, and only visible under `defmt`. A rejection
                // channel when a consumer needs programmatic feedback.
                Ok(())
            }
        }
    }

    /// Hand the frame to the application, or drop it.
    ///
    /// Never awaits: a full channel must not delay the action the engine just
    /// decided. Application traffic is sacrificial, address management is not.
    fn forward_frame(&mut self, frame: CanFrame) {
        let Some(tx) = self.frame_tx.as_ref() else {
            return;
        };
        match tx.try_send(frame) {
            Ok(()) => {}
            // dropped silently. A count belongs on `AddressFrames`,
            // where the consumer that cares about the gap actually lives.
            Err(TrySendError::Full(_)) => {}
            Err(TrySendError::Closed(_)) => self.frame_tx = None,
        }
    }
}

/// Clear the published address when the runner goes away.
///
/// The runner is the only thing defending that address. Once it is gone, nobody
/// answers a competing claim, so a handle still reporting `Some(142)` would
/// invite the application to emit from an address it no longer holds.
///
/// `Drop` rather than a cleanup at the end of `drive`: it also covers the task
/// being cancelled and the future being dropped, which no `?` can catch.
impl<'a, C, T> Drop for AddressRunner<'a, C, T>
where
    C: CanBus,
    C::Error: Debug,
    T: KorriTimer,
{
    fn drop(&mut self) {
        self.claimed.set(None);
    }
}

/// Transmission handle (optional).
pub struct AddressHandle {
    sender: Sender<SupervisorCommand>,
    claimed: Arc<ClaimedAddress>,
}

impl AddressHandle {
    /// The current claim state, or `None` before the runner starts or after it stops.
    ///
    /// Best effort: see [`ClaimedAddress`].
    pub fn claim_status(&self) -> Option<ClaimStatus> {
        self.claimed.status()
    }

    /// The address this handle emits from, or `None` while none is held.
    ///
    /// Best effort: see [`ClaimedAddress`].
    pub fn claimed_address(&self) -> Option<u8> {
        self.claimed.get()
    }

    /// Queue a command built by the caller. Same contract as
    /// [`AddressHandle::send_pgn`]: it confirms queueing, not emission.
    pub async fn send_command(&self, command: SupervisorCommand) -> Result<(), AddressHandleError> {
        self.sender
            .send(command)
            .await
            .map_err(|_| AddressHandleError::RunnerGone)
    }

    /// Queue a frame. Returns once the runner has taken it from the channel,
    /// not once it reached the bus: see [`AddressHandle::send_pgn`].
    pub async fn send_raw_frame(&self, frame: &CanFrame) -> Result<(), AddressHandleError> {
        let command = SupervisorCommand::SendRawFrame(*frame);
        self.sender
            .send(command)
            .await
            .map_err(|_| AddressHandleError::RunnerGone)
    }

    /// Queue a PGN for the runner to emit.
    ///
    /// **This confirms queueing, not emission.** The runner may still refuse the
    /// command: no address acquired, or a conflict between now and execution.
    /// A refusal is dropped, not returned here. Check
    /// [`AddressHandle::claimed_address`] before queueing if that matters.
    pub async fn send_pgn<P: PgnData>(
        &self,
        pgn_data: &P,
        pgn: u32,
        priority: u8,
        destination: Option<u8>,
    ) -> Result<(), AddressHandleError> {
        // Serialized straight into the command: a second buffer would cost
        // another 223 bytes of stack and a copy, for nothing.
        let mut payload = [0u8; MAX_FAST_PACKET_PAYLOAD];
        let len = pgn_data
            .to_payload(&mut payload)
            .map_err(|_| AddressHandleError::Serialization)?;

        let command = SupervisorCommand::SendPayload {
            pgn,
            priority,
            destination,
            len,
            payload,
        };

        self.sender
            .send(command)
            .await
            .map_err(|_| AddressHandleError::RunnerGone)
    }
}

/// Optional receiver returning incoming frames.
///
/// Nothing is filtered out. Address-claim traffic is included, so network
/// discovery can read it too. A frame is dropped rather than queued when this
/// channel is full: address management is never delayed by a slow consumer.
pub struct AddressFrames {
    receiver: Receiver<CanFrame>,
}

impl AddressFrames {
    pub async fn recv(&mut self) -> Option<CanFrame> {
        self.receiver.recv().await
    }
}