import java.net.ServerSocket;
import java.net.Socket;

// The constructor does the "bind": it reserves port 9090 for this process.
try (ServerSocket server = new ServerSocket(9090)) {

    System.out.println("Listening on port " + server.getLocalPort());

    // accept() BLOCKS until a connection arrives.
    // When it returns, it gives back a Socket ALREADY CONNECTED to the client.
    Socket connection = server.accept();

    System.out.println("Client connected from " + connection.getRemoteSocketAddress());

    // From here on, 'connection' is exactly the Socket of 09-02:
    // getInputStream(), getOutputStream(), setSoTimeout(), close()...
}

// 1. Port. The bare minimum.
new ServerSocket(9090);

// 2. Port + backlog: size of the queue of pending connections.
new ServerSocket(9090, 100);

// 3. Port + backlog + listen address.
new ServerSocket(9090, 100, InetAddress.getByName("127.0.0.1"));

// 4. Unbound, to configure before the bind (we will see it in section 4).
ServerSocket s = new ServerSocket();
s.setReuseAddress(true);
s.bind(new InetSocketAddress(9090), 100);

                 pending queue (backlog = 3)
  new       -->  [ conn ][ conn ][ conn ]  -->  accept()  -->  your code
  clients                                                      (one at a time)

  If the queue is full, the OS refuses new connections
  and the client receives ConnectException: Connection refused.

try (ServerSocket server = new ServerSocket(0)) {
    int realPort = server.getLocalPort();         // for example, 43127
    System.out.println("The system gave me port " + realPort);
}

Socket connection = server.accept();

server.setSoTimeout(1000);          // poll once a second
while (running) {
    try {
        Socket connection = server.accept();
        serve(connection);
    } catch (SocketTimeoutException e) {
        // One second with no connections: we take the chance to check
        // the stop flag and go back to waiting.
        continue;
    }
}

Exception in thread "main" java.net.BindException: Address already in use
	at java.base/sun.nio.ch.Net.bind0(Native Method)
	at java.base/java.net.ServerSocket.bind(ServerSocket.java:395)
	at java.base/java.net.ServerSocket.<init>(ServerSocket.java:262)

ss -tlnp | grep 9090
# LISTEN 0 50 0.0.0.0:9090 0.0.0.0:* users:(("java",pid=4711,fd=7))

kill 4711          # and if it resists, kill -9 4711

lsof -i :9090
netstat -an | grep 9090

// The correct form: create UNBOUND, configure, and then bind.
ServerSocket server = new ServerSocket();
server.setReuseAddress(true);                         // BEFORE the bind
server.bind(new InetSocketAddress(9090), 100);

package com.nexussoftware.bibliotech.network;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.logging.Logger;

/**
 * SEQUENTIAL echo server: it serves one client at a time.
 * It exists to demonstrate the limit of the model, not to be used.
 */
public class SequentialServer {

    private static final Logger LOG = Logger.getLogger(SequentialServer.class.getName());

    public static void main(String[] args) throws IOException {
        ServerSocket server = new ServerSocket();
        server.setReuseAddress(true);
        server.bind(new InetSocketAddress(9090), 50);

        LOG.info("Sequential server listening on 9090");

        try (server) {
            while (true) {
                // ONE connection is accepted...
                Socket connection = server.accept();
                LOG.info(() -> "Connection from " + connection.getRemoteSocketAddress());

                // ...and served IN FULL before returning to accept().
                // While this conversation lasts, nobody else is served.
                try (connection) {
                    connection.setSoTimeout(60_000);
                    serve(connection);
                } catch (IOException e) {
                    LOG.warning("Failure serving the connection: " + e.getMessage());
                }

                LOG.info("Connection finished; back to accept()");
            }
        }
    }

    private static void serve(Socket connection) throws IOException {
        BufferedReader reader = new BufferedReader(
                new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8));
        PrintWriter writer = new PrintWriter(
                new java.io.BufferedWriter(
                        new java.io.OutputStreamWriter(connection.getOutputStream(),
                                StandardCharsets.UTF_8)),
                true);

        writer.println("200 ECHO SERVER. Type QUIT to finish.");

        String line;
        while ((line = reader.readLine()) != null) {
            if (line.equalsIgnoreCase("QUIT")) {
                writer.println("221 BYE");
                return;
            }
            writer.println("ECHO: " + line);
        }
        // readLine() has returned null: the client closed without saying goodbye.
    }
}

TERMINAL 1 (first client)            TERMINAL 2 (second client)
------------------------------       ------------------------------
$ telnet localhost 9090              $ telnet localhost 9090
Trying 127.0.0.1...                  Trying 127.0.0.1...
Connected to localhost.              Connected to localhost.
200 ECHO SERVER. Type QUIT...        _
hello                                (nothing. No greeting.
ECHO: hello                           The TCP connection is made,
                                      but the server has not called
                                      accept() and reads nothing)

TERMINAL 1                           TERMINAL 2
------------------------------       ------------------------------
QUIT                                 200 ECHO SERVER. Type QUIT...
221 BYE                              (now it does! With the thread freed,
Connection closed.                    the accept() picks up this connection)

INFO: Sequential server listening on 9090
INFO: Connection from /127.0.0.1:52310
INFO: Connection finished; back to accept()
INFO: Connection from /127.0.0.1:52311

while (true) {
    Socket connection = server.accept();

    // A new thread per connection. The loop returns to accept() instantly.
    Thread thread = new Thread(() -> {
        try (connection) {
            serve(connection);
        } catch (IOException e) {
            LOG.warning("Failure while serving: " + e.getMessage());
        }
    }, "client-" + connection.getPort());
    thread.start();
}

package com.nexussoftware.bibliotech.network;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Thread factory with readable names. A thread called "bibliotech-client-3"
 * in a thread dump or in a log line is worth a thousand times more than
 * "pool-1-thread-3". This is 08-02 and 08-05 applied.
 */
