prns-runtime-embassy 0.3.6

use embassy_sync::blocking_mutex::raw::RawMutex;
use embassy_sync::channel::Sender;
use embassy_sync::signal::Signal;

use crate::interfaces::{InterfaceDescriptor, InterfaceId};
use crate::manifold::driver::{EmbassyGrantConsumer, EmbassyGrantProducer, InterfaceLifecycle};
use crate::manifold::grant::{FrameTarget, GrantConsumer, GrantProducer};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InboundDeliveryError {
    FrameTooLarge { len: usize, capacity: usize },
    LaneFull,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OutboundFrame<const FRAME: usize> {
    target: FrameTarget,
    bytes: [u8; FRAME],
    len: usize,
}

impl<const FRAME: usize> OutboundFrame<FRAME> {
    #[must_use]
    pub fn target(&self) -> FrameTarget {
        self.target
    }

    #[must_use]
    pub fn bytes(&self) -> &[u8] {
        &self.bytes[..self.len]
    }

    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.len == 0
    }

    #[must_use]
    pub fn len(&self) -> usize {
        self.len
    }
}

pub(super) struct FleetWire<M: RawMutex + 'static, const FRAME: usize, const NOTIFY: usize> {
    pub(super) inbound: EmbassyGrantProducer<'static, M, FRAME>,
    pub(super) outbound: EmbassyGrantConsumer<'static, M, FRAME>,
    pub(super) notify: Sender<'static, M, InterfaceId, NOTIFY>,
    pub(super) outbound_wake: &'static Signal<M, ()>,
}

pub struct Fleet<
    M: RawMutex + 'static,
    const FRAME: usize,
    const NOTIFY: usize,
    const LIFECYCLE: usize,
> {
    wire: FleetWire<M, FRAME, NOTIFY>,
    lifecycle: Sender<'static, M, InterfaceLifecycle, LIFECYCLE>,
}

impl<M: RawMutex + 'static, const FRAME: usize, const NOTIFY: usize, const LIFECYCLE: usize>
    Fleet<M, FRAME, NOTIFY, LIFECYCLE>
{
    #[must_use]
    pub(super) fn new(
        wire: FleetWire<M, FRAME, NOTIFY>,
        lifecycle: Sender<'static, M, InterfaceLifecycle, LIFECYCLE>,
    ) -> Self {
        Self { wire, lifecycle }
    }

    pub async fn register_member(&self, descriptor: InterfaceDescriptor) {
        self.lifecycle
            .send(InterfaceLifecycle::Add { descriptor })
            .await;
    }

    pub async fn deregister_member(&self, id: InterfaceId) {
        self.lifecycle.send(InterfaceLifecycle::Remove { id }).await;
    }

    pub fn try_deliver_inbound(
        &mut self,
        child: InterfaceId,
        bytes: &[u8],
    ) -> Result<(), InboundDeliveryError> {
        if bytes.len() > FRAME {
            return Err(InboundDeliveryError::FrameTooLarge {
                len: bytes.len(),
                capacity: FRAME,
            });
        }
        let Some(grant) = self.wire.inbound.try_grant() else {
            self.wire.inbound.note_pressure();
            return Err(InboundDeliveryError::LaneFull);
        };
        grant.fill_for(child, bytes);
        self.wire.inbound.commit();
        let _ = self.wire.notify.try_send(child);
        Ok(())
    }

    /// Delivers one member frame with end-to-end backpressure.
    ///
    /// Supervisors backed by a reliable transport should use this path: when the manifold is
    /// still consuming the previous frame, the transport can stop admitting further frames
    /// instead of silently turning transient scheduler latency into packet loss.
    pub async fn deliver_inbound(
        &mut self,
        child: InterfaceId,
        bytes: &[u8],
    ) -> Result<(), InboundDeliveryError> {
        if bytes.len() > FRAME {
            return Err(InboundDeliveryError::FrameTooLarge {
                len: bytes.len(),
                capacity: FRAME,
            });
        }
        let grant = self.wire.inbound.grant().await;
        grant.fill_for(child, bytes);
        self.wire.inbound.commit();
        self.wire.notify.send(child).await;
        Ok(())
    }

    pub async fn next_outbound(&mut self) -> OutboundFrame<FRAME> {
        self.wire.outbound.release();
        let slot = self.wire.outbound.peek().await;
        let target = slot.target;
        let len = slot.len;
        let mut bytes = [0; FRAME];
        bytes[..len].copy_from_slice(slot.frame());
        self.wire.outbound.release();
        OutboundFrame { target, bytes, len }
    }

    /// Waits for a shared-lane commit; drain with [`try_next_outbound`](Self::try_next_outbound) after waking.
    pub async fn outbound_ready(&self) {
        self.wire.outbound_wake.wait().await;
    }

    pub fn try_next_outbound(&mut self) -> Option<OutboundFrame<FRAME>> {
        let slot = self.wire.outbound.try_peek()?;
        let target = slot.target;
        let len = slot.len;
        let mut bytes = [0; FRAME];
        bytes[..len].copy_from_slice(slot.frame());
        self.wire.outbound.release();
        Some(OutboundFrame { target, bytes, len })
    }
}

#[cfg(test)]
mod tests;