byte[] data = "WHERE IS BIBLIOTECH?".getBytes(StandardCharsets.UTF_8);

DatagramPacket packet = new DatagramPacket(
        data,                                    // the content
        data.length,                             // how many bytes of that array
        InetAddress.getByName("192.168.1.50"),   // where it goes
        9091);                                   // to which port

byte[] buffer = new byte[1024];

// No address or port: receive() will fill them in with the SENDER's.
DatagramPacket packet = new DatagramPacket(buffer, buffer.length);

socket.receive(packet);         // blocks until something arrives

// Now the packet DOES have a sender, and we know how many bytes arrived.
InetAddress sender = packet.getAddress();
int senderPort = packet.getPort();
int bytesReceived = packet.getLength();

// EPHEMERAL socket: the system assigns a free port.
// It serves to send and to receive replies to what was sent.
DatagramSocket client = new DatagramSocket();

// Socket bound to a WELL-KNOWN port: this is what a UDP server does.
DatagramSocket server = new DatagramSocket(9091);

// Bound to a port and to a particular interface.
DatagramSocket local = new DatagramSocket(9091, InetAddress.getByName("127.0.0.1"));

package com.nexussoftware.bibliotech.network;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.nio.charset.StandardCharsets;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * UDP echo server. One thread, one socket, every client.
 * No connections, no pool, no per-client state.
 */
public class UdpEchoServer implements AutoCloseable {

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

    /** Size of the receive buffer. See section 8 about the MTU. */
    private static final int MAX_DATAGRAM = 1400;

    private final int port;
    private DatagramSocket socket;
    private volatile boolean running = false;

    public UdpEchoServer(int port) {
        this.port = port;
    }

    public void run() throws SocketException {
        socket = new DatagramSocket(port);
        running = true;
        LOG.info(() -> "UDP echo server on port " + port);

        // ONE buffer and ONE packet, reused throughout the loop:
        // creating a 1400-byte array for every received packet
        // would generate needless garbage thousands of times a second.
        byte[] buffer = new byte[MAX_DATAGRAM];
        DatagramPacket packet = new DatagramPacket(buffer, buffer.length);

        while (running) {
            try {
                // ESSENTIAL before every receive: restore the length.
                // If the previous packet had 20 bytes, getLength() is 20
                // and only 20 bytes of the next one would be received. Section 6.
                packet.setLength(buffer.length);

                socket.receive(packet);         // BLOCKS until something arrives

                // getLength() -> what REALLY arrived, not buffer.length.
                String text = new String(packet.getData(), packet.getOffset(),
                        packet.getLength(), StandardCharsets.UTF_8);

                LOG.info(() -> "From " + packet.getSocketAddress() + " ("
                        + packet.getLength() + " bytes): " + text);

                // We reply TO THE SENDER, which comes in the packet itself.
                byte[] reply = ("ECHO: " + text).getBytes(StandardCharsets.UTF_8);
                DatagramPacket back = new DatagramPacket(
                        reply, reply.length,
                        packet.getAddress(), packet.getPort());
                socket.send(back);

            } catch (SocketException e) {
                if (!running) {
                    LOG.info("UDP echo server stopped");
                    return;
                }
                LOG.log(Level.SEVERE, "UDP socket failure", e);
                return;

            } catch (IOException e) {
                // A failure with ONE packet cannot bring the server down.
                LOG.log(Level.WARNING, "Error processing a datagram", e);
            }
        }
    }

    @Override
    public void close() {
        running = false;
        if (socket != null) {
            // Closing the socket unblocks the receive(), just as closing
            // the ServerSocket unblocked the accept() in 09-03.
            socket.close();
        }
    }

    public static void main(String[] args) throws SocketException {
        UdpEchoServer server = new UdpEchoServer(9095);
        Runtime.getRuntime().addShutdownHook(new Thread(server::close));
        server.run();
    }
}

package com.nexussoftware.bibliotech.network;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketTimeoutException;
import java.nio.charset.StandardCharsets;

/** UDP echo client with a time limit. */
public class UdpEchoClient {

    private static final int MAX_DATAGRAM = 1400;

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

        // Ephemeral socket: the system gives us a free port.
        // try-with-resources: DatagramSocket is Closeable.
        try (DatagramSocket socket = new DatagramSocket()) {

            // MANDATORY. Without this, if the reply is lost
            // -and with UDP it can be- the receive() never returns.
            socket.setSoTimeout(2000);

            InetAddress destination = InetAddress.getByName(host);

            for (String message : new String[]{"Hello", "BiblioTech", "Naïve Set Theory"}) {
                byte[] data = message.getBytes(StandardCharsets.UTF_8);
                socket.send(new DatagramPacket(data, data.length, destination, port));
                System.out.println("-> " + message + " (" + data.length + " bytes)");

                byte[] buffer = new byte[MAX_DATAGRAM];
                DatagramPacket reply = new DatagramPacket(buffer, buffer.length);
                try {
                    socket.receive(reply);
                    System.out.println("<- " + new String(reply.getData(),
                            reply.getOffset(), reply.getLength(),
                            StandardCharsets.UTF_8));
                } catch (SocketTimeoutException e) {
                    // With UDP this is NOT a program error: it is the
                    // normal behaviour when something gets lost.
                    System.out.println("<- (no reply in 2 s: packet lost"
                            + " or server down)");
                }
            }
        }
    }
}

# Terminal 1
java -cp classes com.nexussoftware.bibliotech.network.UdpEchoServer

# Terminal 2
java -cp classes com.nexussoftware.bibliotech.network.UdpEchoClient

Terminal 2:
-> Hello (5 bytes)
<- ECHO: Hello
-> BiblioTech (10 bytes)
<- ECHO: BiblioTech
-> Naïve Set Theory (17 bytes)
<- ECHO: Naïve Set Theory

echo "Hello from nc" | nc -u -w1 localhost 9095

ECHO: Hello from nc

// BROKEN CODE. Do not copy it.
byte[] buffer = new byte[1400];
DatagramPacket packet = new DatagramPacket(buffer, buffer.length);

while (true) {
    socket.receive(packet);         // <-- the setLength before it is missing
    process(packet);
}

Initial state:  packet.getLength() = 1400   (the whole buffer)

1st receive:  "Hello" arrives (5 bytes)
              -> receive writes 5 bytes and sets getLength() = 5

2nd receive:  the limit is now 5 bytes.
              "BiblioTech" arrives (10 bytes)
              -> ONLY 5 are accepted: "Bibli"
              -> the remaining 5 ARE DISCARDED WITH NO WARNING
              -> getLength() = 5

3rd receive:  the limit is still 5. And so on forever.

From /127.0.0.1:51234 (5 bytes): Hello
From /127.0.0.1:51234 (5 bytes): Bibli
From /127.0.0.1:51234 (5 bytes): Refac

// CORRECT: restore the length BEFORE every receive.
while (true) {
    packet.setLength(buffer.length);        // <-- the missing line
    socket.receive(packet);
    process(packet);
}

// Alternative: a new packet on every turn. Correct, but it generates
// garbage. In a server receiving thousands of packets per second,
// reuse with setLength is markedly better.
while (true) {
    DatagramPacket packet = new DatagramPacket(new byte[1400], 1400);
    socket.receive(packet);
    process(packet);
}

// BROKEN CODE. Do not copy it.
socket.receive(packet);
String text = new String(packet.getData(), StandardCharsets.UTF_8);
//                       ^^^^^^^^^^^^^^^^ the WHOLE array, 1400 bytes

With the bug, receiving "Hello" and then "Hi":

  1st message: "Hello" + 1395 bytes of zeros
               -> "Hello\0\0\0\0\0\0..."  (seems to work, deceptive)

  2nd message: "Hi" + "llo" (remains of the previous one) + zeros
               -> "Hillo\0\0\0..."   <-- MIXED-UP DATA!