public class ServerThreadFactory implements ThreadFactory {

    private final String prefix;
    private final AtomicInteger counter = new AtomicInteger(1);

    public ServerThreadFactory(String prefix) {
        this.prefix = prefix;
    }

    @Override
    public Thread newThread(Runnable r) {
        Thread thread = new Thread(r, prefix + "-" + counter.getAndIncrement());
        // NOT daemon: we want the shutdown to be explicit and orderly,
        // not the JVM killing half-finished conversations when main ends.
        thread.setDaemon(false);
        return thread;
    }
}

// Bounded pool with ThreadPoolExecutor: total control over the behaviour.
ThreadPoolExecutor pool = new ThreadPoolExecutor(
        8,                                     // core threads: always alive
        8,                                     // maximum: bounded, no surprises
        60L, TimeUnit.SECONDS,                 // idle time before dying
        new LinkedBlockingQueue<>(100),        // BOUNDED queue of waiting tasks
        new ServerThreadFactory("bibliotech-client"),
        new ThreadPoolExecutor.AbortPolicy()); // what to do if it does not fit

try {
    pool.execute(() -> serveConnection(connection));
} catch (RejectedExecutionException e) {
    // The server is overloaded. We tell the client and close.
    // Refusing fast is BETTER than accepting and not replying.
    refusePolitely(connection, "503 SERVER OVERLOADED");
}

private volatile boolean running = true;
private ServerSocket server;

/** Main loop. Runs in the acceptor thread. */
public void run() {
    while (running) {
        try {
            Socket connection = server.accept();
            pool.execute(() -> serve(connection));

        } catch (SocketException e) {
            // It may be a real failure OR our own close.
            // The 'running' flag tells the two cases apart:
            // without it, a normal shutdown appears in the log as a serious error.
            if (!running) {
                LOG.info("ServerSocket closed: orderly end of the accept loop");
                break;
            }
            LOG.log(Level.SEVERE, "Unexpected failure in accept()", e);
            break;

        } catch (IOException e) {
            // A failure accepting ONE connection must not bring the server down:
            // it is logged and we carry on accepting.
            LOG.log(Level.WARNING, "Error accepting a connection", e);
        }
    }
}

/** Called from another thread (menu, shutdown hook, signal). */
public void stop() {
    running = false;                 // 1. volatile flag: visible from the other thread
    closeQuietly(server);            // 2. this unblocks the accept() instantly
    // 3. and then, the two-phase shutdown of the pool
}

public void stop() {
    running = false;
    closeQuietly(server);

    // PHASE 1: no new tasks are accepted, but the ones in progress finish.
    pool.shutdown();
    try {
        // We give a reasonable deadline for the conversations to end by themselves.
        if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
            LOG.warning("Conversations still active after 10 s: forcing the close");

            // PHASE 2: interrupt whatever threads are left.
            pool.shutdownNow();

            if (!pool.awaitTermination(5, TimeUnit.SECONDS)) {
                LOG.severe("The pool has not finished; there are stuck threads");
            }
        }
    } catch (InterruptedException e) {
        // We were interrupted while waiting: force it and RESTORE the flag.
        pool.shutdownNow();
        Thread.currentThread().interrupt();    // 08-02: never swallow the interruption
    }
}

// Inside the thread serving Marta, and at the same time inside the one serving Diego,
// and at the same time inside the one serving Nuria:
Material material = catalog.findByIsbn(isbn);          // safe: ConcurrentHashMap
catalog.registerQuery();                               // safe: LongAdder

package com.nexussoftware.bibliotech.network;

import com.nexussoftware.bibliotech.domain.Material;
import com.nexussoftware.bibliotech.service.ConcurrentCatalog;
import com.nexussoftware.bibliotech.service.SafeLoanRegistry;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Server for BiblioTech's BTCP/1 protocol.
 *
 * - One acceptor thread that only accepts and delegates.
 * - A BOUNDED pool of threads that serve the conversations (module 8).
 * - Shared state in ConcurrentCatalog (ConcurrentHashMap): it was already safe.
 * - Strict validation of everything arriving over the network.
 * - setSoTimeout per connection to evict idle clients.
 * - Two-phase graceful shutdown.
 */
public class CatalogServer implements AutoCloseable {

    private static final Logger LOG = Logger.getLogger(CatalogServer.class.getName());

    // --- Defensive limits: every value coming from the network is bounded ---
    /** No legitimate BTCP/1 request exceeds this. Prevents memory exhaustion. */
    private static final int MAX_LINE_LENGTH = 512;
    /** Consecutive requests per connection: stops a client monopolising a thread. */
    private static final int MAX_REQUESTS = 1_000;
    /** A BiblioTech ISBN has this shape and nothing else. */
    private static final int MAX_ISBN = 20;
    private static final int MAX_EMPLOYEE = 60;

    private final int port;
    private final String bindAddress;
    private final int threads;
    private final int waitQueue;
    private final int idleTimeoutMs;

    private final ConcurrentCatalog catalog;
    private final SafeLoanRegistry loans;

    private ServerSocket server;
    private ThreadPoolExecutor pool;

    /** volatile: written by the stopping thread, read by the acceptor thread (08-04). */
    private volatile boolean running = false;

    private final AtomicInteger activeConnections = new AtomicInteger();
    private final AtomicLong totalConnections = new AtomicLong();
    private final AtomicLong servedRequests = new AtomicLong();
    private final AtomicLong refusedConnections = new AtomicLong();

    public CatalogServer(int port, String bindAddress, int threads,
                         int waitQueue, int idleTimeoutMs,
                         ConcurrentCatalog catalog,
                         SafeLoanRegistry loans) {
        this.port = port;
        this.bindAddress = bindAddress;
        this.threads = threads;
        this.waitQueue = waitQueue;
        this.idleTimeoutMs = idleTimeoutMs;
        this.catalog = catalog;
        this.loans = loans;
    }

