TCP Echo Server

A multi-client TCP echo and command server binds a TcpListener, accepts concurrent client connections across worker threads, enforces per-socket read/write timeouts, tracks active connection metrics with atomic counters, and processes framed commands (ECHO, UPPER, STATS, QUIT).

Concurrent TCP Server Architecture:

                     +------------------------------------+
                     | `TcpListener` (127.0.0.1:ephemeral)|
                     | `ServerStats` (`Arc<ServerStats>`) |
                     +------------------------------------+
                                       |
              +------------------------+------------------------+
              | `accept()`                                      | `accept()`
              v                                                 v
  [ Worker Thread 1 (`TcpStream`) ]                 [ Worker Thread 2 (`TcpStream`) ]
  - `set_read_timeout(5s)`                          - `set_read_timeout(5s)`
  - `BufReader::new(&stream)`                       - `BufReader::new(&stream)`
  - Handles `ECHO`, `UPPER`, `STATS`, `QUIT`        - Handles `ECHO`, `UPPER`, `STATS`, `QUIT`

Protocol Commands and Server Telemetry

Complete Runnable Implementation

use std::io::{self, BufRead, BufReader, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::thread;
use std::time::Duration;

#[derive(Default)]
pub struct ServerStats {
    pub active_connections: AtomicUsize,
    pub total_commands: AtomicU64,
    pub bytes_echoed: AtomicU64,
}

/// RAII guard that increments `active_connections` on connect and decrements it on disconnect!
struct ConnectionGuard {
    stats: Arc<ServerStats>,
}

impl ConnectionGuard {
    fn enter(stats: Arc<ServerStats>) -> Self {
        stats.active_connections.fetch_add(1, Ordering::Relaxed);
        Self { stats }
    }
}

impl Drop for ConnectionGuard {
    fn drop(&mut self) {
        self.stats.active_connections.fetch_sub(1, Ordering::Relaxed);
    }
}

pub fn handle_client(mut stream: TcpStream, stats: Arc<ServerStats>) -> io::Result<()> {
    let _guard = ConnectionGuard::enter(Arc::clone(&stats));
    stream.set_read_timeout(Some(Duration::from_secs(5)))?;
    stream.set_write_timeout(Some(Duration::from_secs(5)))?;
    stream.set_nodelay(true)?;

    let mut reader = BufReader::new(&stream);
    let mut line = String::new();

    while reader.read_line(&mut line)? > 0 {
        let trimmed = line.trim_end_matches(['\r', '\n']);
        stats.total_commands.fetch_add(1, Ordering::Relaxed);

        if trimmed == "QUIT" {
            stream.write_all(b"BYE\n")?;
            break;
        } else if trimmed == "STATS" {
            let reply = format!(
                "active={} cmds={} bytes={}\n",
                stats.active_connections.load(Ordering::Relaxed),
                stats.total_commands.load(Ordering::Relaxed),
                stats.bytes_echoed.load(Ordering::Relaxed),
            );
            stream.write_all(reply.as_bytes())?;
        } else if let Some(rest) = trimmed.strip_prefix("UPPER ") {
            let upper = format!("{}\n", rest.to_uppercase());
            stats
                .bytes_echoed
                .fetch_add(rest.len() as u64, Ordering::Relaxed);
            stream.write_all(upper.as_bytes())?;
        } else {
            let payload = trimmed.strip_prefix("ECHO ").unwrap_or(trimmed);
            stats
                .bytes_echoed
                .fetch_add(payload.len() as u64, Ordering::Relaxed);
            writeln!(stream, "{payload}")?;
        }

        line.clear();
    }

    Ok(())
}

fn run_test_session(addr: SocketAddr, requests: &[&str]) -> io::Result<Vec<String>> {
    let mut stream = TcpStream::connect(addr)?;
    let mut reader = BufReader::new(&stream);
    let mut responses = Vec::new();

    for &req in requests {
        writeln!(stream, "{req}")?;
        let mut reply = String::new();
        reader.read_line(&mut reply)?;
        responses.push(reply.trim_end().to_string());
    }
    Ok(responses)
}

fn main() -> io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let server_addr = listener.local_addr()?;
    let stats = Arc::new(ServerStats::default());

    let stats_server = Arc::clone(&stats);
    let server_thread = thread::spawn(move || -> io::Result<()> {
        let mut workers = Vec::new();
        for stream_res in listener.incoming().take(2) {
            let stream = stream_res?;
            let st = Arc::clone(&stats_server);
            workers.push(thread::spawn(move || handle_client(stream, st)));
        }
        for w in workers {
            w.join().expect("worker panicked")?;
        }
        Ok(())
    });

    // Session 1: ECHO and UPPER
    let r1 = run_test_session(
        server_addr,
        &["ECHO hello world", "UPPER rust 2024", "QUIT"],
    )?;
    assert_eq!(r1, vec!["hello world", "RUST 2024", "BYE"]);

    // Session 2: Query server telemetry via STATS
    let r2 = run_test_session(server_addr, &["STATS", "QUIT"])?;
    assert_eq!(r2[0], "active=1 cmds=4 bytes=20");

    server_thread.join().expect("server panicked")?;
    assert_eq!(stats.active_connections.load(Ordering::Relaxed), 0);

    println!("Client 1 replies: {r1:?}");
    println!("Client 2 stats:   {}", r2[0]);
    Ok(())
}

Client 1 replies: ["hello world", "RUST 2024", "BYE"]
Client 2 stats:   active=1 cmds=4 bytes=20