// CORRECT: always offset and getLength().
String text = new String(
        packet.getData(),
        packet.getOffset(),         // from where
        packet.getLength(),         // how many bytes are valid
        StandardCharsets.UTF_8);    // explicit charset, as always

byte[] useful = Arrays.copyOfRange(
        packet.getData(),
        packet.getOffset(),
        packet.getOffset() + packet.getLength());

  1500 bytes of Ethernet MTU
-   20 bytes of IPv4 header   (40 if it is IPv6)
-    8 bytes of UDP header
= 1472 bytes of data that fit in ONE Ethernet frame

// Useful trick for detecting truncation: ask for one byte more.
// If getLength() == buffer.length, it is VERY likely that a bigger
// packet arrived and was truncated.
byte[] buffer = new byte[MAX_DATAGRAM + 1];
DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
socket.receive(packet);

if (packet.getLength() > MAX_DATAGRAM) {
    LOG.warning("Datagram of " + packet.getLength()
            + " bytes: it exceeds the expected maximum. Discarded.");
    return;
}

package com.nexussoftware.bibliotech.network;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketTimeoutException;
import java.nio.charset.StandardCharsets;
import java.util.HashSet;
import java.util.Set;

/**
 * Shows that a UDP receiver must tolerate loss, duplication and disorder.
 * The sender sends N numbered packets; the receiver analyses what arrived.
 */
public class UdpReliabilityDemo {

    private static final int PORT = 9096;
    private static final int TOTAL = 20;

    /** Receiver: counts what arrives and detects gaps, duplicates and disorder. */
    static void receiver() throws IOException {
        try (DatagramSocket socket = new DatagramSocket(PORT)) {
            socket.setSoTimeout(3000);      // without this, if the last one is lost,
                                            // the loop would never finish

            byte[] buffer = new byte[512];
            DatagramPacket packet = new DatagramPacket(buffer, buffer.length);

            Set<Integer> seen = new HashSet<>();
            int duplicates = 0;
            int outOfOrder = 0;
            int lastSeen = -1;

            while (true) {
                packet.setLength(buffer.length);        // classic mistake 1
                try {
                    socket.receive(packet);
                } catch (SocketTimeoutException e) {
                    break;      // 3 s with nothing: we call it finished
                }

                String text = new String(packet.getData(), packet.getOffset(),
                        packet.getLength(), StandardCharsets.UTF_8);   // classic mistake 2
                int number = Integer.parseInt(text.substring(text.indexOf('#') + 1));

                if (!seen.add(number)) {
                    // add() returns false if it was already there: a DUPLICATE.
                    duplicates++;
                    System.out.println("  DUPLICATE: #" + number);
                    continue;
                }
                if (number < lastSeen) {
                    // It arrived after a higher one: DISORDER.
                    outOfOrder++;
                    System.out.println("  OUT OF ORDER: #" + number
                            + " after #" + lastSeen);
                }
                lastSeen = Math.max(lastSeen, number);
            }

            // The missing ones are LOST.
            StringBuilder lost = new StringBuilder();
            for (int i = 0; i < TOTAL; i++) {
                if (!seen.contains(i)) {
                    lost.append('#').append(i).append(' ');
                }
            }

            System.out.println();
            System.out.println("=== ANALYSIS OF " + TOTAL + " PACKETS ===");
            System.out.println("Unique received  : " + seen.size());
            System.out.println("Lost             : " + (TOTAL - seen.size())
                    + (lost.length() == 0 ? "" : "  -> " + lost));
            System.out.println("Duplicated       : " + duplicates);
            System.out.println("Out of order     : " + outOfOrder);
        }
    }

    /** Sender: sends TOTAL numbered packets as fast as possible. */
    static void sender() throws IOException, InterruptedException {
        try (DatagramSocket socket = new DatagramSocket()) {
            InetAddress destination = InetAddress.getByName("localhost");

            for (int i = 0; i < TOTAL; i++) {
                byte[] data = ("PACKET#" + i).getBytes(StandardCharsets.UTF_8);
                socket.send(new DatagramPacket(data, data.length, destination, PORT));
                // send() has "succeeded". That does NOT mean it arrived.
            }
            System.out.println("Sent " + TOTAL + " packets");
        }
    }

    public static void main(String[] args) throws Exception {
        if (args.length > 0 && args[0].equals("sender")) {
            sender();
        } else {
            receiver();
        }
    }
}

=== ANALYSIS OF 100000 PACKETS ===
Unique received  : 73412
Lost             : 26588
Duplicated       : 0
Out of order     : 0

# Add 10% loss, 5% duplication and variable delay to loopback
sudo tc qdisc add dev lo root netem loss 10% duplicate 5% delay 20ms 10ms

# ... run the test ...

# ALWAYS remove it when you finish
sudo tc qdisc del dev lo root

  DUPLICATE: #3
  OUT OF ORDER: #7 after #8
  DUPLICATE: #11

=== ANALYSIS OF 20 PACKETS ===
Unique received  : 18
Lost             : 2  -> #5 #14
Duplicated       : 2
Out of order     : 1

package com.nexussoftware.bibliotech.network;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketTimeoutException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ThreadLocalRandom;
import java.util.logging.Logger;

/**
 * Reliable request-reply over UDP: identifier, time limit,
 * retry with increasing backoff and discarding of unmatched replies.
 *
 * WARNING: if you need this for all your traffic, use TCP. This makes
 * sense for short, isolated exchanges (as DNS does), not as a general
 * replacement for TCP.
 */
public class ReliableUdpRequest {

    private static final Logger LOG = Logger.getLogger(ReliableUdpRequest.class.getName());
    private static final int MAX_DATAGRAM = 1400;

    private final InetAddress destination;
    private final int port;
    private final int maxAttempts;
    private final int initialWaitMs;

    public ReliableUdpRequest(InetAddress destination, int port,
                              int maxAttempts, int initialWaitMs) {
        this.destination = destination;
        this.port = port;
        this.maxAttempts = maxAttempts;
        this.initialWaitMs = initialWaitMs;
    }

    /**
     * Sends a request and waits for a reply, retrying.
     * Returns null if there was no reply after all the attempts.
     *
     * Format: "<id> <request>"  ->  "<id> <reply>"
     */
    public String ask(String request) throws IOException {
        // Random identifier: to match replies AND to make it hard for
        // a third party to guess the id and inject a fake reply.
        int id = ThreadLocalRandom.current().nextInt(1, Integer.MAX_VALUE);
        byte[] data = (id + " " + request).getBytes(StandardCharsets.UTF_8);

        try (DatagramSocket socket = new DatagramSocket()) {
            // connect() in UDP does not connect: it filters. We will only
            // accept packets from this destination, which rules out injections.
            socket.connect(destination, port);

            int wait = initialWaitMs;
            byte[] buffer = new byte[MAX_DATAGRAM];
            DatagramPacket reply = new DatagramPacket(buffer, buffer.length);

            for (int attempt = 1; attempt <= maxAttempts; attempt++) {
                socket.send(new DatagramPacket(data, data.length, destination, port));
                socket.setSoTimeout(wait);

                // Inner loop: replies from earlier attempts we had given
                // up for lost may arrive. They have to be discarded and
                // we must keep waiting for ours within the same deadline.
                long limit = System.currentTimeMillis() + wait;
                while (System.currentTimeMillis() < limit) {
                    try {
                        reply.setLength(buffer.length);         // classic mistake 1
                        socket.receive(reply);

                        String text = new String(reply.getData(),
                                reply.getOffset(), reply.getLength(),
                                StandardCharsets.UTF_8);        // classic mistake 2

                        int sep = text.indexOf(' ');
                        if (sep < 0) {
                            LOG.fine("Reply with an invalid format; discarded");
                            continue;
                        }
                        int receivedId = Integer.parseInt(text.substring(0, sep));
                        if (receivedId != id) {
                            // Reply to another request: discarded.
                            LOG.fine("Reply with id " + receivedId
                                    + ", we expected " + id + "; discarded");
                            continue;
                        }
                        return text.substring(sep + 1);         // ours!

                    } catch (SocketTimeoutException e) {
                        break;      // this attempt's deadline expired
                    } catch (NumberFormatException e) {
                        LOG.fine("Non-numeric identifier; discarded");
                    }
                }

                LOG.info("Attempt " + attempt + "/" + maxAttempts
                        + " with no reply (wait " + wait + " ms)");

                // INCREASING BACKOFF: double it on every attempt. Retrying
                // at the same rate over a congested network makes it worse.
                wait *= 2;
            }
            return null;    // no reply after all the attempts
        }
    }
}