    // =================================================================
    // Startup
    // =================================================================

    /** Prepares the socket and the pool. Does not block. */
    public void start() throws IOException {
        // Three-step pattern: create unbound, configure, bind.
        // setReuseAddress MUST come before the bind, or it achieves nothing.
        server = new ServerSocket();
        server.setReuseAddress(true);
        server.bind(new InetSocketAddress(bindAddress, port), waitQueue);

        ThreadFactory factory = new ThreadFactory() {
            private final AtomicInteger n = new AtomicInteger(1);

            @Override
            public Thread newThread(Runnable r) {
                // Readable names: in a thread dump and in every log line
                // you will know exactly who is doing what (08-02).
                Thread t = new Thread(r, "bibliotech-client-" + n.getAndIncrement());
                t.setDaemon(false);
                return t;
            }
        };

        pool = new ThreadPoolExecutor(
                threads, threads,
                60L, TimeUnit.SECONDS,
                // BOUNDED queue: newFixedThreadPool uses an unbounded one and that
                // turns an overload into an OutOfMemoryError.
                new LinkedBlockingQueue<>(waitQueue),
                factory,
                // Refusing fast and telling the client is better than
                // accepting and not replying.
                new ThreadPoolExecutor.AbortPolicy());

        running = true;
        LOG.info(() -> "CatalogServer BTCP/1 listening on "
                + bindAddress + ":" + port
                + " (pool=" + threads + ", queue=" + waitQueue + ")");
    }

    /** Accept loop. Blocks until close() is called. */
    public void run() {
        while (running) {
            Socket connection = null;
            try {
                connection = server.accept();       // BLOCKS here

                // IDLE time limit per connection. It is what evicts the
                // clients that connect and go quiet, and also what lets
                // shutdownNow() actually stop the threads.
                connection.setSoTimeout(idleTimeoutMs);
                connection.setTcpNoDelay(true);     // interactive protocol: no Nagle

                final Socket accepted = connection;
                totalConnections.incrementAndGet();

                // The acceptor thread does NOT serve: it delegates and returns to accept().
                pool.execute(() -> serveConnection(accepted));

            } catch (RejectedExecutionException e) {
                // Pool and queue full: the server is overloaded.
                refusedConnections.incrementAndGet();
                LOG.warning("Server overloaded; connection refused");
                refuse(connection, "503 SERVER OVERLOADED");

            } catch (SocketException e) {
                // Our own close() OR a real failure. The flag decides.
                if (!running) {
                    LOG.info("Accept loop finished in an orderly way");
                    return;
                }
                LOG.log(Level.SEVERE, "Unexpected failure in accept()", e);
                return;

            } catch (IOException e) {
                // A failure accepting ONE connection does not bring the server down.
                LOG.log(Level.WARNING, "Error accepting a connection", e);
                closeQuietly(connection);
            }
        }
    }

    // =================================================================
    // Serving one connection (runs in a pool thread)
    // =================================================================

    private void serveConnection(Socket connection) {
        String client = String.valueOf(connection.getRemoteSocketAddress());
        int active = activeConnections.incrementAndGet();
        LOG.info(() -> "[" + client + "] connected (" + active + " active)");

        // try-with-resources: the socket is closed whatever happens.
        try (connection) {
            BufferedReader reader = new BufferedReader(
                    new InputStreamReader(connection.getInputStream(),
                            StandardCharsets.UTF_8));
            PrintWriter writer = new PrintWriter(
                    new BufferedWriter(
                            new OutputStreamWriter(connection.getOutputStream(),
                                    StandardCharsets.UTF_8)),
                    false);     // no autoFlush: we flush by hand after each reply

            // The server speaks first, as the protocol specifies.
            respond(writer, "200 BIBLIOTECH BTCP/1");

            int requests = 0;
            while (!Thread.currentThread().isInterrupted()) {

                if (requests++ >= MAX_REQUESTS) {
                    // A client cannot keep a thread indefinitely.
                    respond(writer, "421 TOO MANY REQUESTS");
                    LOG.warning("[" + client + "] exceeded the request limit");
                    return;
                }

                String line = readBoundedLine(reader);
                if (line == null) {
                    LOG.info(() -> "[" + client + "] closed the connection");
                    return;
                }

                servedRequests.incrementAndGet();
                boolean carryOn = process(line, writer, client);
                if (!carryOn) {
                    return;
                }
            }

        } catch (SocketTimeoutException e) {
            // The client connected and went quiet. It is evicted: the thread returns to the pool.
            LOG.info(() -> "[" + client + "] evicted for inactivity ("
                    + idleTimeoutMs + " ms)");

        } catch (LineTooLongException e) {
            LOG.warning("[" + client + "] sent a line longer than "
                    + MAX_LINE_LENGTH + " characters; connection cut");

        } catch (SocketException e) {
            // "Connection reset": the client went down. It is normal, not a serious error.
            LOG.info(() -> "[" + client + "] connection lost: " + e.getMessage());

        } catch (IOException e) {
            LOG.log(Level.WARNING, "[" + client + "] I/O failure", e);

        } catch (RuntimeException e) {
            // A programming failure while handling ONE client cannot
            // kill the pool thread or affect the others. This catch
            // is the error boundary of 06-07 applied to each connection.
            LOG.log(Level.SEVERE, "[" + client + "] unexpected failure", e);

        } finally {
            int left = activeConnections.decrementAndGet();
            LOG.info(() -> "[" + client + "] disconnected (" + left + " active)");
        }
    }

    // =================================================================
    // Processing one request
    // =================================================================

