liminal-server 0.25.4

//! Keyed links have separate delivery queues; transport ACKs remain transport facts.
use super::{Arrival, Datagram, LinkCancellation, Links, UdpCarrier, UdpError, send_datagram};
use std::sync::{Arc, mpsc};
use std::time::Instant;

/// One keyed link's authenticated traffic, independent of other handshakes.
#[derive(Debug)]
pub struct Inbox(mpsc::Receiver<Arrival>, LinkCancellation);
impl Inbox {
    pub(crate) fn cancellation(&self) -> LinkCancellation {
        self.1.clone()
    }
    pub(crate) fn refuse_unsolicited(&self) -> Result<(), UdpError> {
        match self.0.try_recv() {
            Err(mpsc::TryRecvError::Empty) => Ok(()),
            Err(mpsc::TryRecvError::Disconnected) => Err(UdpError::Stopped),
            Ok(_) => Err(UdpError::InvalidDatagram),
        }
    }
    /// Parks on this link's next data frame with no idle timer.
    /// # Errors
    /// The carrier stopped or authentication refused a frame.
    pub fn receive(&self) -> Result<Datagram, UdpError> {
        self.0
            .recv()
            .map_err(|_| UdpError::Stopped)?
            .map(|(datagram, _)| datagram)
    }
    /// Observes one in-flight answer through its already armed deadline.
    /// # Errors
    /// The carrier stopped, refused a frame, or the message deadline passed.
    pub fn until(&self, deadline: Instant) -> Result<Datagram, UdpError> {
        self.0
            .recv_timeout(deadline.saturating_duration_since(Instant::now()))
            .map_err(|error| match error {
                mpsc::RecvTimeoutError::Timeout => UdpError::Deadline,
                mpsc::RecvTimeoutError::Disconnected => UdpError::Stopped,
            })?
            .map(|(datagram, _)| datagram)
    }
}
impl UdpCarrier {
    /// Binds the only application reader of this keyed index.
    /// # Errors
    /// Unknown or already watched link, stopped carrier, or poisoned state.
    pub fn subscribe(&self, index: u32) -> Result<Inbox, UdpError> {
        let mut incoming = self
            .incoming
            .as_ref()
            .ok_or(UdpError::Stopped)?
            .lock()
            .map_err(|_| UdpError::Poisoned)?;
        let links = self.links.lock().map_err(|_| UdpError::Poisoned)?;
        let link = links.get(&index).ok_or(UdpError::UnknownIndex)?;
        let mut route = link
            .acknowledgements
            .route
            .lock()
            .map_err(|_| UdpError::Poisoned)?;
        if link.acknowledgements.is_cancelled() {
            return Err(UdpError::Stopped);
        }
        if route.is_some() {
            return Err(UdpError::AcknowledgementsAlreadyWatched);
        }
        let (sender, receiver) = mpsc::sync_channel(64);
        *route = Some(sender.clone());
        let cancellation = LinkCancellation::new(Arc::clone(&link.acknowledgements));
        let mut parked = std::mem::take(&mut incoming.parked);
        for _ in 0..64 {
            match incoming.receiver.try_recv() {
                Ok(arrival) => parked.push_back(arrival),
                Err(mpsc::TryRecvError::Empty | mpsc::TryRecvError::Disconnected) => break,
            }
        }
        while let Some(arrival) = parked.pop_front() {
            match arrival {
                Ok((datagram, endpoint)) if datagram.receiver_index == index => {
                    if datagram.kind == 3 {
                        sender
                            .try_send(Ok((datagram, endpoint)))
                            .map_err(|_| UdpError::ObservationInProgress)?;
                    }
                }
                other => incoming.parked.push_back(other),
            }
        }
        drop(route);
        drop(links);
        drop(incoming);
        Ok(Inbox(receiver, cancellation))
    }
    /// Seals and retains an importing reply only after its application completed.
    /// # Errors
    /// Unknown request, oversize response, poisoned state, or send failure.
    pub fn reply(&self, index: u32, request: u64, plain: &[u8]) -> Result<(), UdpError> {
        let datagram = self.seal(index, plain)?;
        let mut links = self.links.lock().map_err(|_| UdpError::Poisoned)?;
        let link = links.get_mut(&index).ok_or(UdpError::UnknownIndex)?;
        let cached = link
            .cached_acks
            .iter_mut()
            .find(|ack| ack.counter == request)
            .ok_or(UdpError::InvalidAck)?;
        cached.receipt = Some(datagram.clone());
        let endpoint = link.endpoint;
        drop(links);
        send_datagram(self.socket.as_ref(), endpoint, &datagram)
    }
}
pub(super) fn route_arrival(
    links: &Links,
    index: Option<u32>,
    arrival: Arrival,
    outgoing: &mpsc::SyncSender<Arrival>,
) -> Result<bool, UdpError> {
    // Registration and queue transfer hold the same lock as this dispatch decision.
    let links = links.lock().map_err(|_| UdpError::Poisoned)?;
    let route = index
        .and_then(|index| links.get(&index))
        .map(|link| {
            link.acknowledgements
                .route
                .lock()
                .map_err(|_| UdpError::Poisoned)
        })
        .transpose()?;
    if let Some(sender) = route.as_ref().and_then(|route| route.as_ref()) {
        match arrival {
            Ok((datagram, endpoint)) if datagram.kind == 3 => {
                match sender.try_send(Ok((datagram, endpoint))) {
                    Ok(()) => {}
                    Err(mpsc::TrySendError::Full(_)) => tracing::warn!(
                        "udp_application_queue_full; crossing must recover at its deadline"
                    ),
                    Err(mpsc::TrySendError::Disconnected(_)) => {
                        tracing::debug!("UDP application reader ended");
                    }
                }
            }
            Ok(_) => {}
            Err(error) => tracing::warn!(%error,"UDP application datagram refused"),
        }
        drop(route);
        drop(links);
        return Ok(true);
    }
    drop(route);
    let result = match outgoing.try_send(arrival) {
        Ok(()) => true,
        Err(mpsc::TrySendError::Full(_)) => {
            tracing::warn!("udp_open_queue_full; opening must refuse at its deadline");
            true
        }
        Err(mpsc::TrySendError::Disconnected(_)) => false,
    };
    drop(links);
    Ok(result)
}