liminal-server 0.25.4

//! One owned reader observes carrier ends even while no crossing is pending.

use super::{CarrierError, Frame, TcpStream, read_frame};
use crate::link::{LinkEnd, LinkLife};
use std::net::Shutdown;
use std::sync::mpsc::{Receiver, TryRecvError, sync_channel};
use std::thread::JoinHandle;

type Answer = Result<Option<Frame>, CarrierError>;

#[derive(Debug)]
pub(super) struct Reader {
    inbox: Option<Receiver<Answer>>,
    worker: Option<JoinHandle<()>>,
}

impl Reader {
    pub(super) fn start(
        socket: &TcpStream,
        stream: u32,
        life: LinkLife,
        failure: std::sync::Arc<std::sync::OnceLock<String>>,
    ) -> Result<Self, CarrierError> {
        let mut socket = socket.try_clone()?;
        let (sender, inbox) = sync_channel(1);
        let dispatcher = tracing::dispatcher::get_default(Clone::clone);
        let worker = std::thread::Builder::new()
            .name("liminal-wire-reader".to_owned())
            .spawn(move || {
                tracing::dispatcher::with_default(&dispatcher, || {
                    loop {
                        let answer = read_frame(&mut socket).and_then(|frame| {
                            if let Some(frame) = &frame {
                                validate(frame, stream)?;
                            }
                            Ok(frame)
                        });
                        if let Some(name) = end_name(&answer) {
                            failure.get_or_init(|| name);
                        }
                        let end = end_cause(&answer);
                        let terminal = end.is_some();
                        if let Some(reason) = end
                            && let Err(error) = life.end(reason)
                        {
                            tracing::error!(%error, "wire reader end publication failed");
                        }
                        if let Err(error) = sender.send(answer) {
                            tracing::debug!(%error, "wire reader owner departed");
                            break;
                        }
                        if let Err(error) = life.activity() {
                            tracing::error!(%error, "wire reader readiness publication failed");
                        }
                        if terminal {
                            break;
                        }
                    }
                });
            })?;
        Ok(Self {
            inbox: Some(inbox),
            worker: Some(worker),
        })
    }

    pub(super) fn receive(&self) -> Answer {
        self.inbox
            .as_ref()
            .ok_or_else(|| CarrierError::Lifecycle("wire reader inbox was retired".to_owned()))?
            .recv()
            .map_err(|error| {
                CarrierError::Lifecycle(format!("wire reader departed without an end: {error}"))
            })?
    }

    pub(super) fn try_receive(&self) -> Result<Option<Answer>, CarrierError> {
        let inbox = self
            .inbox
            .as_ref()
            .ok_or_else(|| CarrierError::Lifecycle("wire reader inbox was retired".to_owned()))?;
        match inbox.try_recv() {
            Ok(answer) => Ok(Some(answer)),
            Err(TryRecvError::Empty) => Ok(None),
            Err(TryRecvError::Disconnected) => Err(CarrierError::Lifecycle(
                "wire reader departed without an end".to_owned(),
            )),
        }
    }

    pub(super) fn close(self, socket: &TcpStream) {
        self.close_with(|| socket.shutdown(Shutdown::Both));
    }

    fn close_with(mut self, interrupt: impl FnOnce() -> std::io::Result<()>) {
        let interruption = interrupt();
        // Dropping the receiver also releases a reader blocked on its bounded inbox.
        drop(self.inbox.take());
        if let Some(worker) = self.worker.take()
            && worker.join().is_err()
        {
            tracing::error!("wire link reader panicked");
        }
        if let Err(error) = interruption
            && error.kind() != std::io::ErrorKind::NotConnected
        {
            tracing::error!(%error, "wire link reader interruption failed");
        }
    }
}

fn validate(frame: &Frame, expected: u32) -> Result<(), CarrierError> {
    match frame {
        Frame::PublishAck { stream_id, .. } | Frame::PublishError { stream_id, .. }
            if *stream_id == expected =>
        {
            Ok(())
        }
        Frame::LinkMessage {
            stream_id, subject, ..
        } if *stream_id == expected => {
            crate::auth_lys::SubjectBinding::new(
                "wire".to_owned(),
                subject.clone(),
                subject.clone(),
            )?;
            Ok(())
        }
        Frame::Disconnect { .. } | Frame::DisconnectReason { .. } => Ok(()),
        _ => Err(CarrierError::UnexpectedFrame),
    }
}

/// The end an answer carries, by name, for the carrier's record: the peer's
/// close reason, `wire_eof`, or the failure's own text.
fn end_name(answer: &Answer) -> Option<String> {
    match answer {
        Ok(Some(Frame::DisconnectReason { reason, .. })) => Some(reason.name().to_owned()),
        Ok(Some(Frame::Disconnect { .. })) => Some("peer_disconnect".to_owned()),
        Ok(None) => Some("wire_eof".to_owned()),
        Err(error) => Some(error.to_string()),
        Ok(Some(_)) => None,
    }
}

fn end_cause(answer: &Answer) -> Option<LinkEnd> {
    match answer {
        Ok(Some(Frame::DisconnectReason { .. })) => Some(LinkEnd::Revoked),
        Ok(None | Some(Frame::Disconnect { .. }))
        | Err(CarrierError::Io(_) | CarrierError::Ended) => Some(LinkEnd::PeerGone),
        Err(error) => Some(LinkEnd::Refused(error.to_string())),
        Ok(Some(_)) => None,
    }
}

#[cfg(test)]
#[path = "reader_close_tests.rs"]
mod close_tests;