net-lattice-platform 0.9.1

use std::any::Any;
use std::pin::Pin;
use std::sync::{
    Arc, Mutex,
    atomic::{AtomicBool, Ordering},
};
use std::task::{Context, Poll};

use net_lattice_core::Result;

/// Tokio-backed half of a bounded event channel.
///
/// This is an implementation detail of the feature-gated async provider
/// contract. Backends use [`send`](Self::send) exactly as their synchronous
/// counterparts use `EventSender`: when the consumer falls behind, one
/// resynchronization event is scheduled instead of allowing unbounded memory
/// growth.
pub struct TokioEventSender<E> {
    sender: tokio::sync::mpsc::Sender<Result<E>>,
    resync_pending: Arc<AtomicBool>,
    terminal_error: Arc<Mutex<Option<net_lattice_core::Error>>>,
}

/// Tokio receiver that owns the native subscription which produces its events.
///
/// The subscription is dropped with the receiver, cancelling platform handles,
/// tasks, or reader threads before their callback state can be freed.
pub struct TokioEventReceiver<E> {
    receiver: tokio::sync::mpsc::Receiver<Result<E>>,
    subscription: Option<Box<dyn Any + Send>>,
    terminal_error: Arc<Mutex<Option<net_lattice_core::Error>>>,
}

impl<E> TokioEventReceiver<E> {
    /// Creates a bounded transport using the same capacity as `EventReceiver`.
    pub fn bounded() -> (TokioEventSender<E>, Self) {
        let (sender, receiver) = tokio::sync::mpsc::channel(EventReceiverCapacity::VALUE);
        let pending = Arc::new(AtomicBool::new(false));
        let terminal_error = Arc::new(Mutex::new(None));
        (
            TokioEventSender {
                sender,
                resync_pending: pending,
                terminal_error: Arc::clone(&terminal_error),
            },
            Self {
                receiver,
                subscription: None,
                terminal_error,
            },
        )
    }

    /// Attaches the resource which owns the OS subscription.
    pub fn with_subscription<S>(mut self, subscription: S) -> Self
    where
        S: Any + Send + 'static,
    {
        self.subscription = Some(Box::new(subscription));
        self
    }

    /// Polls the next event without tying the platform crate to a futures API.
    pub fn poll_recv(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Result<E>>> {
        match Pin::new(&mut self.receiver).poll_recv(cx) {
            Poll::Ready(None) => Poll::Ready(
                self.terminal_error
                    .lock()
                    .expect("Tokio event sender poisoned")
                    .take()
                    .map(Err),
            ),
            other => other,
        }
    }
}

impl<E> TokioEventSender<E> {
    /// Attempts to enqueue an event without ever blocking an OS callback.
    pub fn send(&self, event: E, resync: impl FnOnce() -> E) -> bool {
        if self.resync_pending.swap(false, Ordering::AcqRel) {
            match self.sender.try_send(Ok(resync())) {
                Ok(()) => {}
                Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
                    self.resync_pending.store(true, Ordering::Release);
                    return true;
                }
                Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => return false,
            }
        }
        match self.sender.try_send(Ok(event)) {
            Ok(()) => true,
            Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
                self.resync_pending.store(true, Ordering::Release);
                true
            }
            Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => false,
        }
    }

    /// Reports a terminal backend error if the consumer is still connected.
    pub fn send_error(&self, error: net_lattice_core::Error) -> bool {
        if self.sender.is_closed() {
            return false;
        }
        *self
            .terminal_error
            .lock()
            .expect("Tokio event sender poisoned") = Some(error);
        true
    }
}

/// Kept here instead of exposing a second public capacity constant.
struct EventReceiverCapacity;
impl EventReceiverCapacity {
    const VALUE: usize = 256;
}

#[cfg(test)]
mod tests {
    use super::*;

    fn poll<E>(receiver: &mut TokioEventReceiver<E>) -> Poll<Option<Result<E>>> {
        let waker = std::task::Waker::noop();
        let mut context = Context::from_waker(waker);
        Pin::new(receiver).poll_recv(&mut context)
    }

    #[test]
    fn overflow_delivers_resync_before_a_later_event() {
        let (sender, mut receiver) = TokioEventReceiver::bounded();
        for event in 0..EventReceiverCapacity::VALUE {
            assert!(sender.send(event, || 999));
        }
        assert!(sender.send(256, || 999));
        for event in 0..EventReceiverCapacity::VALUE {
            assert!(matches!(poll(&mut receiver), Poll::Ready(Some(Ok(value))) if value == event));
        }
        assert!(sender.send(257, || 999));
        assert!(matches!(poll(&mut receiver), Poll::Ready(Some(Ok(999)))));
    }

    #[test]
    fn terminal_error_survives_a_full_queue() {
        let (sender, mut receiver) = TokioEventReceiver::bounded();
        for event in 0..EventReceiverCapacity::VALUE {
            assert!(sender.send(event, || 999));
        }
        assert!(sender.send_error(net_lattice_core::Error::InvalidState));
        drop(sender);
        for _ in 0..EventReceiverCapacity::VALUE {
            assert!(matches!(poll(&mut receiver), Poll::Ready(Some(Ok(_)))));
        }
        assert!(matches!(
            poll(&mut receiver),
            Poll::Ready(Some(Err(net_lattice_core::Error::InvalidState)))
        ));
    }
}