try (DatagramSocket socket = new DatagramSocket()) {
    // MANDATORY. Without this, sending to a broadcast address
    // throws SocketException: Permission denied.
    socket.setBroadcast(true);

    byte[] data = "WHERE IS BIBLIOTECH?".getBytes(StandardCharsets.UTF_8);
    socket.send(new DatagramPacket(data, data.length,
            InetAddress.getByName("255.255.255.255"), 9091));
}

try (DatagramSocket socket = new DatagramSocket(9091)) {
    byte[] buffer = new byte[1400];
    DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
    socket.receive(packet);         // it also receives broadcasts to 9091
}

socket.setTimeToLive(1);        // it does not leave the local network

package com.nexussoftware.bibliotech.network;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MulticastSocket;
import java.net.NetworkInterface;
import java.nio.charset.StandardCharsets;
import java.util.Enumeration;
import java.util.logging.Logger;

/**
 * Multicast sender and receiver for BiblioTech's internal notices
 * ("the catalogue has been updated", "maintenance in 10 minutes").
 */
public class MulticastNotices implements AutoCloseable {

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

    /** Range 239.x.x.x: administrative scope, reserved for private use. */
    private static final String GROUP = "239.10.10.10";
    private static final int PORT = 9097;
    private static final int MAX_DATAGRAM = 1400;

    private MulticastSocket socket;
    private InetSocketAddress group;
    private NetworkInterface networkInterface;
    private volatile boolean running = false;

    /** Joins the group and gets ready to receive. */
    public void join() throws IOException {
        socket = new MulticastSocket(PORT);
        socket.setTimeToLive(1);            // it does not leave the local network

        group = new InetSocketAddress(InetAddress.getByName(GROUP), PORT);
        networkInterface = chooseInterface();

        // MODERN API (Java 14+): joinGroup(SocketAddress, NetworkInterface).
        // The old joinGroup(InetAddress) is deprecated because it does not let
        // you state the interface, and on a machine with Wi-Fi + Ethernet + Docker
        // the system almost never picks the one you want.
        socket.joinGroup(group, networkInterface);
        running = true;

        LOG.info(() -> "Joined group " + GROUP + ":" + PORT
                + " through interface " + networkInterface.getName());
    }

    /**
     * Chooses an interface that is up, not loopback and supports multicast.
     * In production this ought to be configurable in bibliotech.properties.
     */
    private NetworkInterface chooseInterface() throws IOException {
        Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
        while (interfaces.hasMoreElements()) {
            NetworkInterface ni = interfaces.nextElement();
            if (ni.isUp() && !ni.isLoopback() && ni.supportsMulticast()) {
                return ni;
            }
        }
        // Fallback: on a machine with no network, loopback at least lets
        // a sender and a receiver on the same machine talk to each other.
        return NetworkInterface.getByName("lo");
    }

    /** Sends a notice to everybody subscribed to the group. */
    public void announce(String message) throws IOException {
        byte[] data = message.getBytes(StandardCharsets.UTF_8);
        if (data.length > MAX_DATAGRAM) {
            throw new IOException("Notice too long: " + data.length + " bytes");
        }
        socket.send(new DatagramPacket(data, data.length,
                InetAddress.getByName(GROUP), PORT));
        LOG.info(() -> "Notice sent to the group: " + message);
    }

    /** Listening loop. Call it from a thread of its own. */
    public void listen(java.util.function.Consumer<String> onReceive) {
        byte[] buffer = new byte[MAX_DATAGRAM];
        DatagramPacket packet = new DatagramPacket(buffer, buffer.length);

        while (running) {
            try {
                packet.setLength(buffer.length);        // classic mistake 1
                socket.receive(packet);

                String message = new String(packet.getData(), packet.getOffset(),
                        packet.getLength(), StandardCharsets.UTF_8);    // classic mistake 2

                // CAREFUL: we also receive OUR OWN notices. If that bothers you,
                // it is disabled with socket.setOption(StandardSocketOptions.IP_MULTICAST_LOOP, false)
                // or filtered by sender, comparing against the local addresses.
                onReceive.accept(message);

            } catch (IOException e) {
                if (!running) {
                    return;     // orderly close
                }
                LOG.warning("Error receiving a notice: " + e.getMessage());
            }
        }
    }

    @Override
    public void close() {
        running = false;
        if (socket != null) {
            try {
                socket.leaveGroup(group, networkInterface);     // modern API here too
            } catch (IOException e) {
                LOG.fine("Failure leaving the group: " + e.getMessage());
            }
            socket.close();     // unblocks the receive()
        }
    }
}

BTDP/1 - BiblioTech Discovery Protocol
======================================
Transport : UDP, broadcast to port 9091
Encoding : UTF-8
Maximum size : 512 bytes (well below any MTU)

Request (workstation -> broadcast):
  BTDP/1 WHERE <requestId>

Reply (server -> sender, unicast):
  BTDP/1 HERE <requestId> <ip> <tcpPort> <serverName>

Example:
  -> BTDP/1 WHERE 748291
  <- BTDP/1 HERE 748291 192.168.1.50 9090 bibliotech-central

package com.nexussoftware.bibliotech.network;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Replies to BiblioTech's discovery requests.
 * It runs alongside the CatalogServer of 09-03, in its own thread.
 *
 * One socket, one thread, every desk in the office:
 * that is what a UDP server is like, with no pool and no per-client state.
 */
public class DiscoveryResponder implements AutoCloseable, Runnable {

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

    private static final int DISCOVERY_PORT = 9091;
    /** 512 bytes: the universal prudent limit, the same one DNS uses. */
    private static final int MAX_DATAGRAM = 512;
    private static final String VERSION = "BTDP/1";

    private final int tcpPort;
    private final String serverName;

    private DatagramSocket socket;
    private volatile boolean running = false;
    private final AtomicLong answered = new AtomicLong();
    private final AtomicLong discarded = new AtomicLong();

    public DiscoveryResponder(int tcpPort, String serverName) {
        this.tcpPort = tcpPort;
        this.serverName = serverName;
    }

    public void start() throws SocketException {
        socket = new DatagramSocket(DISCOVERY_PORT);
        running = true;
        LOG.info(() -> "Discovery responder on UDP:" + DISCOVERY_PORT);
    }

    @Override
    public void run() {
        // A buffer one byte bigger than the maximum, to detect truncation.
        byte[] buffer = new byte[MAX_DATAGRAM + 1];
        DatagramPacket packet = new DatagramPacket(buffer, buffer.length);

        while (running) {
            try {
                packet.setLength(buffer.length);        // CLASSIC MISTAKE 1
                socket.receive(packet);

                if (packet.getLength() > MAX_DATAGRAM) {
                    // A datagram bigger than expected: suspicious. Discarded.
                    discarded.incrementAndGet();
                    LOG.warning("Datagram of " + packet.getLength()
                            + " bytes discarded because of its size");
                    continue;
                }

                String request = new String(packet.getData(), packet.getOffset(),
                        packet.getLength(), StandardCharsets.UTF_8);    // CLASSIC MISTAKE 2

                process(request.strip(), packet.getAddress(), packet.getPort());

            } catch (SocketException e) {
                if (!running) {
                    LOG.info("Discovery responder stopped");
                    return;
                }
                LOG.log(Level.SEVERE, "Discovery socket failure", e);
                return;

            } catch (IOException e) {
                // A bad packet cannot bring the service down.
                LOG.log(Level.WARNING, "Error processing a discovery request", e);
            }
        }
    }

