liminal-server 0.25.4

//! An owned close signal preserves the existing bridge fault handler's socket contract.
use crate::link_open::{OpenError, Stop};
use std::io::Read;
use std::net::{Shutdown, TcpListener, TcpStream};
use std::thread::JoinHandle;
pub struct Interrupt {
    socket: TcpStream,
    stop: Option<Stop>,
    worker: Option<JoinHandle<Result<(), OpenError>>>,
}
impl std::fmt::Debug for Interrupt {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Interrupt").finish_non_exhaustive()
    }
}
impl Interrupt {
    pub(super) fn wire(socket: TcpStream) -> Self {
        Self {
            socket,
            stop: None,
            worker: None,
        }
    }
    pub(super) fn chosen(stop: Stop) -> Result<Self, OpenError> {
        let listener = TcpListener::bind("127.0.0.1:0").map_err(named)?;
        let socket = TcpStream::connect(listener.local_addr().map_err(named)?).map_err(named)?;
        let (mut receiving, _) = listener.accept().map_err(named)?;
        let stop_worker = stop.clone();
        let worker = std::thread::Builder::new()
            .name("link-carrier-stop".to_owned())
            .spawn(move || {
                let mut signal = [0; 1];
                match receiving.read(&mut signal) {
                    Ok(0) => stop_worker(),
                    Ok(_) => Err(OpenError("unexpected_carrier_stop_bytes".to_owned())),
                    Err(error) => {
                        let stopped = stop_worker();
                        stopped.and_then(|()| Err(named(error)))
                    }
                }
            })
            .map_err(named)?;
        Ok(Self {
            socket,
            stop: Some(stop),
            worker: Some(worker),
        })
    }
    pub(crate) fn try_clone(&self) -> std::io::Result<TcpStream> {
        self.socket.try_clone()
    }
    pub(crate) fn shutdown(&self, direction: Shutdown) -> std::io::Result<()> {
        let stopped = self
            .stop
            .as_ref()
            .map_or(Ok(()), |stop| stop().map_err(std::io::Error::other));
        let closed = self.socket.shutdown(direction);
        stopped.and(closed)
    }
}
fn named(error: impl std::fmt::Display) -> OpenError {
    OpenError(error.to_string())
}
impl Drop for Interrupt {
    fn drop(&mut self) {
        if let Err(error) = self.socket.shutdown(Shutdown::Both)
            && error.kind() != std::io::ErrorKind::NotConnected
        {
            tracing::error!(%error,"carrier stop signal failed");
        }
        if let Some(worker) = self.worker.take() {
            match worker.join() {
                Ok(Ok(())) => {}
                Ok(Err(error)) => tracing::error!(%error,"carrier cancellation failed"),
                Err(_) => tracing::error!("carrier cancellation worker panicked"),
            }
        }
    }
}