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
6 changed files with 321 additions and 93 deletions
@@ -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");
}
int reqId = idGenerator.incrementAndGet();
ArrayBlockingQueue<ParsedResponse> q = new ArrayBlockingQueue<>(1);
pendingResponses.put(reqId, q);
Future<?> writeFuture =
executor.submit(
() -> { () -> {
try { try {
writeToTransport(message); clienttcptransport.write(new RawPacket(reqId, message));
String responseText = clienttcptransport.read().payload(); } catch (IOException e) {
logger.info("Raw message '{}'", responseText); throw new RuntimeException(e);
boolean hasStatus = false;
boolean success = false;
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;
}
// ignore any lines before the status indicator
continue;
}
if ("END".equals(line)) {
break;
}
// strip a single leading tab if present (protocol formatting)
if (line.startsWith("\t")) {
line = line.substring(1);
}
response.add(line);
}
if (!hasStatus) {
// Fallback for servers that do not place the status on a
// 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) {
return;
}
throw new RuntimeException("Error in " + message + ": " + response);
} catch (Exception e) {
throw getRuntimeException(e);
} }
}); });
return response;
try {
writeFuture.get();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
pendingResponses.remove(reqId);
throw new RuntimeException(e);
} catch (ExecutionException e) {
pendingResponses.remove(reqId);
throw getRuntimeException(e);
}
ParsedResponse pr;
try {
pr = q.take();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
pendingResponses.remove(reqId);
throw new RuntimeException(e);
} finally {
pendingResponses.remove(reqId);
}
if (pr.success) {
return pr.lines;
}
throw new RuntimeException("Error in " + message + ": " + pr.lines);
} }
/** /**
@@ -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");
@@ -34,7 +34,7 @@ public class LobbyClient {
public String fetchLobbyStatusString(int lobbyId) { public String fetchLobbyStatusString(int lobbyId) {
List<String> lines = client.processCommand("GET_LOBBY_STATUS ID=" + lobbyId); List<String> lines = client.processCommand("GET_LOBBY_STATUS ID=" + lobbyId);
// Prefer an explicit STATUS parameter if provided by the server // Prefer explicit STATUS parameter when available
List<RequestParameter> params = ClientService.convertToRequestParameters(lines); List<RequestParameter> params = ClientService.convertToRequestParameters(lines);
for (RequestParameter p : params) { for (RequestParameter p : params) {
if ("STATUS".equalsIgnoreCase(p.key())) { if ("STATUS".equalsIgnoreCase(p.key())) {
@@ -42,11 +42,11 @@ public class LobbyClient {
} }
} }
// Fallback: look for plain status tokens in the body // Fallback: some servers may return a plain token as the first line
for (String l : lines) { if (!lines.isEmpty()) {
String t = l.trim(); String first = lines.get(0).trim();
if ("CREATED".equalsIgnoreCase(t) || "RUNNING".equalsIgnoreCase(t)) { if ("CREATED".equalsIgnoreCase(first) || "RUNNING".equalsIgnoreCase(first)) {
return t; return first.toUpperCase();
} }
} }
@@ -86,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 -> {
@@ -1,6 +1,9 @@
package ch.unibas.dmi.dbis.cs108.casono.server.domain.lobby; package ch.unibas.dmi.dbis.cs108.casono.server.domain.lobby;
import ch.unibas.dmi.dbis.cs108.casono.server.domain.lobby.Lobby.AddResult; import ch.unibas.dmi.dbis.cs108.casono.server.domain.lobby.Lobby.AddResult;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collection; import java.util.Collection;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
@@ -17,6 +20,7 @@ public class LobbyManager {
private final Map<LobbyId, Lobby> activeLobbies = new ConcurrentHashMap<>(); private final Map<LobbyId, Lobby> activeLobbies = new ConcurrentHashMap<>();
private final Map<String, LobbyId> playerToLobby = new ConcurrentHashMap<>(); private final Map<String, LobbyId> playerToLobby = new ConcurrentHashMap<>();
private final Map<LobbyId, Instant> creationTimes = new ConcurrentHashMap<>();
private final List<LobbyEventListener> listeners = new CopyOnWriteArrayList<>(); private final List<LobbyEventListener> listeners = new CopyOnWriteArrayList<>();
private final int maxPlayersPerLobby; private final int maxPlayersPerLobby;
private static final Logger LOGGER = Logger.getLogger(LobbyManager.class.getName()); private static final Logger LOGGER = Logger.getLogger(LobbyManager.class.getName());
@@ -39,6 +43,7 @@ public class LobbyManager {
if (!activeLobbies.containsKey(id)) { if (!activeLobbies.containsKey(id)) {
Lobby lobby = new Lobby(id, name == null ? ("Room " + i) : name); Lobby lobby = new Lobby(id, name == null ? ("Room " + i) : name);
activeLobbies.put(id, lobby); activeLobbies.put(id, lobby);
creationTimes.put(id, Instant.now());
return id; return id;
} }
} }
@@ -136,6 +141,35 @@ public class LobbyManager {
return activeLobbies.values(); return activeLobbies.values();
} }
/** Find all empty lobbies that were created more than the given {@code age} ago. */
public List<LobbyId> findEmptyLobbiesOlderThan(Duration age) {
List<LobbyId> result = new ArrayList<>();
Instant cutoff = Instant.now().minus(age);
for (Map.Entry<LobbyId, Lobby> e : activeLobbies.entrySet()) {
LobbyId id = e.getKey();
Lobby l = e.getValue();
if (l.getPlayerNames().isEmpty()) {
Instant created = creationTimes.get(id);
if (created != null && created.isBefore(cutoff)) {
result.add(id);
}
}
}
return result;
}
/** Remove a lobby and clean up internal mappings. */
public void removeLobby(LobbyId id) {
Lobby removed = activeLobbies.remove(id);
creationTimes.remove(id);
if (removed == null) {
return;
}
for (String username : removed.getPlayerNames()) {
playerToLobby.remove(username);
}
}
/** /**
* Apply the given action to every player username in the lobby identified by {@code lobbyId}. * Apply the given action to every player username in the lobby identified by {@code lobbyId}.
* This is a small helper that keeps iteration logic centralized and avoids leaking internal * This is a small helper that keeps iteration logic centralized and avoids leaking internal