Feat: Expiry broadcast + client event handling and robustness #278

Merged
jona.walpert merged 3 commits from feat/lobby-expiry-broadcast into main 2026-04-12 14:57:43 +02:00
5 changed files with 300 additions and 88 deletions
Showing only changes of commit bc36892699 - Show all commits
@@ -7,11 +7,17 @@ import java.io.IOException;
import java.net.Socket; import java.net.Socket;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.Future; import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.regex.Matcher; import java.util.regex.Matcher;
import java.util.regex.Pattern; import java.util.regex.Pattern;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
@@ -30,10 +36,17 @@ public class ClientService {
private final ExecutorService executor; private final ExecutorService executor;
private final boolean offlineMode; private final boolean offlineMode;
public static ArrayList<String> response;
private final AtomicInteger idGenerator; private final AtomicInteger idGenerator;
private Logger logger; private Logger logger;
private final Map<Integer, ArrayBlockingQueue<ParsedResponse>> pendingResponses =
new ConcurrentHashMap<>();
private final CopyOnWriteArrayList<Consumer<List<String>>> eventListeners =
new CopyOnWriteArrayList<>();
private Thread readerThread = null;
private final AtomicBoolean running = new AtomicBoolean(false);
private static final int READER_JOIN_TIMEOUT_MS = 500;
/** /**
* Constructs a ClientService with the given server IP and port. It establishes a socket * Constructs a ClientService with the given server IP and port. It establishes a socket
* connection to the server and initializes the TcpTransport and ExecutorService for * connection to the server and initializes the TcpTransport and ExecutorService for
@@ -43,7 +56,6 @@ public class ClientService {
* @param port The port number of the server to connect to. * @param port The port number of the server to connect to.
*/ */
public ClientService(String ip, int port) { public ClientService(String ip, int port) {
this.idGenerator = new AtomicInteger(0); this.idGenerator = new AtomicInteger(0);
this.logger = LogManager.getLogger(ClientService.class); this.logger = LogManager.getLogger(ClientService.class);
@@ -59,6 +71,100 @@ public class ClientService {
} }
executor = Executors.newSingleThreadExecutor(); executor = Executors.newSingleThreadExecutor();
startReaderThread();
}
private void startReaderThread() {
running.set(true);
readerThread =
new Thread(
() -> {
while (running.get()) {
try {
RawPacket rp = clienttcptransport.read();
processRawPacket(rp);
} catch (IOException e) {
if (running.get()) {
logger.warn("IO error on transport reader", e);
}
break;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
},
"casono-client-reader");
readerThread.setDaemon(true);
readerThread.start();
}
private void processRawPacket(RawPacket rp) throws InterruptedException, IOException {
int rid = rp.requestId();
String responseText = rp.payload();
logger.info("Raw message '{}'", responseText);
boolean hasStatus = false;
boolean success = false;
List<String> lines = new ArrayList<>();
for (String rawLine : responseText.split("\\n")) {
String line = rawLine;
if (!hasStatus) {
if ("+OK".equals(line)) {
success = true;
hasStatus = true;
continue;
}
if (line.startsWith("-ERR") || line.startsWith("-ERROR")) {
success = false;
hasStatus = true;
continue;
}
continue;
}
if ("END".equals(line)) {
break;
}
if (line.startsWith("\t")) {
line = line.substring(1);
}
lines.add(line);
}
if (!hasStatus) {
if (responseText.contains("+OK")) {
success = true;
hasStatus = true;
} else if (responseText.contains("-ERR") || responseText.contains("-ERROR")) {
success = false;
hasStatus = true;
} else {
// best effort: treat as success
success = true;
hasStatus = true;
}
}
if (rid == 0) {
for (Consumer<List<String>> l : eventListeners) {
try {
l.accept(List.copyOf(lines));
} catch (Exception e) {
logger.warn("Event listener threw", e);
}
}
} else {
ArrayBlockingQueue<ParsedResponse> q = pendingResponses.get(rid);
if (q != null) {
q.put(new ParsedResponse(success, lines));
} else {
logger.warn("No pending response queue for id {}", rid);
}
}
} }
/** /**
@@ -124,6 +230,19 @@ public class ClientService {
.toList(); .toList();
} }
private static record ParsedResponse(boolean success, List<String> lines) {}
/**
* Register an event listener that receives unsolicited event payload lines (no status prefix).
*/
public void addEventListener(Consumer<List<String>> listener) {
eventListeners.add(listener);
}
public void removeEventListener(Consumer<List<String>> listener) {
eventListeners.remove(listener);
}
/** /**
* Sends a command to the server and processes the multi-line response. It handles the protocol * Sends a command to the server and processes the multi-line response. It handles the protocol
* handshake (expecting +OK), strips leading tabs from response lines, and collects them until * handshake (expecting +OK), strips leading tabs from response lines, and collects them until
@@ -135,74 +254,51 @@ public class ClientService {
* occurs. * occurs.
*/ */
protected List<String> processCommand(String message) { protected List<String> processCommand(String message) {
List<String> response = new ArrayList<>(); if (offlineMode) {
sendRequest( throw new RuntimeException("ClientService is offline");
() -> { }
try {
writeToTransport(message);
String responseText = clienttcptransport.read().payload();
logger.info("Raw message '{}'", responseText);
boolean hasStatus = false; int reqId = idGenerator.incrementAndGet();
boolean success = false; ArrayBlockingQueue<ParsedResponse> q = new ArrayBlockingQueue<>(1);
for (String rawLine : responseText.split("\n")) { pendingResponses.put(reqId, q);
String line = rawLine;
if (!hasStatus) { Future<?> writeFuture =
if ("+OK".equals(line)) { executor.submit(
success = true; () -> {
hasStatus = true; try {
continue; clienttcptransport.write(new RawPacket(reqId, message));
} } catch (IOException e) {
if (line.startsWith("-ERR") || line.startsWith("-ERROR")) { throw new RuntimeException(e);
success = false;
hasStatus = true;
continue;
}
// ignore any lines before the status indicator
continue;
} }
});
if ("END".equals(line)) { try {
break; writeFuture.get();
} } catch (InterruptedException e) {
Thread.currentThread().interrupt();
pendingResponses.remove(reqId);
throw new RuntimeException(e);
} catch (ExecutionException e) {
pendingResponses.remove(reqId);
throw getRuntimeException(e);
}
// strip a single leading tab if present (protocol formatting) ParsedResponse pr;
if (line.startsWith("\t")) { try {
line = line.substring(1); pr = q.take();
} } catch (InterruptedException e) {
response.add(line); Thread.currentThread().interrupt();
} pendingResponses.remove(reqId);
throw new RuntimeException(e);
} finally {
pendingResponses.remove(reqId);
}
if (!hasStatus) { if (pr.success) {
// Fallback for servers that do not place the status on a return pr.lines;
// dedicated line: try to detect status markers anywhere }
// in the payload to remain compatible with older servers.
if (responseText.contains("+OK")) {
success = true;
hasStatus = true;
} else if (responseText.contains("-ERR")
|| responseText.contains("-ERROR")) {
success = false;
hasStatus = true;
} else {
throw new RuntimeException(
"No status line in response for '"
+ message
+ "': "
+ responseText);
}
}
if (success) { throw new RuntimeException("Error in " + message + ": " + pr.lines);
return;
}
throw new RuntimeException("Error in " + message + ": " + response);
} catch (Exception e) {
throw getRuntimeException(e);
}
});
return response;
} }
/** /**
@@ -250,7 +346,17 @@ public class ClientService {
*/ */
public void closeSocket() { public void closeSocket() {
try { try {
executor.shutdown(); running.set(false);
if (readerThread != null) {
readerThread.interrupt();
try {
readerThread.join(READER_JOIN_TIMEOUT_MS);
} catch (InterruptedException ignored) {
Thread.currentThread().interrupt();
}
}
executor.shutdownNow();
clienttcptransport.close(); clienttcptransport.close();
socket.close(); socket.close();
logger.info("Socket closed"); logger.info("Socket closed");
@@ -259,16 +365,9 @@ public class ClientService {
} }
} }
/** // removed unused helper writeToTransport; write is done via processCommand()
* Method to write with the tcp transport to the server // which
* // manages request ids and response matching
* @param s - Message to be sent
* @throws IOException
*/
private void writeToTransport(String s) throws IOException {
int id = this.idGenerator.incrementAndGet();
this.clienttcptransport.write(new RawPacket(id, s));
}
public void ping() { public void ping() {
processCommand("PING"); processCommand("PING");
@@ -32,7 +32,25 @@ public class LobbyClient {
* @return A string representing the current status of the lobby, as returned by the server. * @return A string representing the current status of the lobby, as returned by the server.
*/ */
public String fetchLobbyStatusString(int lobbyId) { public String fetchLobbyStatusString(int lobbyId) {
return client.processCommand("GET_LOBBY_STATUS ID=" + lobbyId).getFirst(); List<String> lines = client.processCommand("GET_LOBBY_STATUS ID=" + lobbyId);
// Prefer explicit STATUS parameter when available
List<RequestParameter> params = ClientService.convertToRequestParameters(lines);
for (RequestParameter p : params) {
if ("STATUS".equalsIgnoreCase(p.key())) {
return p.value();
}
}
// Fallback: some servers may return a plain token as the first line
if (!lines.isEmpty()) {
String first = lines.get(0).trim();
if ("CREATED".equalsIgnoreCase(first) || "RUNNING".equalsIgnoreCase(first)) {
return first.toUpperCase();
}
}
return null;
} }
/** /**
@@ -68,8 +86,15 @@ public class LobbyClient {
* @return The id of the lobby that the client is currently in, as returned by the server. * @return The id of the lobby that the client is currently in, as returned by the server.
*/ */
public int getLobbyId() { public int getLobbyId() {
String response = client.processCommand("GET_LOBBY_ID").getFirst(); List<String> lines = client.processCommand("GET_LOBBY_ID");
return Integer.parseInt(response); if (lines.isEmpty()) {
throw new RuntimeException("GET_LOBBY_ID returned empty response");
}
try {
return Integer.parseInt(lines.get(0).trim());
} catch (NumberFormatException e) {
throw new RuntimeException("Invalid GET_LOBBY_ID response: " + lines, e);
}
} }
/** /**
@@ -2,6 +2,7 @@ package ch.unibas.dmi.dbis.cs108.casono.client.ui.lobbyui;
import ch.unibas.dmi.dbis.cs108.casono.client.network.ClientService; import ch.unibas.dmi.dbis.cs108.casono.client.network.ClientService;
import ch.unibas.dmi.dbis.cs108.casono.client.network.LobbyClient; import ch.unibas.dmi.dbis.cs108.casono.client.network.LobbyClient;
import ch.unibas.dmi.dbis.cs108.casono.server.network.command.parsing.RequestParameter;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collections; import java.util.Collections;
import java.util.List; import java.util.List;
@@ -53,7 +54,7 @@ public class LobbyButtonGridManager {
this.translationManager = LobbyButtonTranslationManager.getInstance(); this.translationManager = LobbyButtonTranslationManager.getInstance();
this.lobbyClient = lobbyClient; this.lobbyClient = lobbyClient;
startPeriodicRefresh(REFRESH_INTERVAL_SECONDS, INITIAL_DELAY_SECONDS); startPeriodicRefresh(INITIAL_DELAY_SECONDS, REFRESH_INTERVAL_SECONDS);
} }
public LobbyButtonGridManager( public LobbyButtonGridManager(
@@ -62,6 +63,43 @@ public class LobbyButtonGridManager {
ClientService clientService) { ClientService clientService) {
this(gridPane, translationManager, new LobbyClient(clientService)); this(gridPane, translationManager, new LobbyClient(clientService));
// Subscribe to server-initiated events so closed lobbies are removed
// immediately
clientService.addEventListener(
lines -> {
List<RequestParameter> params = ClientService.convertToRequestParameters(lines);
String evt = null;
String lidStr = null;
for (RequestParameter p : params) {
if ("EVENT".equalsIgnoreCase(p.key())) {
evt = p.value();
} else if ("LOBBY_ID".equalsIgnoreCase(p.key())) {
lidStr = p.value();
}
}
if (evt != null && "LOBBY_CLOSED".equalsIgnoreCase(evt) && lidStr != null) {
int lid;
try {
lid = Integer.parseInt(lidStr);
} catch (NumberFormatException ex) {
return;
}
// find button id(s) for this lobby and remove mapping
Map<Integer, Integer> mapping = translationManager.getButtonIdToLobbyId();
Integer toRemove = null;
for (Map.Entry<Integer, Integer> e : mapping.entrySet()) {
if (e.getValue() != null && e.getValue().intValue() == lid) {
toRemove = e.getKey();
break;
}
}
if (toRemove != null) {
translationManager.removeLobbyButton(toRemove);
javafx.application.Platform.runLater(this::renderLobbyButtons);
}
}
});
} }
private void startPeriodicRefresh(long initialDelay, long period) { private void startPeriodicRefresh(long initialDelay, long period) {
@@ -80,7 +118,11 @@ public class LobbyButtonGridManager {
for (Map.Entry<Integer, Integer> e : entries) { for (Map.Entry<Integer, Integer> e : entries) {
int buttonId = e.getKey(); int buttonId = e.getKey();
int lobbyId = e.getValue(); Integer lobbyIdObj = e.getValue();
if (lobbyIdObj == null) {
continue;
}
int lobbyId = lobbyIdObj.intValue();
CompletableFuture.supplyAsync( CompletableFuture.supplyAsync(
() -> { () -> {
@@ -97,8 +139,7 @@ public class LobbyButtonGridManager {
if (status == null) { if (status == null) {
translationManager.removeLobbyButton(buttonId); translationManager.removeLobbyButton(buttonId);
javafx.application.Platform.runLater( javafx.application.Platform.runLater(this::renderLobbyButtons);
this::updateLobbyButtonImages);
} }
}); });
} }
@@ -118,7 +159,11 @@ public class LobbyButtonGridManager {
for (int index = 0; index < buttonIds.size(); index++) { for (int index = 0; index < buttonIds.size(); index++) {
Integer buttonId = buttonIds.get(index); Integer buttonId = buttonIds.get(index);
int lobbyId = mapping.get(buttonId); Integer lobbyIdObj = mapping.get(buttonId);
if (lobbyIdObj == null) {
continue;
}
int lobbyId = lobbyIdObj.intValue();
Button btn = createLobbyButton(buttonId, lobbyId); Button btn = createLobbyButton(buttonId, lobbyId);
@@ -253,7 +298,11 @@ public class LobbyButtonGridManager {
} }
for (Integer buttonId : mapping.keySet()) { for (Integer buttonId : mapping.keySet()) {
int lobbyId = mapping.get(buttonId); Integer lobbyIdObj = mapping.get(buttonId);
if (lobbyIdObj == null) {
continue;
}
int lobbyId = lobbyIdObj.intValue();
CompletableFuture.supplyAsync( CompletableFuture.supplyAsync(
() -> { () -> {
@@ -361,8 +410,4 @@ public class LobbyButtonGridManager {
public LobbyClient getLobbyClient() { public LobbyClient getLobbyClient() {
return lobbyClient; return lobbyClient;
} }
public void refreshNow() {
refreshMappings();
}
} }
@@ -33,7 +33,11 @@ import ch.unibas.dmi.dbis.cs108.casono.server.network.command.execution.CommandR
import ch.unibas.dmi.dbis.cs108.casono.server.network.command.parsing.CommandParserDispatcher; import ch.unibas.dmi.dbis.cs108.casono.server.network.command.parsing.CommandParserDispatcher;
import ch.unibas.dmi.dbis.cs108.casono.server.network.events.DisconnectEvent; import ch.unibas.dmi.dbis.cs108.casono.server.network.events.DisconnectEvent;
import ch.unibas.dmi.dbis.cs108.casono.server.network.events.EventBus; import ch.unibas.dmi.dbis.cs108.casono.server.network.events.EventBus;
import ch.unibas.dmi.dbis.cs108.casono.server.network.protocol.request.RequestContext;
import ch.unibas.dmi.dbis.cs108.casono.server.network.protocol.response.SuccessResponse;
import ch.unibas.dmi.dbis.cs108.casono.server.network.protocol.response.builder.ResponseBody;
import ch.unibas.dmi.dbis.cs108.casono.server.network.protocol.response.dispatcher.ResponseDispatcher; import ch.unibas.dmi.dbis.cs108.casono.server.network.protocol.response.dispatcher.ResponseDispatcher;
import ch.unibas.dmi.dbis.cs108.casono.server.network.sessions.Session;
import ch.unibas.dmi.dbis.cs108.casono.server.network.sessions.SessionDisconnectJob; import ch.unibas.dmi.dbis.cs108.casono.server.network.sessions.SessionDisconnectJob;
import ch.unibas.dmi.dbis.cs108.casono.server.network.sessions.SessionManager; import ch.unibas.dmi.dbis.cs108.casono.server.network.sessions.SessionManager;
import java.time.Duration; import java.time.Duration;
@@ -51,6 +55,9 @@ public class ServerApp {
private static final int SESSION_DISCONNECT_JOB_DELAY = 0; private static final int SESSION_DISCONNECT_JOB_DELAY = 0;
private static final int SESSION_DISCONNECT_JOB_PERIOD = 2; private static final int SESSION_DISCONNECT_JOB_PERIOD = 2;
private static final int SESSION_DISCONNECT_JOB_TIMEOUT = 5; private static final int SESSION_DISCONNECT_JOB_TIMEOUT = 5;
private static final int LOBBY_EXPIRY_SECONDS = 30;
private static final int LOBBY_CLEANUP_INITIAL_DELAY_SECONDS = 5;
private static final int LOBBY_CLEANUP_PERIOD_SECONDS = 5;
public static void start(String arg) { public static void start(String arg) {
int port = Integer.parseInt(arg); int port = Integer.parseInt(arg);
@@ -89,6 +96,41 @@ public class ServerApp {
LobbyManager lobbyManager = new LobbyManager(); LobbyManager lobbyManager = new LobbyManager();
registerCommands(dispatcher, router, responseDispatcher, userRegistry, lobbyManager); registerCommands(dispatcher, router, responseDispatcher, userRegistry, lobbyManager);
// Periodic cleanup: remove empty lobbies older than 30s and notify affected
// users
scheduler.scheduleAtFixedRate(
() -> {
try {
var expired =
lobbyManager.findEmptyLobbiesOlderThan(
Duration.ofSeconds(LOBBY_EXPIRY_SECONDS));
for (var lid : expired) {
// remove lobby from manager first
lobbyManager.removeLobby(lid);
// broadcast LOBBY_CLOSED event to all connected sessions
// (requestId=0)
for (Session s : sessionManager.getAllSessions()) {
RequestContext ctx = new RequestContext(s.getId(), 0);
SuccessResponse ev =
new SuccessResponse(
ctx,
ResponseBody.builder()
.param("EVENT", "LOBBY_CLOSED")
.param("LOBBY_ID", lid.value())
.build()) {};
responseDispatcher.dispatch(ev);
}
}
} catch (Exception e) {
logger.warn("Lobby expiry job failed", e);
}
},
LOBBY_CLEANUP_INITIAL_DELAY_SECONDS,
LOBBY_CLEANUP_PERIOD_SECONDS,
TimeUnit.SECONDS);
NetworkManager networkManager = new NetworkManager(port, sessionManager, router); NetworkManager networkManager = new NetworkManager(port, sessionManager, router);
networkManager.start(); networkManager.start();
} }
@@ -22,6 +22,7 @@ public class GetLobbyStatusResponse extends SuccessResponse {
super( super(
context, context,
ResponseBody.builder() ResponseBody.builder()
.param("STATUS", lobby.getGameController() != null ? "RUNNING" : "CREATED")
.block( .block(
"LOBBY", "LOBBY",
lb -> { lb -> {