uggly reliable

ChatApp/app/src/main/java/ChatApp.java

@@ -8,14 +8,16 @@ public class ChatApp {
         Server server = new Server(6665);
         ClientConsole client = new ClientConsole(new InetSocketAddress("localhost", 6665));
 
-        client.getClientInterface().SendCreateRoom("101");
+        client.onConnect.connect(() -> {
+            client.getClientInterface().SendCreateRoom("101");
+        });
 
-        client.joinThread();
+        client.onDisconnect.connect(() -> {
+            System.out.println("Stopping server ...");
     
-        System.out.println("Stopping server ...");
+            server.close();
     
-        server.close();
+            System.out.println("Done !");
-
+        });
-        System.out.println("Done !");
     }
 }

ChatApp/app/src/main/java/client/Client.java

@@ -17,16 +17,16 @@ public class Client {
 		login(pseudo);
 	}
 
+	private void login(String pseudo) {
+		this.connexion.sendPacket(new LoginPacket(pseudo));
+	}
+
 	public void close() {
 		this.connexion.sendPacket(new DisconnectPacket("Leaving"));
 		this.connexion.close();
 		this.callback.handleDisconnect();
 	}
 
-	private void login(String pseudo) {
-		this.connexion.sendPacket(new LoginPacket(pseudo));
-	}
-
 	public void SendChatMessage(String message) {
 		this.connexion.sendPacket(new SendChatMessagePacket(message));
 	}

ChatApp/app/src/main/java/client/ClientConnexion.java

@@ -8,40 +8,19 @@ import network.Socket;
 import network.protocol.Packet;
 import network.protocol.PacketVisitor;
 import network.protocol.packets.*;
