liminal-server 0.24.9

//! Configured dials register admission sockets and cancel connection establishment at shutdown.
use super::{Network, OpenError, named};
use std::future::Future;
use std::net::TcpStream;
use std::sync::Arc;
use std::task::Poll;
use std::time::Duration;
impl Network {
    pub(crate) fn wire_socket(
        &self,
        address: &str,
    ) -> Result<(TcpStream, Arc<TcpStream>), OpenError> {
        self.require_running()?;
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .map_err(named)?;
        let bound = Duration::from_millis(self.choice.crossing_deadline_ms);
        let connected = runtime.block_on(async {
            let mut cancelled = std::pin::pin!(self.cancellation.notified());
            cancelled.as_mut().enable();
            self.require_running()?;
            let mut connection = std::pin::pin!(tokio::time::timeout(
                bound,
                tokio::net::TcpStream::connect(address)
            ));
            std::future::poll_fn(|context| {
                if cancelled.as_mut().poll(context).is_ready() {
                    return Poll::Ready(Err(OpenError("network_stopped".to_owned())));
                }
                match connection.as_mut().poll(context) {
                    Poll::Ready(Ok(Ok(socket))) => Poll::Ready(Ok(socket)),
                    Poll::Ready(Ok(Err(error))) => Poll::Ready(Err(named(error))),
                    Poll::Ready(Err(_)) => {
                        Poll::Ready(Err(OpenError("wire_connect_deadline".to_owned())))
                    }
                    Poll::Pending => Poll::Pending,
                }
            })
            .await
        });
        let connected = connected.and_then(|socket| socket.into_std().map_err(named));
        // A cancelled resolver must not hold the retry owner during shutdown.
        runtime.shutdown_background();
        let socket = connected?;
        socket.set_nonblocking(false).map_err(named)?;
        let interruption = self.retain(socket.try_clone().map_err(named)?)?;
        socket.set_read_timeout(Some(bound)).map_err(named)?;
        socket.set_write_timeout(Some(bound)).map_err(named)?;
        Ok((socket, interruption))
    }
}