1 files changed, 50 insertions, 0 deletions
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<AtomicBool>,
+    server_event_tx: &mpsc::Sender<ServerEvent>,
+) -> 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(())
+}