    /** Returns false if the connection must be closed after this request. */
    private boolean process(String line, PrintWriter writer, String client) {
        // The command is separated from the rest at the FIRST space: the
        // employee name may contain spaces and we do not want to split it.
        String command;
        String rest;
        int space = line.indexOf(' ');
        if (space < 0) {
            command = line;
            rest = "";
        } else {
            command = line.substring(0, space);
            rest = line.substring(space + 1).strip();
        }

        // The commands are compared in upper case to be tolerant,
        // but the protocol documents them in upper case.
        switch (command.toUpperCase()) {

            case "QUERY" -> {
                if (!validIsbn(rest)) {
                    respond(writer, "400 BAD REQUEST invalid isbn");
                    return true;
                }
                Material material = catalog.findByIsbn(rest);
                if (material == null) {
                    respond(writer, "404 NOT FOUND " + rest);
                } else {
                    respond(writer, "200 OK " + format(material));
                }
                return true;
            }

            case "LIST" -> {
                if (!rest.isEmpty()) {
                    respond(writer, "400 BAD REQUEST LIST takes no arguments");
                    return true;
                }
                List<Material> materials = catalog.all();
                respond(writer, "201 LIST " + materials.size());
                for (Material m : materials) {
                    respond(writer, format(m));
                }
                respond(writer, ".");
                return true;
            }

            case "LEND" -> {
                int sep = rest.indexOf(' ');
                if (sep < 0) {
                    respond(writer, "400 BAD REQUEST missing arguments");
                    return true;
                }
                String isbn = rest.substring(0, sep);
                String employee = rest.substring(sep + 1).strip();

                if (!validIsbn(isbn) || !validEmployee(employee)) {
                    respond(writer, "400 BAD REQUEST invalid arguments");
                    return true;
                }
                if (catalog.findByIsbn(isbn) == null) {
                    respond(writer, "404 NOT FOUND " + isbn);
                    return true;
                }

                // ATOMIC COMPOUND OPERATION: checking availability and lending
                // in a single call to the service. Chaining two safe
                // operations does NOT give a safe operation (08-06).
                boolean lent = loans.lendIfAvailable(isbn, employee);
                if (lent) {
                    LOG.info(() -> "[" + client + "] loan of " + isbn
                            + " to " + employee);
                    respond(writer, "200 OK loan registered");
                } else {
                    respond(writer, "409 NOT AVAILABLE " + isbn);
                }
                return true;
            }

            case "QUIT" -> {
                respond(writer, "221 BYE");
                return false;       // close the connection
            }

            default -> {
                // We do NOT return the command to the client unsanitised:
                // it could contain terminal escape sequences.
                respond(writer, "400 BAD REQUEST unknown command: "
                        + sanitise(command));
                return true;
            }
        }
    }

    private String format(Material m) {
        return m.getIsbn() + "|" + m.getTitle() + "|"
                + m.getType() + "|" + m.isAvailable();
    }

    // =================================================================
    // Low-level protocol I/O
    // =================================================================

    /** Writes one protocol line and FLUSHES. Without flush, the client waits forever. */
    private void respond(PrintWriter writer, String line) {
        writer.print(line + "\n");          // explicit \n: the protocol demands it
        writer.flush();                     // MANDATORY
        if (writer.checkError()) {
            // PrintWriter swallows IOException; checkError is the only clue.
            LOG.fine("Failed to write the reply; the client probably closed");
        }
    }

    /** Internal exception for a line that exceeds the limit. */
    private static class LineTooLongException extends IOException {
        LineTooLongException(String m) {
            super(m);
        }
    }

    /**
     * Reads a BOUNDED line. BufferedReader.readLine() has no limit:
     * a malicious client can send 100 MB with not one line break
     * and exhaust the server's memory. This is a mandatory defence
     * in any exposed server.
     */
    private String readBoundedLine(BufferedReader reader) throws IOException {
        StringBuilder sb = new StringBuilder();
        int c;
        while ((c = reader.read()) != -1) {
            if (c == '\n') {
                // We also accept \r\n by dropping the trailing carriage return.
                int end = sb.length();
                if (end > 0 && sb.charAt(end - 1) == '\r') {
                    sb.setLength(end - 1);
                }
                return sb.toString();
            }
            if (sb.length() >= MAX_LINE_LENGTH) {
                throw new LineTooLongException(
                        "Line longer than " + MAX_LINE_LENGTH + " characters");
            }
            sb.append((char) c);
        }
        // End of stream. Anything half-written is discarded: it is not a valid line.
        return null;
    }

    // =================================================================
    // Validation: NEVER trust what arrives over the network
    // =================================================================

    private boolean validIsbn(String isbn) {
        if (isbn == null || isbn.isBlank() || isbn.length() > MAX_ISBN) {
            return false;
        }
        // ALLOW list of characters: digits and hyphens only. Everything else out.
        // An allow list is always safer than a deny list.
        for (int i = 0; i < isbn.length(); i++) {
            char c = isbn.charAt(i);
            if (!Character.isDigit(c) && c != '-') {
                return false;
            }
        }
        return true;
    }

    private boolean validEmployee(String employee) {
        if (employee == null || employee.isBlank() || employee.length() > MAX_EMPLOYEE) {
            return false;
        }
        for (int i = 0; i < employee.length(); i++) {
            char c = employee.charAt(i);
            // Letters, digits, spaces, underscores and hyphens. No
            // control characters, which would break the protocol or the log.
            if (!Character.isLetterOrDigit(c) && c != ' ' && c != '_' && c != '-') {
                return false;
            }
        }
        return true;
    }

    /**
     * Sanitises a text coming from the network before returning or logging it.
     * Without this, a client can inject \n (a fake reply in the
     * protocol) or ANSI escape sequences (which manipulate the terminal
     * of whoever reads the log). This is LOG INJECTION, a real problem.
     */
    private String sanitise(String text) {
        StringBuilder sb = new StringBuilder();
        int limit = Math.min(text.length(), 40);
        for (int i = 0; i < limit; i++) {
            char c = text.charAt(i);
            sb.append(Character.isISOControl(c) ? '?' : c);
        }
        return sb.toString();
    }

    // =================================================================
    // Graceful shutdown
    // =================================================================