    private void process(String request, InetAddress sender, int senderPort) {
        // EVERYTHING arriving over the network is untrusted, and here it
        // arrives from ANYBODY on the local network, with no connection to identify it.
        String[] parts = request.split(" ");

        if (parts.length != 3 || !parts[0].equals(VERSION)
                || !parts[1].equals("WHERE")) {
            discarded.incrementAndGet();
            // We do not reply to what we do not understand: replying to arbitrary
            // packets would turn this service into an AMPLIFIER for reflected
            // denial-of-service attacks.
            LOG.fine(() -> "Unrecognised discovery request from " + sender);
            return;
        }

        String id = parts[2];
        if (!validIdentifier(id)) {
            discarded.incrementAndGet();
            return;
        }

        try {
            String myIp = InetAddress.getLocalHost().getHostAddress();
            String reply = VERSION + " HERE " + id + " " + myIp + " "
                    + tcpPort + " " + serverName;

            byte[] data = reply.getBytes(StandardCharsets.UTF_8);

            // UNICAST to the sender: there is no need to bother the whole
            // network with the reply. The question broadcasts; the answer does not.
            socket.send(new DatagramPacket(data, data.length,
                    sender, senderPort));

            answered.incrementAndGet();
            LOG.info(() -> "Discovery answered to " + sender.getHostAddress()
                    + " (id " + id + ")");

        } catch (IOException e) {
            // If the send fails, nothing serious happens: the client will retry.
            LOG.log(Level.WARNING, "Could not answer the discovery request", e);
        }
    }

    /** Allow list: digits only, 10 at most. No arbitrary data. */
    private boolean validIdentifier(String id) {
        if (id.isEmpty() || id.length() > 10) {
            return false;
        }
        for (int i = 0; i < id.length(); i++) {
            if (!Character.isDigit(id.charAt(i))) {
                return false;
            }
        }
        return true;
    }

    @Override
    public void close() {
        running = false;
        if (socket != null) {
            socket.close();     // unblocks the receive()
        }
        LOG.info(() -> "Discovery: " + answered.get() + " answered, "
                + discarded.get() + " discarded");
    }
}

package com.nexussoftware.bibliotech.network;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketTimeoutException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ThreadLocalRandom;
import java.util.logging.Logger;

/**
 * Looks for the BiblioTech server on the local network by broadcast.
 * Removes the need to configure the server's IP on every desk.
 */
public class ServerFinder {

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

    private static final int DISCOVERY_PORT = 9091;
    private static final int MAX_DATAGRAM = 512;
    private static final String VERSION = "BTDP/1";
    private static final int MAX_ATTEMPTS = 3;

    /** What is discovered: where the server is. A record from 04-07. */
    public record FoundServer(String ip, int tcpPort, String name) {
    }

    /**
     * Looks for the server. Returns null if nobody replies after the retries.
     * @param initialWaitMs deadline of the first attempt; it doubles on each one.
     */
    public FoundServer find(int initialWaitMs) throws IOException {
        int id = ThreadLocalRandom.current().nextInt(1, 1_000_000);
        String request = VERSION + " WHERE " + id;
        byte[] data = request.getBytes(StandardCharsets.UTF_8);

        try (DatagramSocket socket = new DatagramSocket()) {
            // Without this, sending to 255.255.255.255 throws
            // SocketException: Permission denied.
            socket.setBroadcast(true);

            // CAREFUL: connect() cannot be used here, because the reply
            // will come from the server's IP and not from the broadcast
            // address we sent to. connect() would filter it and we would discard it.

            byte[] buffer = new byte[MAX_DATAGRAM + 1];
            DatagramPacket reply = new DatagramPacket(buffer, buffer.length);
            InetAddress broadcast = InetAddress.getByName("255.255.255.255");

            int wait = initialWaitMs;

            for (int attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
                socket.send(new DatagramPacket(data, data.length,
                        broadcast, DISCOVERY_PORT));
                LOG.info("Looking for BiblioTech on the local network (attempt "
                        + attempt + "/" + MAX_ATTEMPTS + ")...");

                socket.setSoTimeout(wait);
                long limit = System.currentTimeMillis() + wait;

                while (System.currentTimeMillis() < limit) {
                    try {
                        reply.setLength(buffer.length);         // classic mistake 1
                        socket.receive(reply);

                        String text = new String(reply.getData(),
                                reply.getOffset(), reply.getLength(),
                                StandardCharsets.UTF_8);        // classic mistake 2

                        FoundServer found = parse(text, id);
                        if (found != null) {
                            LOG.info(() -> "BiblioTech found: " + found);
                            return found;
                        }
                        // A reply with another id or malformed: keep waiting.

                    } catch (SocketTimeoutException e) {
                        break;      // deadline expired, next attempt
                    }
                }
                wait *= 2;          // increasing backoff
            }

            LOG.warning("No BiblioTech server has replied on the local network");
            return null;
        }
    }

    /** Parses the reply VALIDATING everything: it comes from a stranger on the network. */
    private FoundServer parse(String text, int expectedId) {
        String[] p = text.strip().split(" ");
        if (p.length != 6 || !p[0].equals(VERSION) || !p[1].equals("HERE")) {
            return null;
        }
        try {
            if (Integer.parseInt(p[2]) != expectedId) {
                return null;    // a reply to another question: discarded
            }
            int port = Integer.parseInt(p[4]);
            if (port < 1 || port > 65535) {
                LOG.warning("Invalid port in the reply: " + port);
                return null;
            }
            // The IP is validated by trying to parse it: if it is not valid,
            // getByName will throw and we will not connect to anything odd.
            InetAddress.getByName(p[3]);

            return new FoundServer(p[3], port, p[5]);

        } catch (NumberFormatException | IOException e) {
            LOG.fine("Invalid discovery reply: " + text);
            return null;
        }
    }

    public static void main(String[] args) throws IOException {
        FoundServer server = new ServerFinder().find(500);
        if (server == null) {
            System.out.println("No server was found.");
            System.out.println("Set the IP by hand in bibliotech.properties.");
        } else {
            System.out.println("Server: " + server.name());
            System.out.println("Connect to " + server.ip() + ":" + server.tcpPort());
            // And from here on, the CatalogClient of 09-02 over TCP.
        }
    }
}

# Terminal 1: the server (or the responder on its own)
java -cp classes com.nexussoftware.bibliotech.network.DiscoveryResponder

# Terminal 2: the workstation
java -cp classes com.nexussoftware.bibliotech.network.ServerFinder

Terminal 2:
INFO: Looking for BiblioTech on the local network (attempt 1/3)...
INFO: BiblioTech found: FoundServer[ip=192.168.1.50, tcpPort=9090, name=bibliotech-central]
Server: bibliotech-central
Connect to 192.168.1.50:9090

Terminal 1:
INFO: Discovery answered to 192.168.1.23 (id 748291)

package com.nexussoftware.bibliotech.network;

import com.nexussoftware.bibliotech.service.BiblioTechStatistics;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * BiblioTech's telemetry publisher.
 *
 * Sends the state of the service over UDP every N seconds. It uses the
 * ScheduledExecutorService of 08-05 and NEVER blocks the server:
 * if the collector is down, the datagrams are lost and nothing happens.
 *
 * Format (one-line text, StatsD style):
 *   bibliotech.loans.active:12|g
 *   bibliotech.queries.total:8471|c
 */
public class TelemetryPublisher implements AutoCloseable {

    private static final Logger LOG = Logger.getLogger(TelemetryPublisher.class.getName());
    private static final int MAX_DATAGRAM = 512;

    private final String collectorHost;
    private final int collectorPort;
    private final int intervalSeconds;
    private final BiblioTechStatistics statistics;
    private final CatalogServer server;

