This commit is contained in:
2025-03-12 12:56:45 +01:00
parent 07272732d4
commit 0900113bb8
7 changed files with 147 additions and 10 deletions

View File

@@ -65,11 +65,6 @@ public class ClientConnexion implements PacketVisitor, PacketHandler {
this.callback.handleConnect(); this.callback.handleConnect();
} }
@Override
public void visitPacket(RequestActualRoomPacket packet) {
throw new UnsupportedOperationException("Unimplemented method 'visitPacket'");
}
@Override @Override
public void visitPacket(ActualRoomPacket packet) { public void visitPacket(ActualRoomPacket packet) {
this.callback.handleActualRoom(packet.getRoomName()); this.callback.handleActualRoom(packet.getRoomName());
@@ -81,6 +76,15 @@ public class ClientConnexion implements PacketVisitor, PacketHandler {
this.callback.handleDisconnect(); this.callback.handleDisconnect();
} }
public void changeCallback(ClientListener callback) {
this.callback = callback;
}
@Override
public void visitPacket(RequestActualRoomPacket packet) {
throw new UnsupportedOperationException("Unimplemented method 'visitPacket'");
}
@Override @Override
public void visitPacket(CreateRoomPacket packet) { public void visitPacket(CreateRoomPacket packet) {
throw new UnsupportedOperationException("Unimplemented method 'visitPacket'"); throw new UnsupportedOperationException("Unimplemented method 'visitPacket'");
@@ -110,8 +114,4 @@ public class ClientConnexion implements PacketVisitor, PacketHandler {
public void visitPacket(SendChatMessagePacket packet) { public void visitPacket(SendChatMessagePacket packet) {
throw new UnsupportedOperationException("Unimplemented method 'visitPacket'"); throw new UnsupportedOperationException("Unimplemented method 'visitPacket'");
} }
public void changeCallback(ClientListener callback) {
this.callback = callback;
}
} }

View File

@@ -42,6 +42,9 @@ public class PacketPool {
} }
} }
/**
* Data specific to one connexion
*/
private static class AdressContext { private static class AdressContext {
public int currentSeqRecv = 0; public int currentSeqRecv = 0;
public int currentSeqSend = 0; public int currentSeqSend = 0;
@@ -50,22 +53,43 @@ public class PacketPool {
public final Map<ReliablePacket, Integer> packetsSentTries = new HashMap<>(); public final Map<ReliablePacket, Integer> packetsSentTries = new HashMap<>();
} }
/**
* @param address the address to send to
* @return The next send sequence
*/
private int getNextSeqSend(InetSocketAddress address) { private int getNextSeqSend(InetSocketAddress address) {
return this.addressContexts.get(address).currentSeqSend++; return this.addressContexts.get(address).currentSeqSend++;
} }
/**
* @param address the address which a packet was recieved
* @return the current recieve sequence number
*/
private int getCurrentSeqRecv(InetSocketAddress address) { private int getCurrentSeqRecv(InetSocketAddress address) {
return this.addressContexts.get(address).currentSeqRecv; return this.addressContexts.get(address).currentSeqRecv;
} }
/**
* Set the recieve sequence number
* @param address
* @param newValue
*/
private void setSeqRecv(InetSocketAddress address, int newValue) { private void setSeqRecv(InetSocketAddress address, int newValue) {
this.addressContexts.get(address).currentSeqRecv = newValue; this.addressContexts.get(address).currentSeqRecv = newValue;
} }
/**
* Try to add the address into memory
* @param address
*/
private void tryAddContext(InetSocketAddress address) { private void tryAddContext(InetSocketAddress address) {
this.addressContexts.putIfAbsent(address, new AdressContext()); this.addressContexts.putIfAbsent(address, new AdressContext());
} }
/**
* Construct a PacketPool
* @param socket
*/
public PacketPool(Socket socket) { public PacketPool(Socket socket) {
this.socket = socket; this.socket = socket;
this.packetQueue = new Stack<>(); this.packetQueue = new Stack<>();
@@ -88,6 +112,12 @@ public class PacketPool {
+ " with seq : " + packet.getSeq()); + " with seq : " + packet.getSeq());
} }
/**
* Send a packet to the socket
* @param packet
* @param address
* @throws IOException
*/
private void sendPacketToSocket(ReliablePacket packet, InetSocketAddress address) throws IOException { private void sendPacketToSocket(ReliablePacket packet, InetSocketAddress address) throws IOException {
var packetsSentTries = this.addressContexts.get(address).packetsSentTries; var packetsSentTries = this.addressContexts.get(address).packetsSentTries;
@@ -112,6 +142,12 @@ public class PacketPool {
} }
} }
/**
* Send a packet to socket and try to resend if an acknowlegment was not recieved
* @param packet
* @param address
* @throws IOException
*/
private synchronized void sendPacket(ReliablePacket packet, InetSocketAddress address) throws IOException { private synchronized void sendPacket(ReliablePacket packet, InetSocketAddress address) throws IOException {
sendPacketToSocket(packet, address); sendPacketToSocket(packet, address);
debugSend(packet, address); debugSend(packet, address);
@@ -125,12 +161,22 @@ public class PacketPool {
} }
} }
/**
* Send a packet (and encapsulate it)
* @param packet
* @param address
* @throws IOException
*/
public synchronized void sendPacket(Packet packet, InetSocketAddress address) throws IOException { public synchronized void sendPacket(Packet packet, InetSocketAddress address) throws IOException {
tryAddContext(address); tryAddContext(address);
ReliablePacket reliablePacket = new ReliablePacket(packet, getNextSeqSend(address)); ReliablePacket reliablePacket = new ReliablePacket(packet, getNextSeqSend(address));
sendPacket(reliablePacket, address); sendPacket(reliablePacket, address);
} }
/**
* Try to resend a packet
* @param reliablePacketAddress
*/
private void tryResend(ReliablePacketAddress reliablePacketAddress) { private void tryResend(ReliablePacketAddress reliablePacketAddress) {
try { try {
while (!Thread.interrupted()) { while (!Thread.interrupted()) {
@@ -166,6 +212,11 @@ public class PacketPool {
ctx.sendThreads.remove(Thread.currentThread()); ctx.sendThreads.remove(Thread.currentThread());
} }
/**
* @param address
* @return The smallest sequence number recieved
*/
private ReliablePacket getMinimumSeqReceived(InetSocketAddress address) { private ReliablePacket getMinimumSeqReceived(InetSocketAddress address) {
List<ReliablePacket> packetRecvBuffer = this.addressContexts.get(address).packetRecvBuffer; List<ReliablePacket> packetRecvBuffer = this.addressContexts.get(address).packetRecvBuffer;
if (packetRecvBuffer.isEmpty()) if (packetRecvBuffer.isEmpty())
@@ -175,6 +226,11 @@ public class PacketPool {
}); });
} }
/**
* Move packets in buffer to the packet recieved queue to be processed by the app
* @param address
* @return the sequence number of the last packet that was added to the queue
*/
private int fillPacketQueue(InetSocketAddress address) { private int fillPacketQueue(InetSocketAddress address) {
List<ReliablePacket> packetRecvBuffer = this.addressContexts.get(address).packetRecvBuffer; List<ReliablePacket> packetRecvBuffer = this.addressContexts.get(address).packetRecvBuffer;
ReliablePacket minimum = getMinimumSeqReceived(address); ReliablePacket minimum = getMinimumSeqReceived(address);
@@ -193,6 +249,12 @@ public class PacketPool {
return lastSeqProcessed; return lastSeqProcessed;
} }
/**
* Process packet when recieved
* @param packet
* @param address
* @throws IOException
*/
public void onPacketReceived(ReliablePacket packet, InetSocketAddress address) throws IOException { public void onPacketReceived(ReliablePacket packet, InetSocketAddress address) throws IOException {
tryAddContext(address); tryAddContext(address);
@@ -236,6 +298,9 @@ public class PacketPool {
} }
/**
* @return The next packet in the queue
*/
public Entry<Packet, InetSocketAddress> getNextPacket() { public Entry<Packet, InetSocketAddress> getNextPacket() {
if (this.packetQueue.isEmpty()) if (this.packetQueue.isEmpty())
return null; return null;
@@ -244,18 +309,29 @@ public class PacketPool {
return entry; return entry;
} }
/**
* Closes the connexion
* @param adress
*/
private void close(InetSocketAddress adress) { private void close(InetSocketAddress adress) {
var ctx = this.addressContexts.get(adress); var ctx = this.addressContexts.get(adress);
if (ctx != null) if (ctx != null)
close(ctx); close(ctx);
} }
/**
* Stop the threads of the connexion
* @param adressContext
*/
private void close(AdressContext adressContext) { private void close(AdressContext adressContext) {
for (Thread thread : adressContext.sendThreads) { for (Thread thread : adressContext.sendThreads) {
thread.interrupt(); thread.interrupt();
} }
} }
/**
* Stop the PacketPool
*/
public void close() { public void close() {
for (AdressContext adressContext : this.addressContexts.values()) { for (AdressContext adressContext : this.addressContexts.values()) {
close(adressContext); close(adressContext);

View File

@@ -14,10 +14,16 @@ public class ReliablePacket implements Serializable {
this.seq = seq; this.seq = seq;
} }
/**
* @return The encapsulated packet
*/
public Packet getPacket() { public Packet getPacket() {
return packet; return packet;
} }
/**
* @return The sequence number of this packet
*/
public int getSeq() { public int getSeq() {
return seq; return seq;
} }

View File

@@ -23,6 +23,10 @@ public class Socket {
public final Signal onClose = new Signal(); public final Signal onClose = new Signal();
/**
* Construct a UDP Socket to connect to a server
* @throws SocketException
*/
public Socket() throws SocketException { public Socket() throws SocketException {
this.socket = new DatagramSocket(); this.socket = new DatagramSocket();
this.handlers = new ArrayList<>(); this.handlers = new ArrayList<>();
@@ -31,6 +35,11 @@ public class Socket {
this.readThread.start(); this.readThread.start();
} }
/**
* Construct a UDP Socket to listen to connexion at the specified port
* @param listeningPort the port to listen to
* @throws SocketException
*/
public Socket(int listeningPort) throws SocketException { public Socket(int listeningPort) throws SocketException {
this.socket = new DatagramSocket(listeningPort); this.socket = new DatagramSocket(listeningPort);
this.handlers = new ArrayList<>(); this.handlers = new ArrayList<>();
@@ -43,11 +52,20 @@ public class Socket {
this.handlers.add(handler); this.handlers.add(handler);
} }
/**
* @return this.socket.getLocalPort()
*/
public int getLocalPort() { public int getLocalPort() {
return this.socket.getLocalPort(); return this.socket.getLocalPort();
} }
// needs to be accessible by PacketPool // needs to be accessible by PacketPool
/**
* Send a packet to the specified address
* @param packet the packet to send
* @param address the address to send to
* @throws IOException
*/
void sendPacket(ReliablePacket packet, InetSocketAddress address) throws IOException { void sendPacket(ReliablePacket packet, InetSocketAddress address) throws IOException {
ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteArrayOutputStream stream = new ByteArrayOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(stream); ObjectOutputStream oos = new ObjectOutputStream(stream);
@@ -59,6 +77,12 @@ public class Socket {
this.socket.send(dataPacket); this.socket.send(dataPacket);
} }
/**
* Try to recieve packets and send them to the PacketPool.
* This method blocks until something has been read
* @throws IOException
* @throws ClassNotFoundException
*/
private void recievePacketReliable() private void recievePacketReliable()
throws IOException, ClassNotFoundException { throws IOException, ClassNotFoundException {
byte[] buffer = new byte[65535]; byte[] buffer = new byte[65535];
@@ -73,10 +97,22 @@ public class Socket {
this.packetPool.onPacketReceived(packet, address); this.packetPool.onPacketReceived(packet, address);
} }
/**
* Send a packet to the specified address
* @param packet the packet to send
* @param address the address to send to
* @throws IOException
*/
public void sendPacket(Packet packet, InetSocketAddress address) throws IOException { public void sendPacket(Packet packet, InetSocketAddress address) throws IOException {
this.packetPool.sendPacket(packet, address); this.packetPool.sendPacket(packet, address);
} }
/**
* Recieve packet in a reliable way (packets are in the right order).
* This method is blocking
* @throws IOException
* @throws ClassNotFoundException
*/
private void recievePacket() throws IOException, ClassNotFoundException { private void recievePacket() throws IOException, ClassNotFoundException {
this.recievePacketReliable(); this.recievePacketReliable();
var entry = this.packetPool.getNextPacket(); var entry = this.packetPool.getNextPacket();
@@ -86,6 +122,9 @@ public class Socket {
} }
} }
/**
* Close the socket
*/
public void close() { public void close() {
this.socket.close(); this.socket.close();
this.packetPool.close(); this.packetPool.close();
@@ -93,12 +132,20 @@ public class Socket {
this.onClose.emit(); this.onClose.emit();
} }
/**
* Dispatch the packet
* @param packet the packet to dispatch
* @param address the incoming address
*/
void handlePacket(Packet packet, InetSocketAddress address) { void handlePacket(Packet packet, InetSocketAddress address) {
for (PacketHandler handler : this.handlers) { for (PacketHandler handler : this.handlers) {
handler.handlePacket(packet, address); handler.handlePacket(packet, address);
} }
} }
/**
* Loop to read all packets
*/
private void readLoop() { private void readLoop() {
try { try {
while (!Thread.interrupted()) { while (!Thread.interrupted()) {

View File

@@ -4,6 +4,10 @@ import java.io.Serializable;
public abstract class Packet implements Serializable { public abstract class Packet implements Serializable {
/**
* Used for double dispatch
* @param packetVisitor the visitor to dispatch to
*/
public abstract void accept(PacketVisitor packetVisitor); public abstract void accept(PacketVisitor packetVisitor);
} }

View File

@@ -4,6 +4,10 @@ import network.protocol.packets.*;
public interface PacketVisitor { public interface PacketVisitor {
/**
* Use double dispatch to process a packet
* @param packet the packet to process
*/
default void visit(Packet packet) { default void visit(Packet packet) {
packet.accept(this); packet.accept(this);
} }

View File

@@ -13,7 +13,7 @@ public class AcknowlegdementPacket extends Packet {
@Override @Override
public void accept(PacketVisitor packetVisitor) { public void accept(PacketVisitor packetVisitor) {
// this packet should not be handled by the app
} }
public int getAck() { public int getAck() {