    @Override
    public void close() {
        if (!running) {
            return;
        }
        LOG.info("Shutting down CatalogServer...");

        // 1. Flag BEFORE the close: it tells a clean shutdown from a failure.
        running = false;

        // 2. Closing the ServerSocket is the ONLY thing that unblocks accept().
        //    Thread.interrupt() cannot do it.
        closeQuietly(server);

        // 3. Two-phase shutdown of the pool (08-05).
        if (pool != null) {
            pool.shutdown();        // no new tasks; the current ones carry on
            try {
                if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
                    LOG.warning("Conversations active after 10 s: forcing the close");
                    pool.shutdownNow();     // interrupts the threads
                    if (!pool.awaitTermination(5, TimeUnit.SECONDS)) {
                        LOG.severe("There are threads left unfinished");
                    }
                }
            } catch (InterruptedException e) {
                pool.shutdownNow();
                Thread.currentThread().interrupt();   // 08-02: restore the flag
            }
        }

        LOG.info(() -> String.format(
                "CatalogServer stopped. Connections: %d total, %d refused. "
                + "Requests served: %d",
                totalConnections.get(), refusedConnections.get(),
                servedRequests.get()));
    }

    private void refuse(Socket connection, String message) {
        if (connection == null) {
            return;
        }
        // Courtesy: telling the client why it is being closed, instead of
        // cutting it off with no explanation. It costs two lines and saves support.
        try (connection) {
            PrintWriter writer = new PrintWriter(
                    new OutputStreamWriter(connection.getOutputStream(),
                            StandardCharsets.UTF_8));
            writer.print(message + "\n");
            writer.flush();
        } catch (IOException e) {
            LOG.fine("Could not notify the refusal: " + e.getMessage());
        }
    }

    private void closeQuietly(java.io.Closeable resource) {
        if (resource != null) {
            try {
                resource.close();
            } catch (IOException ignored) {
                // While closing there is nothing left to save.
            }
        }
    }

    // --- Metrics for the BiblioTech menu ---

    public int activeConnections() {
        return activeConnections.get();
    }

    public long totalConnections() {
        return totalConnections.get();
    }

    public long servedRequests() {
        return servedRequests.get();
    }
}

package com.nexussoftware.bibliotech.presentation;

import com.nexussoftware.bibliotech.persistence.Configuration;
import com.nexussoftware.bibliotech.network.CatalogServer;
import com.nexussoftware.bibliotech.service.ConcurrentCatalog;
import com.nexussoftware.bibliotech.service.SafeLoanRegistry;
import com.nexussoftware.bibliotech.infra.LogConfiguration;

import java.io.IOException;
import java.util.logging.Logger;

/** Startup of BiblioTech's catalogue server. */
public class BiblioTechServerApp {

    private static final Logger LOG = Logger.getLogger(BiblioTechServerApp.class.getName());

    public static void main(String[] args) throws IOException {
        LogConfiguration.initialise();      // the module 6 logger, already configured

        // The configuration comes from bibliotech.properties (07-07).
        Configuration config = Configuration.load();
        int port = config.integer("network.port", 9090);
        String bind = config.text("network.bind", "0.0.0.0");
        int threads = config.integer("network.threads", 16);
        int queue = config.integer("network.queue", 100);
        int idle = config.integer("network.idle.ms", 60_000);

        ConcurrentCatalog catalog = ConcurrentCatalog.loadFromFile();
        SafeLoanRegistry loans = new SafeLoanRegistry(catalog);

        CatalogServer server = new CatalogServer(
                port, bind, threads, queue, idle, catalog, loans);
        server.start();

        // Shutdown hook: it runs on Ctrl+C or on SIGTERM (which is what
        // Docker, systemd or Kubernetes send when stopping a container).
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            LOG.info("Shutdown signal received");
            server.close();
        }, "bibliotech-shutdown"));

        // Blocks here until the ServerSocket is closed.
        server.run();

        LOG.info("Server finished");
    }
}

javac -d classes $(find src -name "*.java")
java -cp classes com.nexussoftware.bibliotech.presentation.BiblioTechServerApp

INFO: CatalogServer BTCP/1 listening on 0.0.0.0:9090 (pool=16, queue=100)

telnet localhost 9090

Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
200 BIBLIOTECH BTCP/1
QUERY 978-0000000001                                     <-- YOU TYPE
200 OK 978-0000000001|Effective Java|BOOK|true
QUERY 999                                                <-- YOU TYPE
404 NOT FOUND 999
QUERY ../../etc/passwd                                   <-- YOU TYPE
400 BAD REQUEST invalid isbn
LIST                                                     <-- YOU TYPE
201 LIST 3
978-0000000001|Effective Java|BOOK|true
978-0000000002|Design Patterns|BOOK|false
978-0000000003|Refactoring|BOOK|true
.
LEND 978-0000000003 Nuria Vidal                          <-- YOU TYPE
200 OK loan registered
LEND 978-0000000003 Diego Alonso                         <-- YOU TYPE
409 NOT AVAILABLE 978-0000000003
MAKE ME A COFFEE                                         <-- YOU TYPE
400 BAD REQUEST unknown command: MAKE
QUIT                                                     <-- YOU TYPE
221 BYE
Connection closed by foreign host.

printf 'LIST\nQUIT\n' | nc localhost 9090

200 BIBLIOTECH BTCP/1
201 LIST 3
978-0000000001|Effective Java|BOOK|true
978-0000000002|Design Patterns|BOOK|false
978-0000000003|Refactoring|BOOK|true
.
221 BYE