    private DatagramSocket socket;
    private InetAddress destination;
    private ScheduledExecutorService scheduler;

    private final AtomicLong sent = new AtomicLong();
    private final AtomicLong failed = new AtomicLong();

    public TelemetryPublisher(String collectorHost, int collectorPort,
                              int intervalSeconds,
                              BiblioTechStatistics statistics,
                              CatalogServer server) {
        this.collectorHost = collectorHost;
        this.collectorPort = collectorPort;
        this.intervalSeconds = intervalSeconds;
        this.statistics = statistics;
        this.server = server;
    }

    public void start() throws IOException {
        socket = new DatagramSocket();       // ephemeral: we only send

        // The name is resolved ONCE at startup, not on every send:
        // a DNS lookup every 5 seconds is a waste and could also
        // block the scheduler's thread.
        destination = InetAddress.getByName(collectorHost);

        scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread t = new Thread(r, "bibliotech-telemetry");
            t.setDaemon(true);      // telemetry must not prevent exiting
            return t;
        });

        // scheduleAtFixedRate from 08-05: every intervalSeconds, without piling up.
        scheduler.scheduleAtFixedRate(this::publish,
                intervalSeconds, intervalSeconds, TimeUnit.SECONDS);

        LOG.info(() -> "Telemetry to " + collectorHost + ":" + collectorPort
                + " every " + intervalSeconds + " s");
    }

    /**
     * Sends a batch of measurements. Runs in the scheduler's thread.
     *
     * CRITICAL: this method must NOT throw anything. An exception escaping
     * a scheduleAtFixedRate task CANCELS the task silently and the
     * telemetry stops publishing without anybody noticing (08-05).
     */
    private void publish() {
        try {
            send("bibliotech.loans.active:"
                    + statistics.activeLoans() + "|g");
            send("bibliotech.queries.total:"
                    + statistics.totalQueries() + "|c");
            send("bibliotech.connections.active:"
                    + server.activeConnections() + "|g");
            send("bibliotech.requests.total:"
                    + server.servedRequests() + "|c");

            Runtime rt = Runtime.getRuntime();
            long memoryMb = (rt.totalMemory() - rt.freeMemory()) / (1024 * 1024);
            send("bibliotech.memory.mb:" + memoryMb + "|g");

        } catch (RuntimeException e) {
            // Logged and we carry on: the next batch will try again.
            LOG.log(Level.WARNING, "Failure publishing telemetry", e);
        }
    }

    private void send(String measurement) {
        byte[] data = measurement.getBytes(StandardCharsets.UTF_8);
        if (data.length > MAX_DATAGRAM) {
            LOG.warning("Measurement too long; discarded: " + measurement);
            return;
        }
        try {
            socket.send(new DatagramPacket(data, data.length,
                    destination, collectorPort));
            sent.incrementAndGet();

            // Remember: this "success" does NOT mean it arrived.
            // With UDP, send() succeeds even if the collector is switched off.

        } catch (IOException e) {
            // It almost never happens. And if it does, it does not matter: it is telemetry.
            failed.incrementAndGet();
            LOG.fine("Telemetry datagram not sent: " + e.getMessage());
        }
    }

    @Override
    public void close() {
        if (scheduler != null) {
            scheduler.shutdown();           // two-phase shutdown (08-05)
            try {
                if (!scheduler.awaitTermination(2, TimeUnit.SECONDS)) {
                    scheduler.shutdownNow();
                }
            } catch (InterruptedException e) {
                scheduler.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
        if (socket != null) {
            socket.close();
        }
        LOG.info(() -> "Telemetry stopped: " + sent.get() + " datagrams sent, "
                + failed.get() + " failed");
    }
}

package com.nexussoftware.bibliotech.network;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.TreeMap;

/** Minimal telemetry collector: shows a dashboard on the console. */
public class TelemetryCollector {

    public static void main(String[] args) throws Exception {
        int port = args.length > 0 ? Integer.parseInt(args[0]) : 9098;

        try (DatagramSocket socket = new DatagramSocket(port)) {
            System.out.println("Collector listening on UDP:" + port);

            byte[] buffer = new byte[512];
            DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
            Map<String, String> dashboard = new TreeMap<>();    // sorted by key
            long received = 0;

            while (true) {
                packet.setLength(buffer.length);
                socket.receive(packet);
                received++;

                String measurement = new String(packet.getData(), packet.getOffset(),
                        packet.getLength(), StandardCharsets.UTF_8);

                int colon = measurement.indexOf(':');
                int bar = measurement.indexOf('|');
                if (colon < 0 || bar < 0 || bar < colon) {
                    continue;       // malformed measurement: ignored
                }
                dashboard.put(measurement.substring(0, colon),
                        measurement.substring(colon + 1, bar));

                // We repaint the whole dashboard with every measurement.
                System.out.print("\033[H\033[2J");      // clear the screen
                System.out.println("=== BIBLIOTECH TELEMETRY ===");
                System.out.println("Datagrams received: " + received);
                System.out.println();
                for (Map.Entry<String, String> e : dashboard.entrySet()) {
                    System.out.printf("  %-38s %s%n", e.getKey(), e.getValue());
                }
            }
        }
    }
}

=== BIBLIOTECH TELEMETRY ===
Datagrams received: 145

  bibliotech.connections.active          3
  bibliotech.loans.active                12
  bibliotech.memory.mb                   34
  bibliotech.queries.total               8471
  bibliotech.requests.total              12093

try (CatalogClient client = new CatalogClient("192.168.1.50", 9090)) {
    client.connect();                                       // 3-way handshake + BTCP greeting
    ProtocolCard card = client.query("978-0000000001");     // request + reply
    System.out.println(card.available());
}                                                           // QUIT + 4-message close

ReliableUdpRequest request = new ReliableUdpRequest(
        InetAddress.getByName("192.168.1.50"), 9099, 3, 300);
String reply = request.ask("QUERY 978-0000000001");
if (reply == null) {
    System.out.println("No reply after 3 attempts");
} else {
    System.out.println(reply);
}

package com.nexussoftware.bibliotech.network;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;

/**
 * UDP reflector: returns every datagram as it is, without interpreting it.
 * No state, no threads, no connections. A UDP server in its pure form.
 */
public class UdpReflector {

    public static void main(String[] args) throws IOException {
        int port = args.length > 0 ? Integer.parseInt(args[0]) : 9100;

        try (DatagramSocket socket = new DatagramSocket(port)) {
            // Generous buffer: we accept up to 1500 bytes without truncating.
            byte[] buffer = new byte[1500];
            DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
            long reflected = 0;

            System.out.println("UDP reflector on port " + port);

            while (true) {
                packet.setLength(buffer.length);        // classic mistake 1
                socket.receive(packet);

                // We return EXACTLY the bytes received, not one more:
                // the packet already carries the sender's address and port,
                // so send() returns it to its origin without touching anything.
                socket.send(packet);

                if (++reflected % 1000 == 0) {
                    System.out.println("Reflected: " + reflected);
                }
            }
        }
    }
}

package com.nexussoftware.bibliotech.network;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * UDP probe: measures loss, latency and jitter of a link.
 * It sends and receives in separate threads, without waiting for a reply
 * per packet, which is how a link is really measured.
 */
public class UdpProbe {

    private final String host;
    private final int port;
    private final int total;
    private final int datagramSize;
    private final int perSecond;

    // State shared between the sending thread and the receiving one.
    private final long[] latenciesUs;
    private final Set<Integer> received = new HashSet<>();
    private volatile int duplicates = 0;
    private volatile int outOfOrder = 0;
    private volatile int receivedTotal = 0;

    public UdpProbe(String host, int port, int total,
                    int datagramSize, int perSecond) {
        this.host = host;
        this.port = port;
        this.total = total;
        this.datagramSize = Math.max(datagramSize, 16);     // minimum header
        this.perSecond = perSecond;
        this.latenciesUs = new long[total];
    }

    public void measure() throws IOException, InterruptedException {
        try (DatagramSocket socket = new DatagramSocket()) {
            socket.setSoTimeout(2000);
            InetAddress destination = InetAddress.getByName(host);

            CountDownLatch sendingFinished = new CountDownLatch(1);

            // --- Receiving thread ---
            // It receives in parallel with the sending: that way the measurement
            // does not include the time of "waiting my turn to send the next one".
            Thread receiver = new Thread(() -> receive(socket, sendingFinished),
                    "probe-receiver");
            receiver.start();

            // --- Sender, in this thread ---
            long intervalNs = 1_000_000_000L / perSecond;
            long next = System.nanoTime();

            for (int i = 0; i < total; i++) {
                // Binary header: sequence number + timestamp.
                // ByteBuffer uses big-endian by default: the network order.
                ByteBuffer bb = ByteBuffer.allocate(datagramSize);
                bb.putInt(i);
                bb.putLong(System.nanoTime());
                // The rest of the datagram is padding up to the requested size.

                socket.send(new DatagramPacket(bb.array(), datagramSize,
                        destination, port));

                // Controlled rate: without this we would swamp the receiver's
                // buffer and measure our own saturation, not the network.
                next += intervalNs;
                long waitNs = next - System.nanoTime();
                if (waitNs > 0) {
                    TimeUnit.NANOSECONDS.sleep(waitNs);
                }
            }

            sendingFinished.countDown();
            // Room for the last replies in flight to arrive.
            receiver.join(3000);
            socket.close();     // unblocks the receiver's receive

            report();
        }
    }

    private void receive(DatagramSocket socket, CountDownLatch sendingFinished) {
        byte[] buffer = new byte[1500];
        DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
        int last = -1;

        while (true) {
            try {
                packet.setLength(buffer.length);
                socket.receive(packet);
                long now = System.nanoTime();

                if (packet.getLength() < 12) {
                    continue;       // not even the header fits: ignored
                }

                ByteBuffer bb = ByteBuffer.wrap(packet.getData(),
                        packet.getOffset(), packet.getLength());
                int sequence = bb.getInt();
                long sentAt = bb.getLong();

                if (sequence < 0 || sequence >= total) {
                    continue;       // sequence out of range: somebody else's packet
                }

                synchronized (received) {
                    if (!received.add(sequence)) {
                        duplicates++;
                        continue;
                    }
                }
                if (sequence < last) {
                    outOfOrder++;
                }
                last = Math.max(last, sequence);

                latenciesUs[sequence] = (now - sentAt) / 1000;
                receivedTotal++;

            } catch (SocketTimeoutException e) {
                // Two seconds with nothing. If the sending is over, we stop.
                if (sendingFinished.getCount() == 0) {
                    return;
                }
            } catch (IOException e) {
                return;     // socket closed: the end
            }
        }
    }

    private void report() {
        // Only the latencies of the packets that actually arrived.
        long[] valid = new long[receivedTotal];
        int j = 0;
        for (int i = 0; i < total && j < valid.length; i++) {
            if (latenciesUs[i] > 0) {
                valid[j++] = latenciesUs[i];
            }
        }
        valid = Arrays.copyOf(valid, j);

        if (valid.length == 0) {
            System.out.println("No reply was received.");
            return;
        }

        // Jitter: mean variation between CONSECUTIVE latencies. It is computed
        // BEFORE sorting, because it depends on the time order.
        double jitterSum = 0;
        for (int i = 1; i < valid.length; i++) {
            jitterSum += Math.abs(valid[i] - valid[i - 1]);
        }
        double jitterUs = valid.length > 1 ? jitterSum / (valid.length - 1) : 0;

        long sum = 0;
        for (long v : valid) {
            sum += v;
        }
        long[] sorted = valid.clone();
        Arrays.sort(sorted);

        double loss = 100.0 * (total - receivedTotal) / total;

        System.out.println();
        System.out.println("=== UDP PROBE: " + host + ":" + port + " ===");
        System.out.printf("%-24s %d bytes%n", "Datagram size", datagramSize);
        System.out.printf("%-24s %d/s%n", "Rate", perSecond);
        System.out.println();
        System.out.printf("%-24s %d%n", "Sent", total);
        System.out.printf("%-24s %d%n", "Received", receivedTotal);
        System.out.printf("%-24s %d (%.2f%%)%n", "Lost",
                total - receivedTotal, loss);
        System.out.printf("%-24s %d%n", "Duplicated", duplicates);
        System.out.printf("%-24s %d%n", "Out of order", outOfOrder);
        System.out.println();
        System.out.println("--- LATENCY (ms) ---");
        System.out.printf("%-24s %.3f%n", "Minimum", sorted[0] / 1000.0);
        System.out.printf("%-24s %.3f%n", "Mean",
                (sum / (double) valid.length) / 1000.0);
        System.out.printf("%-24s %.3f%n", "Median",
                sorted[sorted.length / 2] / 1000.0);
        System.out.printf("%-24s %.3f%n", "p95",
                sorted[(int) (sorted.length * 0.95)] / 1000.0);
        System.out.printf("%-24s %.3f%n", "Maximum",
                sorted[sorted.length - 1] / 1000.0);
        System.out.printf("%-24s %.3f%n", "Mean jitter", jitterUs / 1000.0);
    }

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

        for (int size : new int[]{64, 512, 1400}) {
            new UdpProbe(host, port, 1000, size, 500).measure();
            Thread.sleep(500);
        }
    }
}