+import network.protocol.packets.ServerResponsePacket.Response;
 
 public class ClientConnexion implements PacketVisitor, PacketHandler {
 
 	private final InetSocketAddress serverAddress;
 	private final Socket socket;
 	private final ClientListener callback;
-	private volatile boolean connected = false;
 
 	public ClientConnexion(Socket socket, InetSocketAddress serverAddress, ClientListener callback) {
 		this.serverAddress = serverAddress;
 		this.socket = socket;
 		this.callback = callback;
 		this.socket.addHandler(this);
-		spamHandshake();
-	}
-
-	private void spamHandshake() {
-		for (int i = 0; i < 5; i++) {
-			sendPacket(new HandshakePacket());
-		}
-		new Thread(this::waitForHandshake).start();
-	}
-
-	private void waitForHandshake() {
-		try {
-			Thread.sleep(1000);
-		} catch (InterruptedException e) {
-			e.printStackTrace();
-		}
-		if(!connected) {
-			System.out.println("The server did not respond !");
-			this.close();
-			this.callback.handleConnexionError();
-		}
 	}
 
 	public void close() {
@@ -77,6 +56,8 @@ public class ClientConnexion implements PacketVisitor, PacketHandler {
 	@Override
 	public void visitPacket(ServerResponsePacket packet) {
 		this.callback.handleServerResponse(packet.getResponse());
+		if (packet.getResponse() == Response.AuthSuccess)
+			this.callback.handleConnect();
 	}
 
 	@Override
@@ -92,17 +73,9 @@ public class ClientConnexion implements PacketVisitor, PacketHandler {
 	@Override
 	public void visitPacket(DisconnectPacket packet) {
 		this.close();
-		this.connected = false;
 		this.callback.handleDisconnect();
 	}
 
-	@Override
-	public void visitPacket(HandshakePacket packet) {
-		if (!connected)
-			this.callback.handleConnect();
-		this.connected = true;
-	}
-
 	@Override
 	public void visitPacket(CreateRoomPacket packet) {
 		throw new UnsupportedOperationException("Unimplemented method 'visitPacket'");

ChatApp/app/src/main/java/client/ClientConsole.java

@@ -11,13 +11,16 @@ import java.util.function.Consumer;
 import network.protocol.ANSIColor;
 import network.protocol.packets.ServerResponsePacket;
 import network.protocol.packets.ServerResponsePacket.Response;
+import utilities.Signal;
 
 public class ClientConsole implements ClientListener {
 
 	private Client client;
 	private final Thread inputThread;
 	private final Scanner scanner;
-	private volatile boolean connected = false;
+
+	public final Signal onConnect = new Signal();
+	public final Signal onDisconnect = new Signal();
 
 	public ClientConsole(InetSocketAddress address) {
 		this.inputThread = new Thread(this::inputLoop);
@@ -43,18 +46,6 @@ public class ClientConsole implements ClientListener {
 	}
 
 	private void inputLoop() {
-		// waiting to be connected
-		try {
-			Thread.sleep(2000);
-			return;
-		} catch (InterruptedException e) {
-			if (!connected)
-				return;
-		}
-
-		// resets the interrupt
-		Thread.interrupted();
-
 		while (!Thread.interrupted()) {
 			String message = this.scanner.nextLine();
 			if (Thread.interrupted())
@@ -148,7 +139,6 @@ public class ClientConsole implements ClientListener {
 	@Override
 	public void handleConnexionError() {
 		System.out.println("An error occured during the connexion !");
-		this.connected = false;
 		stop();
 	}
 
@@ -190,8 +180,7 @@ public class ClientConsole implements ClientListener {
 	@Override
 	public void handleConnect() {
 		System.out.println("Connected to server !");
-		this.connected = true;
+		this.onConnect.emit();
-		this.inputThread.interrupt();
 	}
 
 }

ChatApp/app/src/main/java/network/PacketPool.java

@@ -0,0 +1,253 @@
+package network;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.Stack;
+
+import network.protocol.Packet;
+import network.protocol.packets.AcknowlegdementPacket;
+import network.protocol.packets.DisconnectPacket;
+
+public class PacketPool {
+
+	private final Stack<ReliablePacketAddress> packetQueue;
+	private final Map<InetSocketAddress, AdressContext> addressContexts;
+	
+	private final Socket socket;
+
+	private static int MAX_SEND_TRY = 50;
+	private static long RETRY_INTERVAL = 100;
+	private static float PACKET_LOSS_PROBABILITY = 0.75f;
+
+	private static record ReliablePacketAddress(ReliablePacket packet, InetSocketAddress address) {
+		@Override
+		public final int hashCode() {
+			return Objects.hash(packet, address);
+		}
+
+		@Override
+		public final boolean equals(Object arg0) {
+			if (arg0 instanceof ReliablePacketAddress packetAddress)
+				return packetAddress.address().equals(this.address())
+						&& packetAddress.packet().getSeq() == this.packet().getSeq();
+			return false;
+		}
+	}
+
+	private static class AdressContext {
+		public int currentSeqRecv = 0;
+		public int currentSeqSend = 0;
+		public final List<Thread> sendThreads = new ArrayList<>();
+		public final List<ReliablePacket> packetRecvBuffer = new ArrayList<>();
+		public final Map<ReliablePacket, Integer> packetsSentTries = new HashMap<>();
+	}
+
+	private int getNextSeqSend(InetSocketAddress address) {
+		return this.addressContexts.get(address).currentSeqSend++;
+	}
+
+	private int getCurrentSeqRecv(InetSocketAddress address) {
+		return this.addressContexts.get(address).currentSeqRecv;
+	}
+
+	private void setSeqRecv(InetSocketAddress address, int newValue) {
+		this.addressContexts.get(address).currentSeqRecv = newValue;
+	}
+
+	private void tryAddContext(InetSocketAddress address) {
+		this.addressContexts.putIfAbsent(address, new AdressContext());
+	}
+
+	public PacketPool(Socket socket) {
+		this.socket = socket;
+		this.packetQueue = new Stack<>();
+		this.addressContexts = new HashMap<>();
+	}
+
+	private void debugPrint(String msg) {
+		// System.out.println(msg);
+	}
+
+	private void debugSend(ReliablePacket packet, InetSocketAddress address) {
+		boolean client = address.getPort() == 6665;
+		debugPrint((client ? "[Client]" : "[Server]") + " Sent " + packet.getPacket().getClass().getName()
+				+ " with seq : " + packet.getSeq() + " and ack " + packet.getAck());
+	}
+
+	private void debugRecv(ReliablePacket packet, InetSocketAddress address) {
+		boolean client = address.getPort() == 6665;
+		debugPrint((client ? "[Client]" : "[Server]") + " Received " + packet.getPacket().getClass().getName()
+				+ " with seq : " + packet.getSeq() + " and ack " + packet.getAck());
+	}
+
+	private void sendPacketToSocket(ReliablePacket packet, InetSocketAddress address) throws IOException {
+		var packetsSentTries = this.addressContexts.get(address).packetsSentTries;
+		
+		if (Math.random() > PACKET_LOSS_PROBABILITY)
+			this.socket.sendPacket(packet, address);
+
+		if (packet.getPacket() instanceof AcknowlegdementPacket)
+			return;
+
+		Integer count = packetsSentTries.get(packet);
+		if (count == null) {
+			packetsSentTries.put(packet, 1);
+		} else {
+			packetsSentTries.put(packet, count + 1);
+		}
+	}
+
+	private synchronized void sendPacket(ReliablePacket packet, InetSocketAddress address) throws IOException {
+		sendPacketToSocket(packet, address);
+		debugSend(packet, address);
+
+		ReliablePacketAddress reliablePacketAddress = new ReliablePacketAddress(packet, address);
+
+		if (!(packet.getPacket() instanceof AcknowlegdementPacket)) {
+			Thread newThread = new Thread(() -> tryResend(reliablePacketAddress));
+			this.addressContexts.get(address).sendThreads.add(newThread);
+			newThread.start();
+		}
+	}
+
+	public synchronized void sendPacket(Packet packet, InetSocketAddress address) throws IOException {
+		tryAddContext(address);
+		ReliablePacket reliablePacket = new ReliablePacket(packet, getNextSeqSend(address), -1);
+		sendPacket(reliablePacket, address);
+	}
+
+	private void tryResend(ReliablePacketAddress reliablePacketAddress) {
+		try {
+			while (!Thread.interrupted()) {
+				Thread.sleep(RETRY_INTERVAL);
+				var packetsSentTries = this.addressContexts.get(reliablePacketAddress.address()).packetsSentTries;
+				// the packet has been received
+				if (!packetsSentTries.containsKey(reliablePacketAddress.packet()))
+					break;
+				
+				Integer sendCount = packetsSentTries.get(reliablePacketAddress.packet());
+				if (sendCount > MAX_SEND_TRY) {
+					close(reliablePacketAddress.address());
+					debugPrint("Packet" + reliablePacketAddress.packet() + " not send after " + MAX_SEND_TRY + " tries");
+					// simulating a fake disconnect packet
+					this.socket.handlePacket(new DisconnectPacket("Timed out"), reliablePacketAddress.address());
+					break;
+				}
+
+				boolean client = reliablePacketAddress.address().getPort() == 6665;
+				debugPrint((client ? "[Client]" : "[Server]") + " Trying to resend the packet "
+						+ reliablePacketAddress.packet().getSeq() + " ...");
+				sendPacketToSocket(reliablePacketAddress.packet(), reliablePacketAddress.address());
+			}
+		} catch (InterruptedException e) {
+			// e.printStackTrace();
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+
+		AdressContext ctx = this.addressContexts.get(reliablePacketAddress.address());
+		if (ctx != null)
+			ctx.sendThreads.remove(Thread.currentThread());
+	}
+
+	private ReliablePacket getMinimumSeqReceived(InetSocketAddress address) {
+		List<ReliablePacket> packetRecvBuffer = this.addressContexts.get(address).packetRecvBuffer;
+		if (packetRecvBuffer.isEmpty())
+			return null;
+		return Collections.min(packetRecvBuffer, (rel1, rel2) -> {return Integer.compare(rel1.getSeq(), rel2.getSeq());});
+	}
+
+	private int fillPacketQueue(InetSocketAddress address) {
+		List<ReliablePacket> packetRecvBuffer = this.addressContexts.get(address).packetRecvBuffer;
+		ReliablePacket minimum = getMinimumSeqReceived(address);
+		int lastSeqProcessed = -1;
+		while (true) {
+			this.packetQueue.add(new ReliablePacketAddress(minimum, address));
+			packetRecvBuffer.remove(minimum);
+			lastSeqProcessed = minimum.getSeq();
+
+			ReliablePacket nextMinimum = getMinimumSeqReceived(address);
+			if (nextMinimum == null || nextMinimum.getSeq() != minimum.getSeq() + 1)
+				break;
+			minimum = nextMinimum;
+		}
+		Collections.reverse(this.packetQueue);
+		return lastSeqProcessed;
+	}
+
+	public void onPacketReceived(ReliablePacket packet, InetSocketAddress address) throws IOException {
+		tryAddContext(address);
+
+		var packetsSentTries = this.addressContexts.get(address).packetsSentTries;
+
+		if (packet.getPacket() instanceof AcknowlegdementPacket) {
+			assert (packet.getAck() != -1);
+			for (var entry : packetsSentTries.entrySet()) {
+				ReliablePacket reliablePacket = entry.getKey();
+				if (entry.getKey().getSeq() == packet.getAck()) {
+					packetsSentTries.remove(reliablePacket);
+					break;
+				}
+			}
+			return;
+		}
+
+		if (this.addressContexts.get(address).packetRecvBuffer.contains(packet)) {
+			debugPrint("The packet has already been received !");
+			sendPacketToSocket(
+					new ReliablePacket(new AcknowlegdementPacket(), -1, packet.getSeq()), address);
+			return;
+		}
+
+		if (packet.getSeq() < getCurrentSeqRecv(address)) {
+			debugPrint("Packet too old, current : " + getCurrentSeqRecv(address));
+			sendPacketToSocket(
+					new ReliablePacket(new AcknowlegdementPacket(), -1, packet.getSeq()), address);
+			return;
+		}
+
+		this.addressContexts.get(address).packetRecvBuffer.add(packet);
+		debugRecv(packet, address);
+		sendPacketToSocket(
+				new ReliablePacket(new AcknowlegdementPacket(), -1, packet.getSeq()), address);
+
+		// got the packet in the right order
+		if (packet.getSeq() == getCurrentSeqRecv(address)) {
+			setSeqRecv(address, fillPacketQueue(address) + 1);
+		}
+
+	}
+
+	public Entry<Packet, InetSocketAddress> getNextPacket() {
+		if (this.packetQueue.isEmpty())
+			return null;
+		ReliablePacketAddress last = this.packetQueue.pop();
+		var entry = Map.entry(last.packet().getPacket(), last.address);
+		return entry;
+	}
+
+	private void close(InetSocketAddress adress) {
+		var ctx = this.addressContexts.get(adress);
+		if (ctx != null)
+			close(ctx);
+	}
+
+	private void close(AdressContext adressContext) {
+		for (Thread thread : adressContext.sendThreads) {
+			thread.interrupt();
+		}
+	}
+
+	public void close() {
+		for (AdressContext adressContext : this.addressContexts.values()) {
+			close(adressContext);
+		}
+	}
+}

ChatApp/app/src/main/java/network/ReliablePacket.java

@@ -0,0 +1,38 @@
+package network;
+
+import java.io.Serializable;
+
+import network.protocol.Packet;
+
+public class ReliablePacket implements Serializable {
+
+	private final Packet packet;
+	private final int seq;
+	private final int ack;
+
+	public ReliablePacket(Packet packet, int seq, int ack) {
+		this.packet = packet;
+		this.seq = seq;
+		this.ack = ack;
+	}
+
+	public Packet getPacket() {
+		return packet;
+	}
+
+	public int getAck() {
+		return ack;
+	}
+
+	public int getSeq() {
+		return seq;
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		if (obj instanceof ReliablePacket packet)
+			return packet.getSeq() == this.getSeq();
+		return false;
+	}
+
+}

ChatApp/app/src/main/java/network/Socket.java

@@ -11,8 +11,6 @@ import java.net.InetSocketAddress;
 import java.net.SocketException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
 
 import network.protocol.Packet;
 
@@ -20,10 +18,12 @@ public class Socket {
 	private final DatagramSocket socket;
 	private final List<PacketHandler> handlers;
 	private final Thread readThread;
+	private final PacketPool packetPool;
 
 	public Socket() throws SocketException {
 		this.socket = new DatagramSocket();
 		this.handlers = new ArrayList<>();
+		this.packetPool = new PacketPool(this);
 		this.readThread = new Thread(this::readLoop);
 		this.readThread.start();
 	}
@@ -31,6 +31,7 @@ public class Socket {
 	public Socket(int listeningPort) throws SocketException {
 		this.socket = new DatagramSocket(listeningPort);
 		this.handlers = new ArrayList<>();
+		this.packetPool = new PacketPool(this);
 		this.readThread = new Thread(this::readLoop);
 		this.readThread.start();
 	}
@@ -43,7 +44,8 @@ public class Socket {
 		return this.socket.getLocalPort();
 	}
 
-	public void sendPacket(Packet packet, InetSocketAddress address) throws IOException {
+	// needs to be accessible by PacketPool
+	void sendPacket(ReliablePacket packet, InetSocketAddress address) throws IOException {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
 		ObjectOutputStream oos = new ObjectOutputStream(stream);
 		oos.writeObject(packet);
@@ -54,7 +56,8 @@ public class Socket {
 		this.socket.send(dataPacket);
 	}
 
-	public Entry<Packet, InetSocketAddress> recievePacket() throws IOException, ClassNotFoundException {
+	private void recievePacketReliable()
+			throws IOException, ClassNotFoundException {
 		byte[] buffer = new byte[65535];
 		DatagramPacket dataPacket = new DatagramPacket(buffer, buffer.length);
 		socket.receive(dataPacket);
@@ -62,16 +65,30 @@ public class Socket {
 		InetSocketAddress address = new InetSocketAddress(dataPacket.getAddress(), dataPacket.getPort());
 
 		ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(dataPacket.getData()));
-		Packet packet = (Packet) ois.readObject();
+		ReliablePacket packet = (ReliablePacket) ois.readObject();
 
-		return Map.entry(packet, address);
+		this.packetPool.onPacketReceived(packet, address);
+	}
+
+	public void sendPacket(Packet packet, InetSocketAddress address) throws IOException {
+		this.packetPool.sendPacket(packet, address);
+	}
+
+	private void recievePacket() throws IOException, ClassNotFoundException {
+		this.recievePacketReliable();
+		var entry = this.packetPool.getNextPacket();
+		while (entry != null) {
+			this.handlePacket(entry.getKey(), entry.getValue());
+			entry = this.packetPool.getNextPacket();
+		}
 	}
 
 	public void close() {
 		this.socket.close();
+		this.packetPool.close();
 	}
 
-	private void handlePacket(Packet packet, InetSocketAddress address) {
+	void handlePacket(Packet packet, InetSocketAddress address) {
 		for (PacketHandler handler : this.handlers) {
 			handler.handlePacket(packet, address);
 		}
@@ -80,8 +97,7 @@ public class Socket {
 	private void readLoop() {
 		try {
 			while (!Thread.interrupted()) {
-				var entry = this.recievePacket();
+				recievePacket();
-				this.handlePacket(entry.getKey(), entry.getValue());
 			}
 		} catch (IOException | ClassNotFoundException e) {
 			e.printStackTrace();

ChatApp/app/src/main/java/network/protocol/PacketVisitor.java

@@ -11,7 +11,6 @@ public interface PacketVisitor {
 	void visitPacket(ChatMessagePacket packet);
 	void visitPacket(CreateRoomPacket packet);
 	void visitPacket(DisconnectPacket packet);
-	void visitPacket(HandshakePacket packet);
 	void visitPacket(JoinRoomPacket packet);
 	void visitPacket(LeaveRoomPacket packet);
 	void visitPacket(LoginPacket packet);

ChatApp/app/src/main/java/network/protocol/packets/AcknowlegdementPacket.java

@@ -0,0 +1,13 @@
+package network.protocol.packets;
+
+import network.protocol.Packet;
+import network.protocol.PacketVisitor;
+
+public class AcknowlegdementPacket extends Packet {
+
+	@Override
+	public void accept(PacketVisitor packetVisitor) {
+
+	}
+
+}

ChatApp/app/src/main/java/network/protocol/packets/HandshakePacket.java

@@ -1,14 +0,0 @@
-package network.protocol.packets;
-
-import network.protocol.Packet;
-import network.protocol.PacketVisitor;
-
-public class HandshakePacket extends Packet {
-    public HandshakePacket() {
-    }
-
-    @Override
-    public void accept(PacketVisitor packetVisitor) {
-        packetVisitor.visitPacket(this);
-    }
-}

ChatApp/app/src/main/java/server/ServerConnexion.java

@@ -36,8 +36,18 @@ public class ServerConnexion implements PacketVisitor {
 		}
 	}
 
+	private boolean checkLogin() {
+		if (this.chatterName.isEmpty()) {
+			sendPacket(new ServerResponsePacket(Response.AuthError));
+			return false;
+		}
+		return true;
+	}
+
 	@Override
 	public void visitPacket(CreateRoomPacket packet) {
+		if (!checkLogin())
+			return;
 		boolean created = server.createRoom(packet.getRoomName(), this);
 		sendPacket(new ServerResponsePacket(created ? Response.RoomCreated : Response.RoomNotCreated));
 		if (created)
@@ -46,7 +56,9 @@ public class ServerConnexion implements PacketVisitor {
 
 	@Override
 	public void visitPacket(JoinRoomPacket packet) {
-		if(server.getRoomName(this) != null) {
+		if (!checkLogin())
+			return;
+		if (server.getRoomName(this) != null) {
 			server.leaveRoom(this);
 			return;
 		}
@@ -58,6 +70,8 @@ public class ServerConnexion implements PacketVisitor {
 
 	@Override
 	public void visitPacket(LeaveRoomPacket packet) {
+		if (!checkLogin())
+			return;
 		String roomName = this.server.getRoomName(this);
 		boolean left = server.leaveRoom(this);
 		sendPacket(new ServerResponsePacket(left ? Response.RoomLeft : Response.RoomNotLeft));
@@ -73,17 +87,22 @@ public class ServerConnexion implements PacketVisitor {
 		}
 
 		this.chatterName = packet.getPseudo();
+		sendPacket(new ServerResponsePacket(Response.AuthSuccess));
 		sendPacket(new RoomListPacket(server.getRoomNames()));
 		System.out.println("[Server] Chatter " + packet.getPseudo() + " connected !");
 	}
 
 	@Override
 	public void visitPacket(RequestRoomListPacket packet) {
+		if (!checkLogin())
+			return;
 		sendPacket(new RoomListPacket(server.getRoomNames()));
 	}
 
 	@Override
 	public void visitPacket(SendChatMessagePacket packet) {
+		if (!checkLogin())
+			return;
 		boolean messageSent = server.sendToRoom(this, packet);
 		sendPacket(new ServerResponsePacket(messageSent ? Response.MessageSent : Response.MessageNotSent));
 	}
@@ -93,20 +112,13 @@ public class ServerConnexion implements PacketVisitor {
 		this.onDisconnect();
 	}
 
-	@Override
-	public void visitPacket(HandshakePacket packet) {
-		if(this.server.handleConnexionTime(this.clientAddress)) {
-			System.out.println("[Server] Handshake received from " + clientAddress);
-		}
-		sendPacket(packet);
-	}
-
 	private void onDisconnect() {
 		if (this.server.isInRoom(this)) {
 			this.onRoomLeave(this.server.getRoomName(this));
 		}
 		this.server.removeConnexion(this);
-		System.out.println("[Server] Chatter " + chatterName + " disconnected !");
+		if (chatterName != null)
+			System.out.println("[Server] Chatter " + chatterName + " disconnected !");
 	}
 
 	private void onRoomJoin() {

ChatApp/app/src/main/java/utilities/Signal.java

@@ -0,0 +1,28 @@
+package utilities;
+
+import java.util.HashSet;
+import java.util.Set;
+
+public class Signal {
+
+	private final Set<Runnable> listeners;
+
+	public Signal() {
+		this.listeners = new HashSet<>();
+	}
+
+	public void connect(Runnable listener) {
+		this.listeners.add(listener);
+	}
+
+	public void clear() {
+		this.listeners.clear();
+	}
+
+	public void emit() {
+		for (Runnable listener : this.listeners) {
+			listener.run();
+		}
+	}
+
+}