INFO: [/127.0.0.1:52440] connected (1 active)
INFO: [/127.0.0.1:52441] connected (2 active)
INFO: [/127.0.0.1:52441] loan of 978-0000000003 to Nuria Vidal
INFO: [/127.0.0.1:52442] connected (3 active)
INFO: [/127.0.0.1:52442] closed the connection
INFO: [/127.0.0.1:52442] disconnected (2 active)
INFO: [/127.0.0.1:52441] disconnected (1 active)
INFO: [/127.0.0.1:52440] disconnected (0 active)

Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    LOG.info("Shutdown signal received");
    server.close();
}, "bibliotech-shutdown"));

Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    LOG.info("Shutting down BiblioTech...");
    server.close();                   // 1. stop accepting and close conversations
    GracefulShutdown.stopServices();  // 2. notice, maintenance and reservation pools
    LoanStore.dump();                 // 3. persist what is pending (module 7)
    LOG.info("BiblioTech stopped correctly");
}, "bibliotech-shutdown"));

// Non-blocking I/O: ONE thread serves thousands of connections.
Selector selector = Selector.open();
ServerSocketChannel channel = ServerSocketChannel.open();
channel.bind(new InetSocketAddress(9090));
channel.configureBlocking(false);                        // key point: non-blocking
channel.register(selector, SelectionKey.OP_ACCEPT);

while (true) {
    selector.select();          // blocks until SOME channel has activity
    for (SelectionKey key : selector.selectedKeys()) {
        if (key.isAcceptable()) { /* accept */ }
        if (key.isReadable())   { /* read whatever is there, without blocking */ }
    }
}

// Java 21+: one VIRTUAL thread per connection. No pool, no practical limit.
try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
    while (running) {
        Socket connection = server.accept();
        executor.submit(() -> serveConnection(connection));
    }
}

package com.nexussoftware.bibliotech.network;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * BiblioTech statistics server.
 *
 * SECURITY: it listens ONLY on 127.0.0.1. A service's metrics reveal
 * information useful to an attacker (load, saturation, memory, windows of
 * low activity) and are usually the first place something leaks by accident.
 * Exposing them to the network would be a design flaw; if they must be seen
 * from outside, it is done over an SSH tunnel or behind an authenticated proxy (12-07).
 */
public class StatisticsServer implements AutoCloseable {

    private static final Logger LOG = Logger.getLogger(StatisticsServer.class.getName());

    private final int port;
    private final CatalogServer main;
    private final ThreadPoolExecutor mainPool;
    private final ExecutorService pool;
    private final long startInstant = System.currentTimeMillis();

    private ServerSocket server;
    private volatile boolean running = false;

    public StatisticsServer(int port, CatalogServer main,
                            ThreadPoolExecutor mainPool, ExecutorService pool) {
        this.port = port;
        this.main = main;
        this.mainPool = mainPool;
        this.pool = pool;       // the pool is shared: no threads of its own needed
    }

    public void start() throws IOException {
        server = new ServerSocket();
        server.setReuseAddress(true);
        // The third argument of bind TIES the listening to loopback:
        // not even a firewall failure would expose this port.
        server.bind(new InetSocketAddress(
                InetAddress.getLoopbackAddress(), port), 10);
        running = true;

        // Its own acceptor thread, because accept() blocks.
        Thread acceptor = new Thread(this::run, "bibliotech-stats-acceptor");
        acceptor.setDaemon(true);       // this one can be a daemon: it has no state
        acceptor.start();

        LOG.info(() -> "Statistics server on 127.0.0.1:" + port);
    }

    private void run() {
        while (running) {
            try {
                Socket connection = server.accept();
                connection.setSoTimeout(5_000);
                // Delegated to the shared pool: we do not serve in the acceptor.
                pool.execute(() -> serveReport(connection));

            } catch (SocketException e) {
                if (!running) {
                    LOG.info("Statistics server stopped");
                    return;
                }
                LOG.log(Level.SEVERE, "Failure in the statistics accept()", e);
                return;

            } catch (IOException e) {
                LOG.log(Level.WARNING, "Error accepting a statistics connection", e);
            }
        }
    }

    /**
     * This service has no protocol: on connect, it writes and closes.
     * That is what makes it queryable with a simple "nc localhost 9092".
     */
    private void serveReport(Socket connection) {
        try (connection) {
            BufferedWriter writer = new BufferedWriter(
                    new OutputStreamWriter(connection.getOutputStream(),
                            StandardCharsets.UTF_8));
            writer.write(buildReport());
            writer.flush();         // the usual flush
        } catch (IOException e) {
            LOG.fine("Failure serving the report: " + e.getMessage());
        }
    }

    private String buildReport() {
        Runtime rt = Runtime.getRuntime();
        long used = (rt.totalMemory() - rt.freeMemory()) / (1024 * 1024);
        long maximum = rt.maxMemory() / (1024 * 1024);
        long seconds = (System.currentTimeMillis() - startInstant) / 1000;

        StringBuilder sb = new StringBuilder();
        sb.append("=== BIBLIOTECH - SERVER STATUS ===\n");
        sb.append(String.format("%-28s %s%n", "Up for",
                formatDuration(seconds)));
        sb.append('\n');
        sb.append("--- CONNECTIONS ---\n");
        sb.append(String.format("%-28s %d%n", "Active", main.activeConnections()));
        sb.append(String.format("%-28s %d%n", "Total", main.totalConnections()));
        sb.append(String.format("%-28s %d%n", "Requests served",
                main.servedRequests()));
        sb.append('\n');
        sb.append("--- THREAD POOL ---\n");
        sb.append(String.format("%-28s %d%n", "Active threads",
                mainPool.getActiveCount()));
        sb.append(String.format("%-28s %d%n", "Pool size",
                mainPool.getPoolSize()));
        sb.append(String.format("%-28s %d%n", "Queued tasks",
                mainPool.getQueue().size()));
        sb.append(String.format("%-28s %d%n", "Completed tasks",
                mainPool.getCompletedTaskCount()));
        sb.append('\n');
        sb.append("--- MEMORY ---\n");
        sb.append(String.format("%-28s %d MB of %d MB (%.1f%%)%n",
                "Used", used, maximum, 100.0 * used / maximum));
        sb.append(String.format("%-28s %d%n", "Available cores",
                rt.availableProcessors()));
        return sb.toString();
    }