=== UDP PROBE: localhost:9100 ===
Datagram size            64 bytes
Rate                     500/s

Sent                     1000
Received                 1000
Lost                     0 (0.00%)
Duplicated               0
Out of order             0

--- LATENCY (ms) ---
Minimum                  0.041
Mean                     0.078
Median                   0.069
p95                      0.142
Maximum                  1.884
Mean jitter              0.031

package com.nexussoftware.bibliotech.network;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MulticastSocket;
import java.net.NetworkInterface;
import java.net.StandardSocketOptions;
import java.nio.charset.StandardCharsets;
import java.util.Enumeration;
import java.util.logging.Logger;

/**
 * Nexus Software's internal chat over multicast. With no central server:
 * everybody is equal, everybody publishes to the group and receives from it.
 */
public class BiblioTechChat implements AutoCloseable {

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

    /** Range 239.x.x.x: administrative scope, private use. */
    private static final String GROUP = "239.10.10.20";
    private static final int PORT = 9099;
    private static final int MAX_DATAGRAM = 1400;
    private static final int MAX_MESSAGE = 500;

    private final String user;
    private final long startup = System.currentTimeMillis();

    private MulticastSocket socket;
    private InetSocketAddress group;
    private NetworkInterface networkInterface;
    private volatile boolean running = false;

    public BiblioTechChat(String user) {
        this.user = user;
    }

    public void start() throws IOException {
        socket = new MulticastSocket(PORT);
        socket.setTimeToLive(1);        // it does not leave the local network

        // FILTERING OF OUR OWN MESSAGES - the option chosen.
        //
        // There are two ways:
        //   (a) IP_MULTICAST_LOOP = false: the system does not hand our own
        //       sends back to us. It is clean and costs nothing, BUT it also
        //       stops TWO INSTANCES ON THE SAME MACHINE seeing each other,
        //       because the filter is per machine, not per socket.
        //   (b) Comparing the sender against the local addresses and filtering
        //       in the code. More work, but it allows testing with three
        //       instances on the same computer.
        //
        // We choose (b) because the exercise asks to be able to test with three
        // local instances, and besides it is the behaviour you want in a
        // chat: if I open two windows, I want to see both.
        socket.setOption(StandardSocketOptions.IP_MULTICAST_LOOP, true);

        group = new InetSocketAddress(InetAddress.getByName(GROUP), PORT);
        networkInterface = chooseInterface();

        // Modern API (Java 14+): with an explicit NetworkInterface.
        socket.joinGroup(group, networkInterface);
        running = true;

        LOG.info(() -> "Joined " + GROUP + ":" + PORT
                + " through " + networkInterface.getName());
    }

