liminal_server/link_udp/
streams.rs

1//! Keyed links have separate delivery queues; transport ACKs remain transport facts.
2use super::{Arrival, Datagram, LinkCancellation, Links, UdpCarrier, UdpError, send_datagram};
3use std::sync::{Arc, mpsc};
4use std::time::Instant;
5
6/// One keyed link's authenticated traffic, independent of other handshakes.
7#[derive(Debug)]
8pub struct Inbox(mpsc::Receiver<Arrival>, LinkCancellation);
9impl Inbox {
10    pub(crate) fn cancellation(&self) -> LinkCancellation {
11        self.1.clone()
12    }
13    pub(crate) fn refuse_unsolicited(&self) -> Result<(), UdpError> {
14        match self.0.try_recv() {
15            Err(mpsc::TryRecvError::Empty) => Ok(()),
16            Err(mpsc::TryRecvError::Disconnected) => Err(UdpError::Stopped),
17            Ok(_) => Err(UdpError::InvalidDatagram),
18        }
19    }
20    /// Parks on this link's next data frame with no idle timer.
21    /// # Errors
22    /// The carrier stopped or authentication refused a frame.
23    pub fn receive(&self) -> Result<Datagram, UdpError> {
24        self.0
25            .recv()
26            .map_err(|_| UdpError::Stopped)?
27            .map(|(datagram, _)| datagram)
28    }
29    /// Observes one in-flight answer through its already armed deadline.
30    /// # Errors
31    /// The carrier stopped, refused a frame, or the message deadline passed.
32    pub fn until(&self, deadline: Instant) -> Result<Datagram, UdpError> {
33        self.0
34            .recv_timeout(deadline.saturating_duration_since(Instant::now()))
35            .map_err(|error| match error {
36                mpsc::RecvTimeoutError::Timeout => UdpError::Deadline,
37                mpsc::RecvTimeoutError::Disconnected => UdpError::Stopped,
38            })?
39            .map(|(datagram, _)| datagram)
40    }
41}
42impl UdpCarrier {
43    /// Binds the only application reader of this keyed index.
44    /// # Errors
45    /// Unknown or already watched link, stopped carrier, or poisoned state.
46    pub fn subscribe(&self, index: u32) -> Result<Inbox, UdpError> {
47        let mut incoming = self
48            .incoming
49            .as_ref()
50            .ok_or(UdpError::Stopped)?
51            .lock()
52            .map_err(|_| UdpError::Poisoned)?;
53        let links = self.links.lock().map_err(|_| UdpError::Poisoned)?;
54        let link = links.get(&index).ok_or(UdpError::UnknownIndex)?;
55        let mut route = link
56            .acknowledgements
57            .route
58            .lock()
59            .map_err(|_| UdpError::Poisoned)?;
60        if link.acknowledgements.is_cancelled() {
61            return Err(UdpError::Stopped);
62        }
63        if route.is_some() {
64            return Err(UdpError::AcknowledgementsAlreadyWatched);
65        }
66        let (sender, receiver) = mpsc::sync_channel(64);
67        *route = Some(sender.clone());
68        let cancellation = LinkCancellation::new(Arc::clone(&link.acknowledgements));
69        let mut parked = std::mem::take(&mut incoming.parked);
70        for _ in 0..64 {
71            match incoming.receiver.try_recv() {
72                Ok(arrival) => parked.push_back(arrival),
73                Err(mpsc::TryRecvError::Empty | mpsc::TryRecvError::Disconnected) => break,
74            }
75        }
76        while let Some(arrival) = parked.pop_front() {
77            match arrival {
78                Ok((datagram, endpoint)) if datagram.receiver_index == index => {
79                    if datagram.kind == 3 {
80                        sender
81                            .try_send(Ok((datagram, endpoint)))
82                            .map_err(|_| UdpError::ObservationInProgress)?;
83                    }
84                }
85                other => incoming.parked.push_back(other),
86            }
87        }
88        drop(route);
89        drop(links);
90        drop(incoming);
91        Ok(Inbox(receiver, cancellation))
92    }
93    /// Seals and retains an importing reply only after its application completed.
94    /// # Errors
95    /// Unknown request, oversize response, poisoned state, or send failure.
96    pub fn reply(&self, index: u32, request: u64, plain: &[u8]) -> Result<(), UdpError> {
97        let datagram = self.seal(index, plain)?;
98        let mut links = self.links.lock().map_err(|_| UdpError::Poisoned)?;
99        let link = links.get_mut(&index).ok_or(UdpError::UnknownIndex)?;
100        let cached = link
101            .cached_acks
102            .iter_mut()
103            .find(|ack| ack.counter == request)
104            .ok_or(UdpError::InvalidAck)?;
105        cached.receipt = Some(datagram.clone());
106        let endpoint = link.endpoint;
107        drop(links);
108        send_datagram(self.socket.as_ref(), endpoint, &datagram)
109    }
110}
111pub(super) fn route_arrival(
112    links: &Links,
113    index: Option<u32>,
114    arrival: Arrival,
115    outgoing: &mpsc::SyncSender<Arrival>,
116) -> Result<bool, UdpError> {
117    // Registration and queue transfer hold the same lock as this dispatch decision.
118    let links = links.lock().map_err(|_| UdpError::Poisoned)?;
119    let route = index
120        .and_then(|index| links.get(&index))
121        .map(|link| {
122            link.acknowledgements
123                .route
124                .lock()
125                .map_err(|_| UdpError::Poisoned)
126        })
127        .transpose()?;
128    if let Some(sender) = route.as_ref().and_then(|route| route.as_ref()) {
129        match arrival {
130            Ok((datagram, endpoint)) if datagram.kind == 3 => {
131                match sender.try_send(Ok((datagram, endpoint))) {
132                    Ok(()) => {}
133                    Err(mpsc::TrySendError::Full(_)) => tracing::warn!(
134                        "udp_application_queue_full; crossing must recover at its deadline"
135                    ),
136                    Err(mpsc::TrySendError::Disconnected(_)) => {
137                        tracing::debug!("UDP application reader ended");
138                    }
139                }
140            }
141            Ok(_) => {}
142            Err(error) => tracing::warn!(%error,"UDP application datagram refused"),
143        }
144        drop(route);
145        drop(links);
146        return Ok(true);
147    }
148    drop(route);
149    let result = match outgoing.try_send(arrival) {
150        Ok(()) => true,
151        Err(mpsc::TrySendError::Full(_)) => {
152            tracing::warn!("udp_open_queue_full; opening must refuse at its deadline");
153            true
154        }
155        Err(mpsc::TrySendError::Disconnected(_)) => false,
156    };
157    drop(links);
158    Ok(result)
159}
liminal_server/link_udp/streams.rs

liminal_server/link_udp/
streams.rs