    private String formatDuration(long seconds) {
        // Without java.time (that is 10-05): simple integer arithmetic.
        long h = seconds / 3600;
        long m = (seconds % 3600) / 60;
        long s = seconds % 60;
        return String.format("%dh %02dm %02ds", h, m, s);
    }

    @Override
    public void close() {
        running = false;
        try {
            if (server != null) {
                server.close();         // unblocks the accept()
            }
        } catch (IOException ignored) {
            // While closing there is nothing left to save.
        }
    }
}

nc localhost 9092

=== BIBLIOTECH - SERVER STATUS ===
Up for                       0h 04m 17s

--- CONNECTIONS ---
Active                       2
Total                        47
Requests served              183

--- THREAD POOL ---
Active threads               2
Pool size                    16
Queued tasks                 0
Completed tasks              45

--- MEMORY ---
Used                         28 MB of 4096 MB (0.7%)
Available cores              8

package com.nexussoftware.bibliotech.network;

import java.net.Socket;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Logger;

/**
 * Limiter of simultaneous connections per IP address.
 *
 * Stops a single machine (by mistake or on purpose) taking up the whole
 * thread pool of the server and leaving the other workstations with no service.
 */
public class IpLimiter {

    private static final Logger LOG = Logger.getLogger(IpLimiter.class.getName());

    private final int maxPerIp;
    private final boolean exemptLoopback;

    /** Counts active connections per IP. Concurrent: every thread touches it. */
    private final ConcurrentHashMap<String, AtomicInteger> perIp = new ConcurrentHashMap<>();
    private final AtomicLong refusedByLimit = new AtomicLong();

    public IpLimiter(int maxPerIp, boolean exemptLoopback) {
        this.maxPerIp = maxPerIp;
        this.exemptLoopback = exemptLoopback;
    }

    /**
     * Tries to reserve a slot for this IP.
     * Returns true if admitted; false if the limit has been reached.
     */
    public boolean admit(Socket connection) {
        String ip = ipOf(connection);

        if (exemptLoopback && isLoopback(ip)) {
            // Without the exemption, testing the server with several telnet
            // sessions from the machine itself would be impossible.
            return true;
        }

        // computeIfAbsent is atomic: if two threads arrive at once with the
        // same IP, only one creates the AtomicInteger and both use the same one.
        AtomicInteger counter = perIp.computeIfAbsent(ip, key -> new AtomicInteger());

        // Compare-and-swap loop: the atomic way of saying
        // "increment ONLY IF the current value is lower than the maximum".
        // An "if (get() < max) incrementAndGet()" would have a race:
        // two threads could pass the if at once and exceed the limit.
        while (true) {
            int current = counter.get();
            if (current >= maxPerIp) {
                refusedByLimit.incrementAndGet();
                LOG.warning("Limit of " + maxPerIp
                        + " connections reached for " + ip);
                return false;
            }
            if (counter.compareAndSet(current, current + 1)) {
                return true;    // slot secured
            }
            // compareAndSet failed: another thread changed the value in between.
            // We retry with the new value. This is 08-06.
        }
    }

    /**
     * Releases the slot. It MUST always be called, from a finally.
     */
    public void release(Socket connection) {
        String ip = ipOf(connection);

        if (exemptLoopback && isLoopback(ip)) {
            return;     // nothing was reserved, nothing is released
        }

        // computeIfPresent does the "decrement and, if it reaches zero,
        // remove the entry" pair ATOMICALLY. Returning null from the
        // function removes the key from the map.
        //
        // Doing it in two steps would have a classic race:
        //     if (counter.decrementAndGet() == 0) perIp.remove(ip);
        // Between the decrement and the remove, another thread may increment
        // that same counter to 1 (a new connection from the same IP), and
        // the remove would delete a counter that is already 1. The new
        // connection would be counted in an orphaned object and the limit
        // would stop applying to that IP.
        perIp.computeIfPresent(ip, (key, counter) -> {
            int left = counter.decrementAndGet();
            return left <= 0 ? null : counter;       // null removes the entry
        });
    }

    private String ipOf(Socket connection) {
        return connection.getInetAddress().getHostAddress();
    }

    private boolean isLoopback(String ip) {
        return ip.equals("127.0.0.1") || ip.equals("0:0:0:0:0:0:0:1") || ip.equals("::1");
    }

    public long refusedByLimit() {
        return refusedByLimit.get();
    }

    public int connectedIps() {
        return perIp.size();
    }
}

// In run(), after accepting:
Socket accepted = server.accept();
accepted.setSoTimeout(idleTimeoutMs);
accepted.setTcpNoDelay(true);

if (!limiter.admit(accepted)) {
    refuse(accepted, "429 TOO MANY CONNECTIONS");
    continue;       // it never got delegated to the pool: nothing to release
}

try {
    pool.execute(() -> serveConnection(accepted));
} catch (RejectedExecutionException e) {
    // CAREFUL: the slot was ALREADY reserved. If it is not released here,
    // one slot leaks on every pool refusal and the IP ends up
    // blocked forever.
    limiter.release(accepted);
    refuse(accepted, "503 SERVER OVERLOADED");
}

} finally {
    limiter.release(connection);        // ALWAYS, whatever happens
    int left = activeConnections.decrementAndGet();
    LOG.info(() -> "[" + client + "] disconnected (" + left + " active)");
}

package com.nexussoftware.bibliotech.network;

import com.nexussoftware.bibliotech.exception.BiblioTechException;

import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Load test of CatalogServer using the CatalogClient of 09-02.
 * Measures latencies, error rate and requests per second.
 */
public class LoadTest {

    private static final String[] ISBNS = {
            "978-0000000001",
            "978-0000000002",
            "978-0000000003",
            "000-0000000000"    // non-existent on purpose: must give 404, not an error
    };

    private final String host;
    private final int port;