    private NetworkInterface chooseInterface() throws IOException {
        Enumeration<NetworkInterface> nis = NetworkInterface.getNetworkInterfaces();
        NetworkInterface fallback = null;
        while (nis.hasMoreElements()) {
            NetworkInterface ni = nis.nextElement();
            if (!ni.isUp() || !ni.supportsMulticast()) {
                continue;
            }
            if (!ni.isLoopback()) {
                return ni;      // we prefer a real interface
            }
            fallback = ni;      // loopback is useful for local testing
        }
        if (fallback == null) {
            throw new IOException("There is no interface with multicast");
        }
        return fallback;
    }

    /** Receive loop. It runs in its own thread. */
    private void receive() {
        byte[] buffer = new byte[MAX_DATAGRAM];
        DatagramPacket packet = new DatagramPacket(buffer, buffer.length);

        while (running) {
            try {
                packet.setLength(buffer.length);        // classic mistake 1
                socket.receive(packet);

                // Filter (b): we discard what we sent ourselves.
                if (isOurs(packet.getAddress(), packet.getPort())) {
                    continue;
                }

                String message = new String(packet.getData(), packet.getOffset(),
                        packet.getLength(), StandardCharsets.UTF_8);   // classic mistake 2

                // EVERYTHING arriving through the group comes from a stranger.
                if (message.length() > MAX_MESSAGE) {
                    LOG.warning("Message too long, discarded");
                    continue;
                }
                System.out.println(sanitise(message));

            } catch (IOException e) {
                if (!running) {
                    return;     // orderly close
                }
                LOG.warning("Error receiving: " + e.getMessage());
            }
        }
    }

    /**
     * A packet is ours if it comes from a local address AND from our
     * source port. Comparing the address alone would not do: it would also
     * filter out the messages of the other instances on the same machine.
     */
    private boolean isOurs(InetAddress sender, int senderPort) {
        if (senderPort != socket.getLocalPort()) {
            return false;
        }
        try {
            Enumeration<InetAddress> own = networkInterface.getInetAddresses();
            while (own.hasMoreElements()) {
                if (own.nextElement().equals(sender)) {
                    return true;
                }
            }
        } catch (Exception e) {
            // If we cannot check it, we would rather show too much than too little.
            return false;
        }
        return false;
    }

    public void send(String text) throws IOException {
        long mark = (System.currentTimeMillis() - startup) / 1000;
        String message = String.format("[%3ds] %-14s %s", mark, user, text);

        byte[] data = message.getBytes(StandardCharsets.UTF_8);
        if (data.length > MAX_DATAGRAM) {
            System.out.println("(message too long, not sent)");
            return;
        }
        socket.send(new DatagramPacket(data, data.length,
                InetAddress.getByName(GROUP), PORT));
    }

    /**
     * Sanitises before printing to the console: a message from the group may
     * contain ANSI escape sequences that manipulate the terminal
     * of whoever reads it (clearing the screen, moving the cursor, colours).
     */
    private String sanitise(String text) {
        StringBuilder sb = new StringBuilder(text.length());
        for (int i = 0; i < text.length(); i++) {
            char c = text.charAt(i);
            sb.append(Character.isISOControl(c) && c != '\t' ? '?' : c);
        }
        return sb.toString();
    }

    @Override
    public void close() {
        if (!running) {
            return;
        }
        try {
            send("*** has left ***");
        } catch (IOException ignored) {
            // We are leaving anyway.
        }
        running = false;
        try {
            socket.leaveGroup(group, networkInterface);
        } catch (IOException e) {
            LOG.fine("Failure leaving the group: " + e.getMessage());
        }
        socket.close();     // unblocks the receive()
    }

    public static void main(String[] args) throws IOException {
        String user = args.length > 0 ? args[0] : "Anonymous";

        BiblioTechChat chat = new BiblioTechChat(user);
        chat.start();

        Thread receiver = new Thread(chat::receive, "chat-receiver");
        receiver.setDaemon(true);
        receiver.start();

        Runtime.getRuntime().addShutdownHook(new Thread(chat::close));

        chat.send("*** has joined ***");
        System.out.println("BiblioTech chat. Type /quit to finish.");

        try (BufferedReader keyboard = new BufferedReader(
                new InputStreamReader(System.in, StandardCharsets.UTF_8))) {
            String line;
            while ((line = keyboard.readLine()) != null) {
                if (line.equals("/quit")) {
                    break;
                }
                if (!line.isBlank()) {
                    chat.send(line);
                }
            }
        }
        chat.close();
    }
}

java -cp classes com.nexussoftware.bibliotech.network.BiblioTechChat "Marta Ruiz"
java -cp classes com.nexussoftware.bibliotech.network.BiblioTechChat "Diego Alonso"
java -cp classes com.nexussoftware.bibliotech.network.BiblioTechChat "Nuria Vidal"

Marta's terminal:
BiblioTech chat. Type /quit to finish.
[  2s] Diego Alonso   *** has joined ***
[  5s] Nuria Vidal    *** has joined ***
Anyone got Effective Java?
[ 12s] Diego Alonso   I have it, returning it tomorrow
[ 18s] Nuria Vidal    *** has left ***

package com.nexussoftware.bibliotech.network;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.concurrent.ThreadLocalRandom;
import java.util.logging.Logger;

/**
 * Reliable file transfer over UDP with stop and wait.
 *
 * Format of each data datagram:
 *   [int    ] sequence number
 *   [byte   ] 1 if it is the last block, 0 if not
 *   [int    ] length of the data
 *   [bytes  ] data
 *
 * Format of the acknowledgement:
 *   [int    ] sequence number acknowledged
 */
public class ReliableUdpSender {

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

    private static final int DATA_PER_BLOCK = 1024;
    private static final int HEADER = 4 + 1 + 4;        // int + byte + int
    private static final int MAX_ATTEMPTS = 5;
    private static final int INITIAL_WAIT_MS = 200;

    /** Percentage of datagrams discarded on purpose, for testing. */
    private final int simulatedLoss;

    private int retransmissions = 0;

    public ReliableUdpSender(int simulatedLoss) {
        this.simulatedLoss = simulatedLoss;
    }

    public void send(String host, int port, Path file) throws IOException {
        long size = Files.size(file);
        long start = System.currentTimeMillis();

        try (DatagramSocket socket = new DatagramSocket();
             InputStream input = Files.newInputStream(file)) {

            InetAddress destination = InetAddress.getByName(host);
            // connect() filters: we only accept acknowledgements from this destination.
            socket.connect(destination, port);

            byte[] block = new byte[DATA_PER_BLOCK];
            byte[] ackBuffer = new byte[16];
            DatagramPacket ack = new DatagramPacket(ackBuffer, ackBuffer.length);

            int sequence = 0;
            long sentBytes = 0;
            long checksum = 0;

            while (true) {
                int read = input.read(block);
                boolean last = false;

                if (read == -1) {
                    // End of file: we send an empty block marked
                    // as the last, so the receiver knows it has finished.
                    read = 0;
                    last = true;
                } else {
                    // We check whether anything is left without consuming it:
                    // available() is only indicative, so we mark the
                    // last one with a final empty block. Simpler and safer.
                    for (int i = 0; i < read; i++) {
                        checksum += block[i] & 0xFF;
                    }
                    sentBytes += read;
                }

                byte[] datagram = build(sequence, last, block, read);

                if (!sendWithAck(socket, destination, port,
                        datagram, sequence, ack, ackBuffer)) {
                    throw new IOException("The receiver does not acknowledge block "
                            + sequence + " after " + MAX_ATTEMPTS + " attempts");
                }

                if (last) {
                    break;
                }
                sequence++;
            }

            long ms = System.currentTimeMillis() - start;
            System.out.println();
            System.out.println("=== TRANSFER COMPLETED ===");
            System.out.printf("%-26s %s%n", "File", file.getFileName());
            System.out.printf("%-26s %d%n", "Bytes", sentBytes);
            System.out.printf("%-26s %d%n", "Blocks", sequence + 1);
            System.out.printf("%-26s %d%n", "Retransmissions", retransmissions);
            System.out.printf("%-26s %d%%%n", "Simulated loss", simulatedLoss);
            System.out.printf("%-26s %d ms%n", "Time", ms);
            System.out.printf("%-26s %d%n", "Checksum", checksum);
            System.out.printf("%-26s %.1f KB/s%n", "Speed",
                    ms == 0 ? 0 : sentBytes / 1024.0 / (ms / 1000.0));
        }
    }

