From 866aaf740e10fcfce169aaf81380ea0257fbe272 Mon Sep 17 00:00:00 2001 From: Ogulcan Celik Date: Thu, 21 May 2026 20:19:54 +0300 Subject: refactor: separate server client boundaries --- src/server/client_accept.rs | 50 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) create mode 100644 src/server/client_accept.rs (limited to 'src/server/client_accept.rs') diff --git a/src/server/client_accept.rs b/src/server/client_accept.rs new file mode 100644 index 00000000..61db2eb9 --- /dev/null +++ b/src/server/client_accept.rs @@ -0,0 +1,50 @@ +use std::io; +use std::os::unix::net::UnixListener; +use std::sync::{atomic::AtomicBool, Arc}; + +use tokio::sync::mpsc; +use tracing::{debug, error, warn}; + +use crate::server::client_transport::{self, ServerEvent}; + +/// Accepts pending thin-client connections and starts their handshake readers. +pub(crate) fn accept_pending_client_connections( + listener: &UnixListener, + next_client_id: &mut u64, + should_quit: &Arc, + server_event_tx: &mpsc::Sender, +) -> io::Result<()> { + loop { + match listener.accept() { + Ok((stream, _addr)) => { + let client_id = *next_client_id; + *next_client_id = next_client_id.saturating_add(1); + + if let Err(err) = stream.set_nonblocking(true) { + warn!(err = %err, "failed to set client stream nonblocking"); + continue; + } + + let should_quit = should_quit.clone(); + let server_event_tx = server_event_tx.clone(); + std::thread::spawn(move || { + if let Err(err) = client_transport::handle_client_handshake( + stream, + client_id, + &server_event_tx, + &should_quit, + ) { + debug!(client_id, err = %err, "client handshake failed"); + } + }); + } + Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => break, + Err(err) => { + error!(err = %err, "client listener accept failed"); + break; + } + } + } + + Ok(()) +} -- cgit v1.2.3