    private final AtomicInteger successful = new AtomicInteger();
    private final AtomicInteger errors = new AtomicInteger();
    private final AtomicInteger failedConnections = new AtomicInteger();
    private final AtomicLong latencyIndex = new AtomicLong();

    /** Latencies in microseconds. Pre-allocated array: no synchronisation on write. */
    private long[] latencies;

    public LoadTest(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void run(int clients, int requestsPerClient) throws InterruptedException {
        latencies = new long[clients * requestsPerClient];

        // Pool for the test clients, with named threads.
        ExecutorService pool = Executors.newFixedThreadPool(clients, r -> {
            Thread t = new Thread(r);
            t.setName("load-" + t.getId());
            return t;
        });

        // Two latches: one so that ALL of them start at once (real load,
        // not a ramp), and another to wait for ALL of them to finish (08-05).
        CountDownLatch start = new CountDownLatch(1);
        CountDownLatch finish = new CountDownLatch(clients);

        for (int i = 0; i < clients; i++) {
            final int id = i;
            pool.execute(() -> {
                try {
                    start.await();      // everybody waits here for the starting gun
                    work(id, requestsPerClient);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    finish.countDown();
                }
            });
        }

        System.out.printf("Launching %d clients x %d requests = %d requests%n%n",
                clients, requestsPerClient, clients * requestsPerClient);

        long t0 = System.nanoTime();
        start.countDown();                      // go!
        finish.await();                         // we wait for everybody
        long durationNs = System.nanoTime() - t0;

        // Two-phase shutdown of the test pool (08-05).
        pool.shutdown();
        if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
            pool.shutdownNow();
        }

        report(durationNs);
    }

    private void work(int id, int requests) {
        // Each client opens its own connection: CatalogClient is NOT
        // thread-safe, and this way we also measure real connections.
        try (CatalogClient client = new CatalogClient(host, port)) {
            client.connect();

            for (int i = 0; i < requests; i++) {
                String isbn = ISBNS[(id + i) % ISBNS.length];

                long t0 = System.nanoTime();
                try {
                    client.query(isbn);         // null (404) is a success too
                    long us = (System.nanoTime() - t0) / 1000;

                    // Each thread writes into its own array slot:
                    // no collisions, no synchronisation. The index is
                    // handed out by an atomic counter.
                    int pos = (int) latencyIndex.getAndIncrement();
                    if (pos < latencies.length) {
                        latencies[pos] = us;
                    }
                    successful.incrementAndGet();

                } catch (BiblioTechException e) {
                    errors.incrementAndGet();
                }
            }
        } catch (BiblioTechException e) {
            // We could not even connect: counted separately, because it means
            // something different (overload) from an error in one request.
            failedConnections.incrementAndGet();
        }
    }

    private void report(long durationNs) {
        int done = (int) Math.min(latencyIndex.get(), latencies.length);
        if (done == 0) {
            System.out.println("No request was completed.");
            System.out.println("Failed connections: " + failedConnections.get());
            return;
        }

        long[] sorted = Arrays.copyOf(latencies, done);
        Arrays.sort(sorted);        // 05-09: sorting for the percentiles

        long sum = 0;
        for (long v : sorted) {
            sum += v;
        }

        double seconds = durationNs / 1_000_000_000.0;
        double perSecond = successful.get() / seconds;

        System.out.println("=== LOAD TEST RESULT ===");
        System.out.printf("%-26s %.2f s%n", "Total duration", seconds);
        System.out.printf("%-26s %d%n", "Successful requests", successful.get());
        System.out.printf("%-26s %d%n", "Failed requests", errors.get());
        System.out.printf("%-26s %d%n", "Failed connections", failedConnections.get());
        System.out.printf("%-26s %.0f%n", "Requests per second", perSecond);
        System.out.println();
        System.out.println("--- LATENCIES (ms) ---");
        System.out.printf("%-26s %.3f%n", "Minimum", sorted[0] / 1000.0);
        System.out.printf("%-26s %.3f%n", "Mean", (sum / (double) done) / 1000.0);
        System.out.printf("%-26s %.3f%n", "Median (p50)", percentile(sorted, 50) / 1000.0);
        System.out.printf("%-26s %.3f%n", "p95", percentile(sorted, 95) / 1000.0);
        System.out.printf("%-26s %.3f%n", "p99", percentile(sorted, 99) / 1000.0);
        System.out.printf("%-26s %.3f%n", "Maximum",
                sorted[sorted.length - 1] / 1000.0);
    }

    private long percentile(long[] sorted, int p) {
        // Index of the percentile over the already-sorted array.
        int index = (int) Math.ceil(p / 100.0 * sorted.length) - 1;
        return sorted[Math.max(0, Math.min(index, sorted.length - 1))];
    }

    public static void main(String[] args) throws InterruptedException {
        String host = args.length > 0 ? args[0] : "localhost";
        int port = args.length > 1 ? Integer.parseInt(args[1]) : 9090;

        for (int clients : new int[]{4, 16, 64}) {
            System.out.println("############ " + clients + " CLIENTS ############");
            new LoadTest(host, port).run(clients, 200);
            System.out.println();
            Thread.sleep(1000);     // room for the server to recover
        }
    }
}

############ 4 CLIENTS ############
Launching 4 clients x 200 requests = 800 requests

=== LOAD TEST RESULT ===
Total duration             0.42 s
Successful requests        800
Failed requests            0
Failed connections         0
Requests per second        1897

--- LATENCIES (ms) ---
Minimum                    0.118
Mean                       0.204
Median (p50)               0.171
p95                        0.392
p99                        0.884
Maximum                    4.113

############ 16 CLIENTS ############
Requests per second        6104
Median (p50)               0.224
p95                        0.668
p99                        1.942
Maximum                    11.208

############ 64 CLIENTS ############
Failed connections         0
Requests per second        6438
Median (p50)               1.102
p95                        4.873
p99                        12.441
Maximum                    58.306