    /** Builds the datagram with its binary header in big-endian. */
    private byte[] build(int sequence, boolean last, byte[] data, int length)
            throws IOException {
        ByteArrayOutputStream bytes = new ByteArrayOutputStream(HEADER + length);
        DataOutputStream output = new DataOutputStream(bytes);
        output.writeInt(sequence);
        output.writeBoolean(last);
        output.writeInt(length);
        output.write(data, 0, length);
        output.flush();
        return bytes.toByteArray();
    }

    /** Sends and waits for the acknowledgement, retrying with increasing backoff. */
    private boolean sendWithAck(DatagramSocket socket, InetAddress destination,
                                int port, byte[] datagram, int sequence,
                                DatagramPacket ack, byte[] ackBuffer)
            throws IOException {

        int wait = INITIAL_WAIT_MS;

        for (int attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
            if (attempt > 1) {
                retransmissions++;
            }

            // --- SIMULATED LOSS ---
            // The datagram is discarded BEFORE sending it, to prove that
            // the retry mechanism works without needing a bad network.
            if (ThreadLocalRandom.current().nextInt(100) >= simulatedLoss) {
                socket.send(new DatagramPacket(datagram, datagram.length,
                        destination, port));
            } else {
                LOG.fine("Simulated loss of block " + sequence);
            }

            socket.setSoTimeout(wait);
            long limit = System.currentTimeMillis() + wait;

            while (System.currentTimeMillis() < limit) {
                try {
                    ack.setLength(ackBuffer.length);        // classic mistake 1
                    socket.receive(ack);

                    if (ack.getLength() < 4) {
                        continue;
                    }
                    int acknowledged = ByteBuffer.wrap(ack.getData(),
                            ack.getOffset(), ack.getLength()).getInt();

                    if (acknowledged == sequence) {
                        return true;
                    }
                    // Acknowledgement of an earlier block (it arrived late):
                    // discarded and we carry on waiting for ours.
                    LOG.fine("Stale acknowledgement: " + acknowledged
                            + ", we expected " + sequence);

                } catch (SocketTimeoutException e) {
                    break;      // deadline expired: next attempt
                }
            }
            wait *= 2;          // increasing backoff
        }
        return false;
    }

    public static void main(String[] args) throws IOException {
        Path file = Path.of(args.length > 0 ? args[0] : "catalog.csv");
        int loss = args.length > 1 ? Integer.parseInt(args[1]) : 20;
        new ReliableUdpSender(loss).send("localhost", 9101, file);
    }
}

package com.nexussoftware.bibliotech.network;

import java.io.IOException;
import java.io.OutputStream;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.logging.Logger;

/** Receiver of the reliable transfer over UDP. */
public class ReliableUdpReceiver {

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

    private static final int DATA_PER_BLOCK = 1024;
    private static final int HEADER = 4 + 1 + 4;
    private static final int MAX_DATAGRAM = HEADER + DATA_PER_BLOCK;

    public void receive(int port, Path target) throws IOException {
        Files.createDirectories(target.getParent() == null
                ? Path.of(".") : target.getParent());

        try (DatagramSocket socket = new DatagramSocket(port);
             OutputStream output = Files.newOutputStream(target)) {

            System.out.println("Waiting for a file on UDP:" + port);
            socket.setSoTimeout(30_000);        // we give up if nobody sends

            byte[] buffer = new byte[MAX_DATAGRAM + 1];
            DatagramPacket packet = new DatagramPacket(buffer, buffer.length);

            int expected = 0;
            int duplicates = 0;
            long bytes = 0;
            long checksum = 0;
            long start = System.currentTimeMillis();

            while (true) {
                packet.setLength(buffer.length);        // classic mistake 1
                socket.receive(packet);

                if (packet.getLength() < HEADER
                        || packet.getLength() > MAX_DATAGRAM) {
                    LOG.warning("Datagram of invalid size discarded");
                    continue;
                }

                ByteBuffer bb = ByteBuffer.wrap(packet.getData(),
                        packet.getOffset(), packet.getLength());
                int sequence = bb.getInt();
                boolean last = bb.get() != 0;
                int length = bb.getInt();

                // VALIDATION: the length comes from the network. Without this,
                // a malicious sender causes an exception or something worse.
                if (length < 0 || length > DATA_PER_BLOCK
                        || length > bb.remaining()) {
                    LOG.warning("Invalid declared length: " + length);
                    continue;
                }

                if (sequence < expected) {
                    // DUPLICATE: our previous acknowledgement was lost and
                    // the sender resent. We must RE-ACKNOWLEDGE but NOT write
                    // the data again, or the file would come out corrupt.
                    duplicates++;
                    acknowledge(socket, packet, sequence);
                    continue;
                }
                if (sequence > expected) {
                    // With stop and wait this should not happen: it means
                    // we lost an intermediate block. We do not acknowledge, and the
                    // sender will resend the missing one.
                    LOG.warning("Block " + sequence + " out of order; "
                            + "we expected " + expected);
                    continue;
                }

                // Correct block, in order: it is written.
                byte[] data = new byte[length];
                bb.get(data);
                output.write(data);
                for (byte b : data) {
                    checksum += b & 0xFF;
                }
                bytes += length;

                acknowledge(socket, packet, sequence);
                expected++;

                if (last) {
                    output.flush();
                    long ms = System.currentTimeMillis() - start;
                    System.out.println();
                    System.out.println("=== RECEPTION COMPLETED ===");
                    System.out.printf("%-26s %s%n", "Saved to", target);
                    System.out.printf("%-26s %d%n", "Bytes", bytes);
                    System.out.printf("%-26s %d%n", "Blocks", expected);
                    System.out.printf("%-26s %d%n", "Duplicates detected", duplicates);
                    System.out.printf("%-26s %d ms%n", "Time", ms);
                    System.out.printf("%-26s %d%n", "Checksum",
                            checksum);
                    return;
                }
            }
        }
    }

    private void acknowledge(DatagramSocket socket, DatagramPacket original, int sequence)
            throws IOException {
        byte[] ack = ByteBuffer.allocate(4).putInt(sequence).array();
        // We reply TO THE SENDER, which comes in the original packet.
        socket.send(new DatagramPacket(ack, ack.length,
                original.getAddress(), original.getPort()));
    }

    public static void main(String[] args) throws IOException {
        new ReliableUdpReceiver().receive(9101, Path.of("received", "catalog.csv"));
    }
}

Terminal 1 (receiver):
Waiting for a file on UDP:9101

=== RECEPTION COMPLETED ===
Saved to                   received/catalog.csv
Bytes                      2847
Blocks                     4
Duplicates detected        1
Time                       834 ms
Checksum                   291476

Terminal 2 (sender):
=== TRANSFER COMPLETED ===
File                       catalog.csv
Bytes                      2847
Blocks                     4
Retransmissions            3
Simulated loss             20%
Time                       841 ms
Checksum                   291476

try (Socket socket = new Socket()) {
    socket.connect(new InetSocketAddress(host, port), 3000);
    Files.copy(file, socket.getOutputStream());
    socket.shutdownOutput();
}