feat(backend): serve the WebSocket protocol and the HTTP API
RtmtServer ports the WebSocket protocol (auth handshake, sockets per user, ping) and runs on the game loop; the websockets-next endpoint at / only forwards socket events. The HTTP routes, CORS and the access log match the Express app. Backend wires everything and restores live games before handling socket events. End-to-end tests play games over real sockets against Postgres and Valkey.
This commit is contained in:
parent
d86fca8c54
commit
0c112eb716
@ -38,7 +38,11 @@
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.quarkus</groupId>
|
||||
<artifactId>quarkus-rest-jackson</artifactId>
|
||||
<artifactId>quarkus-rest</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.quarkus</groupId>
|
||||
<artifactId>quarkus-jackson</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.quarkus</groupId>
|
||||
|
||||
186
apps/backend-quarkus/src/main/java/app/playmancala/Backend.java
Normal file
186
apps/backend-quarkus/src/main/java/app/playmancala/Backend.java
Normal file
@ -0,0 +1,186 @@
|
||||
package app.playmancala;
|
||||
|
||||
import java.sql.SQLException;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import org.eclipse.microprofile.config.inject.ConfigProperty;
|
||||
import org.jboss.logging.Logger;
|
||||
|
||||
import app.playmancala.auth.TokenVerifier;
|
||||
import app.playmancala.game.GameClockScheduler;
|
||||
import app.playmancala.game.GameManager;
|
||||
import app.playmancala.game.GameRecorder;
|
||||
import app.playmancala.game.GameStore;
|
||||
import app.playmancala.game.LiveGames;
|
||||
import app.playmancala.game.MatchMaker;
|
||||
import app.playmancala.history.GameHistoryService;
|
||||
import app.playmancala.history.Migrations;
|
||||
import app.playmancala.history.PostgresGameHistoryRepository;
|
||||
import app.playmancala.live.ValkeyCache;
|
||||
import app.playmancala.live.ValkeyLiveGameStore;
|
||||
import app.playmancala.loop.GameLoop;
|
||||
import app.playmancala.rtmt.RtmtServer;
|
||||
import io.quarkus.redis.datasource.RedisDataSource;
|
||||
import io.quarkus.runtime.Startup;
|
||||
import jakarta.annotation.PostConstruct;
|
||||
import jakarta.annotation.PreDestroy;
|
||||
import jakarta.enterprise.context.ApplicationScoped;
|
||||
import jakarta.inject.Inject;
|
||||
|
||||
// Builds the backend from the infrastructure Quarkus provides and starts it (docs/specs/quarkus-backend, Startup).
|
||||
@Startup
|
||||
@ApplicationScoped
|
||||
public class Backend {
|
||||
private static final Logger LOG = Logger.getLogger(Backend.class);
|
||||
private static final long MIGRATION_RETRY_MS = 30_000;
|
||||
private static final long VALKEY_STARTUP_WAIT_MS = 5_000;
|
||||
|
||||
@Inject
|
||||
DataSource dataSource;
|
||||
|
||||
@Inject
|
||||
RedisDataSource redis;
|
||||
|
||||
@Inject
|
||||
TokenVerifier tokenVerifier;
|
||||
|
||||
@ConfigProperty(name = "mancala.auth.legacy-userkey-auth")
|
||||
String legacyUserKeyAuth;
|
||||
|
||||
@ConfigProperty(name = "mancala.game-clock-ms")
|
||||
long gameClockMs;
|
||||
|
||||
private GameLoop loop;
|
||||
// Valkey writes, in order.
|
||||
private ExecutorService storeWriter;
|
||||
// Migrations and saving finished games.
|
||||
private ScheduledExecutorService recorderIo;
|
||||
// Token checks.
|
||||
private ExecutorService verifier;
|
||||
|
||||
private RtmtServer rtmt;
|
||||
private GameStore gameStore;
|
||||
private GameManager gameManager;
|
||||
private LiveGames liveGames;
|
||||
private GameHistoryService history;
|
||||
private GameClockScheduler clockScheduler;
|
||||
|
||||
@PostConstruct
|
||||
void start() {
|
||||
loop = new GameLoop();
|
||||
storeWriter = Executors.newSingleThreadExecutor(Thread.ofPlatform().name("valkey-writer").daemon().factory());
|
||||
recorderIo = Executors.newSingleThreadScheduledExecutor(Thread.ofPlatform().name("game-recorder").daemon().factory());
|
||||
verifier = Executors.newVirtualThreadPerTaskExecutor();
|
||||
|
||||
ValkeyLiveGameStore liveGameStore = new ValkeyLiveGameStore(redis);
|
||||
PostgresGameHistoryRepository repository = new PostgresGameHistoryRepository(dataSource);
|
||||
GameRecorder recorder = new GameRecorder(repository, liveGameStore, recorderIo);
|
||||
history = new GameHistoryService(repository, new ValkeyCache(redis), recorder::getUnsaved);
|
||||
recorder.onSaved(history::onGameSaved);
|
||||
|
||||
// Exactly "true", like the Node backend.
|
||||
rtmt = new RtmtServer(tokenVerifier, "true".equals(legacyUserKeyAuth), loop, verifier,
|
||||
RtmtServer.DEFAULT_AUTH_TIMEOUT_MS);
|
||||
gameStore = new GameStore();
|
||||
gameManager = new GameManager(gameStore, rtmt, new MatchMaker(rtmt), liveGameStore, recorder, storeWriter,
|
||||
loop::now, gameClockMs);
|
||||
liveGames = new LiveGames(rtmt, gameStore, gameManager, loop);
|
||||
clockScheduler = new GameClockScheduler(gameManager::endGamesOutOfTime, loop,
|
||||
GameClockScheduler.CLOCK_CHECK_INTERVAL_MS);
|
||||
|
||||
// The game runs without the database; saved games wait in the recorder until migrations have run.
|
||||
recorderIo.execute(this::migrate);
|
||||
recorderIo.scheduleWithFixedDelay(() -> guarded(recorder::retryPending), 0, GameRecorder.RETRY_INTERVAL_MS,
|
||||
TimeUnit.MILLISECONDS);
|
||||
// The first task on the loop, so socket events queue behind it and reconnecting players find their games.
|
||||
loop.execute(() -> {
|
||||
waitForValkey();
|
||||
gameManager.restoreLiveGames();
|
||||
clockScheduler.start();
|
||||
rtmt.start();
|
||||
});
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
void stop() {
|
||||
loop.close();
|
||||
storeWriter.shutdown();
|
||||
recorderIo.shutdownNow();
|
||||
verifier.shutdownNow();
|
||||
}
|
||||
|
||||
private void migrate() {
|
||||
try {
|
||||
Migrations.run(dataSource);
|
||||
} catch (SQLException | RuntimeException e) {
|
||||
LOG.errorf("(DB) Migrations failed, retrying in %ds: %s", MIGRATION_RETRY_MS / 1000, e.getMessage());
|
||||
recorderIo.schedule(this::migrate, MIGRATION_RETRY_MS, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
}
|
||||
|
||||
private void waitForValkey() {
|
||||
long deadline = System.currentTimeMillis() + VALKEY_STARTUP_WAIT_MS;
|
||||
while (true) {
|
||||
try {
|
||||
redis.execute("PING");
|
||||
return;
|
||||
} catch (RuntimeException e) {
|
||||
if (System.currentTimeMillis() >= deadline) {
|
||||
LOG.error("(Valkey) Not reachable yet; starting without restoring live games");
|
||||
return;
|
||||
}
|
||||
}
|
||||
try {
|
||||
Thread.sleep(250);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static void guarded(Runnable task) {
|
||||
try {
|
||||
task.run();
|
||||
} catch (RuntimeException e) {
|
||||
LOG.error("(Backend) Task failed", e);
|
||||
}
|
||||
}
|
||||
|
||||
public GameLoop loop() {
|
||||
return loop;
|
||||
}
|
||||
|
||||
public RtmtServer rtmt() {
|
||||
return rtmt;
|
||||
}
|
||||
|
||||
public GameStore gameStore() {
|
||||
return gameStore;
|
||||
}
|
||||
|
||||
public GameManager gameManager() {
|
||||
return gameManager;
|
||||
}
|
||||
|
||||
public LiveGames liveGames() {
|
||||
return liveGames;
|
||||
}
|
||||
|
||||
public GameHistoryService history() {
|
||||
return history;
|
||||
}
|
||||
|
||||
public TokenVerifier tokenVerifier() {
|
||||
return tokenVerifier;
|
||||
}
|
||||
|
||||
public boolean legacyUserKeyAuth() {
|
||||
return "true".equals(legacyUserKeyAuth);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,23 @@
|
||||
package app.playmancala.auth;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
import org.eclipse.microprofile.config.inject.ConfigProperty;
|
||||
|
||||
import jakarta.enterprise.inject.Produces;
|
||||
import jakarta.inject.Singleton;
|
||||
|
||||
public class TokenVerifierProducer {
|
||||
@Produces
|
||||
@Singleton
|
||||
TokenVerifier tokenVerifier(
|
||||
@ConfigProperty(name = "mancala.auth.issuer") String issuer,
|
||||
@ConfigProperty(name = "mancala.auth.jwks-url") Optional<String> jwksUrl,
|
||||
@ConfigProperty(name = "mancala.auth.client-ids") String clientIds) {
|
||||
// Comma-separated; spaces and empty entries are ignored, as in the Node backend.
|
||||
List<String> allowedClientIds = Arrays.stream(clientIds.split(",")).map(String::trim).filter(id -> !id.isEmpty()).toList();
|
||||
return KeycloakTokenVerifier.create(issuer, allowedClientIds, jwksUrl.orElse(null));
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,35 @@
|
||||
package app.playmancala.http;
|
||||
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
import org.jboss.logging.Logger;
|
||||
|
||||
import app.playmancala.auth.InvalidTokenException;
|
||||
import app.playmancala.auth.TokenVerifier;
|
||||
import app.playmancala.model.AuthUser;
|
||||
import jakarta.ws.rs.WebApplicationException;
|
||||
import jakarta.ws.rs.core.Response;
|
||||
|
||||
// Requires an "Authorization: Bearer <access token>" header with the same Keycloak token the WebSocket uses.
|
||||
final class BearerAuth {
|
||||
private static final Logger LOG = Logger.getLogger(BearerAuth.class);
|
||||
private static final Pattern BEARER = Pattern.compile("^Bearer (.+)$");
|
||||
|
||||
private BearerAuth() {
|
||||
}
|
||||
|
||||
// The user, or a 401 response.
|
||||
static AuthUser requireUser(TokenVerifier tokenVerifier, String authorization) {
|
||||
Matcher match = BEARER.matcher(authorization == null ? "" : authorization);
|
||||
if (!match.matches()) throw new WebApplicationException(Response.Status.UNAUTHORIZED);
|
||||
try {
|
||||
return tokenVerifier.verify(match.group(1));
|
||||
} catch (InvalidTokenException e) {
|
||||
throw new WebApplicationException(Response.Status.UNAUTHORIZED);
|
||||
} catch (RuntimeException e) {
|
||||
LOG.error("(Auth) Token verification failed", e);
|
||||
throw new WebApplicationException(Response.Status.UNAUTHORIZED);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,20 @@
|
||||
package app.playmancala.http;
|
||||
|
||||
import org.jboss.logging.Logger;
|
||||
import org.jboss.resteasy.reactive.server.ServerExceptionMapper;
|
||||
|
||||
import io.vertx.core.http.HttpServerRequest;
|
||||
import jakarta.ws.rs.WebApplicationException;
|
||||
import jakarta.ws.rs.core.Response;
|
||||
|
||||
// A failing dependency (the database, mostly) answers 503, as the Express error handler did.
|
||||
public class ErrorMapper {
|
||||
private static final Logger LOG = Logger.getLogger(ErrorMapper.class);
|
||||
|
||||
@ServerExceptionMapper
|
||||
public Response map(RuntimeException error, HttpServerRequest request) {
|
||||
if (error instanceof WebApplicationException webError) return webError.getResponse();
|
||||
LOG.errorf("(HTTP) %s %s failed: %s", request.method(), request.path(), error.getMessage());
|
||||
return Response.status(Response.Status.SERVICE_UNAVAILABLE).build();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,125 @@
|
||||
package app.playmancala.http;
|
||||
|
||||
import java.math.BigInteger;
|
||||
import java.util.HashMap;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
import app.playmancala.Backend;
|
||||
import app.playmancala.game.BackendGame;
|
||||
import app.playmancala.game.LiveGames;
|
||||
import app.playmancala.history.GameHistoryService;
|
||||
import app.playmancala.model.AuthUser;
|
||||
import app.playmancala.model.GameRecord;
|
||||
import app.playmancala.model.Json;
|
||||
import app.playmancala.model.LiveGameSummary;
|
||||
import jakarta.inject.Inject;
|
||||
import jakarta.ws.rs.GET;
|
||||
import jakarta.ws.rs.HeaderParam;
|
||||
import jakarta.ws.rs.NotFoundException;
|
||||
import jakarta.ws.rs.Path;
|
||||
import jakarta.ws.rs.PathParam;
|
||||
import jakarta.ws.rs.Produces;
|
||||
import jakarta.ws.rs.QueryParam;
|
||||
import jakarta.ws.rs.core.HttpHeaders;
|
||||
import jakarta.ws.rs.core.MediaType;
|
||||
import jakarta.ws.rs.core.Response;
|
||||
|
||||
// The HTTP API. Blocking: runs on worker threads, reads live games on the game loop and saved games itself.
|
||||
// Bodies are written with the shared mapper, so they are the same JSON as the WebSocket messages.
|
||||
@Path("/")
|
||||
public class GameResource {
|
||||
private static final Pattern LEADING_INTEGER = Pattern.compile("^\\s*([+-]?\\d+)");
|
||||
|
||||
@Inject
|
||||
Backend backend;
|
||||
|
||||
@GET
|
||||
@Produces(MediaType.TEXT_PLAIN)
|
||||
public String root() {
|
||||
return "Server up and running! ci test1";
|
||||
}
|
||||
|
||||
// Issues userKeys for legacy clients; token clients get their identity from Keycloak.
|
||||
@GET
|
||||
@Path("register{slash: /?}")
|
||||
@Produces(MediaType.TEXT_PLAIN)
|
||||
public String register() {
|
||||
if (!backend.legacyUserKeyAuth()) throw new NotFoundException();
|
||||
return UUID.randomUUID().toString();
|
||||
}
|
||||
|
||||
// A live game as players receive it, or the saved record of a finished game.
|
||||
@GET
|
||||
@Path("game/{id}")
|
||||
public Response game(@PathParam("id") String id) {
|
||||
// Serialised on the loop, because the client game refers to the live engine objects.
|
||||
String live = backend.loop().call(() -> {
|
||||
BackendGame game = backend.gameStore().get(id);
|
||||
return game == null ? null : Json.write(backend.gameManager().toClientGame(game));
|
||||
});
|
||||
if (live != null) return json(live);
|
||||
GameRecord record = backend.history().get(id);
|
||||
if (record == null) throw new NotFoundException();
|
||||
Map<String, Object> response = new LinkedHashMap<>();
|
||||
response.put("status", "finished");
|
||||
response.put("record", record);
|
||||
return json(response);
|
||||
}
|
||||
|
||||
// Live games if any are being played, otherwise the most spectated finished games. With kind=finished always the
|
||||
// finished games, for clients that get live games over the WebSocket.
|
||||
@GET
|
||||
@Path("games/popular")
|
||||
public Response popular(@QueryParam("limit") String limitParam, @QueryParam("kind") String kind) {
|
||||
int limit = parseLimit(limitParam, 5);
|
||||
List<LiveGameSummary> live = "finished".equals(kind) ? List.of()
|
||||
: backend.loop().call(() -> backend.liveGames().getTop(limit));
|
||||
Map<String, Object> response = new LinkedHashMap<>();
|
||||
if (!live.isEmpty()) {
|
||||
response.put("kind", "live");
|
||||
response.put("games", live);
|
||||
} else {
|
||||
response.put("kind", "finished");
|
||||
response.put("games", backend.history().getPopular(limit));
|
||||
}
|
||||
return json(response);
|
||||
}
|
||||
|
||||
@GET
|
||||
@Path("games/me/active")
|
||||
public Response activeGame(@HeaderParam(HttpHeaders.AUTHORIZATION) String authorization) {
|
||||
AuthUser user = BearerAuth.requireUser(backend.tokenVerifier(), authorization);
|
||||
LiveGameSummary game = backend.loop().call(() -> {
|
||||
BackendGame active = backend.gameStore().getGameByUser(user.id());
|
||||
return active == null ? null : LiveGames.toLiveGameSummary(active);
|
||||
});
|
||||
Map<String, Object> response = new HashMap<>();
|
||||
response.put("game", game);
|
||||
return json(response);
|
||||
}
|
||||
|
||||
@GET
|
||||
@Path("games/me")
|
||||
public Response myGames(@HeaderParam(HttpHeaders.AUTHORIZATION) String authorization,
|
||||
@QueryParam("limit") String limit) {
|
||||
AuthUser user = BearerAuth.requireUser(backend.tokenVerifier(), authorization);
|
||||
return json(Map.of("games", backend.history().getRecentForUser(user.id(), parseLimit(limit, 3))));
|
||||
}
|
||||
|
||||
private static Response json(Object body) {
|
||||
return Response.ok(body instanceof String text ? text : Json.write(body), MediaType.APPLICATION_JSON).build();
|
||||
}
|
||||
|
||||
// Like parseInt in the Node backend: the leading integer, clamped to 1..MAX_LIST_LIMIT.
|
||||
static int parseLimit(String value, int fallback) {
|
||||
Matcher match = LEADING_INTEGER.matcher(value == null ? "" : value);
|
||||
if (!match.find()) return fallback;
|
||||
BigInteger limit = new BigInteger(match.group(1));
|
||||
return limit.max(BigInteger.ONE).min(BigInteger.valueOf(GameHistoryService.MAX_LIST_LIMIT)).intValue();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,93 @@
|
||||
package app.playmancala.rtmt;
|
||||
|
||||
import java.net.URLDecoder;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
import org.jboss.logging.Logger;
|
||||
|
||||
import app.playmancala.Backend;
|
||||
import io.quarkus.websockets.next.CloseReason;
|
||||
import io.quarkus.websockets.next.OnClose;
|
||||
import io.quarkus.websockets.next.OnError;
|
||||
import io.quarkus.websockets.next.OnOpen;
|
||||
import io.quarkus.websockets.next.OnTextMessage;
|
||||
import io.quarkus.websockets.next.WebSocket;
|
||||
import io.quarkus.websockets.next.WebSocketConnection;
|
||||
import io.smallrye.common.annotation.NonBlocking;
|
||||
import jakarta.inject.Inject;
|
||||
|
||||
// Clients connect to the root path, as they did to the Node backend. Events run on the connection's event loop, in
|
||||
// order, and are only handed to the game loop, where RtmtServer handles them.
|
||||
@WebSocket(path = "/")
|
||||
public class RtmtEndpoint {
|
||||
private static final Logger LOG = Logger.getLogger(RtmtEndpoint.class);
|
||||
|
||||
@Inject
|
||||
Backend backend;
|
||||
|
||||
@OnOpen
|
||||
@NonBlocking
|
||||
public void onOpen(WebSocketConnection connection) {
|
||||
RtmtSocket socket = new ConnectionSocket(connection);
|
||||
backend.loop().execute(() -> backend.rtmt().onOpen(socket));
|
||||
}
|
||||
|
||||
@OnTextMessage
|
||||
@NonBlocking
|
||||
public void onTextMessage(WebSocketConnection connection, String text) {
|
||||
RtmtSocket socket = new ConnectionSocket(connection);
|
||||
backend.loop().execute(() -> backend.rtmt().onText(socket, text));
|
||||
}
|
||||
|
||||
@OnClose
|
||||
@NonBlocking
|
||||
public void onClose(WebSocketConnection connection) {
|
||||
RtmtSocket socket = new ConnectionSocket(connection);
|
||||
backend.loop().execute(() -> backend.rtmt().onClose(socket));
|
||||
}
|
||||
|
||||
@OnError
|
||||
@NonBlocking
|
||||
public void onError(WebSocketConnection connection, Throwable error) {
|
||||
// A close event follows, which cleans up.
|
||||
LOG.debugf("(RTMT) WebSocket error: %s", error.getMessage());
|
||||
}
|
||||
|
||||
private record ConnectionSocket(WebSocketConnection connection) implements RtmtSocket {
|
||||
@Override
|
||||
public String id() {
|
||||
return connection.id();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String queryParam(String name) {
|
||||
String query = connection.handshakeRequest().query();
|
||||
if (query == null || query.isEmpty()) return null;
|
||||
for (String pair : query.split("&")) {
|
||||
int equals = pair.indexOf('=');
|
||||
String key = equals < 0 ? pair : pair.substring(0, equals);
|
||||
if (URLDecoder.decode(key, StandardCharsets.UTF_8).equals(name)) {
|
||||
return equals < 0 ? "" : URLDecoder.decode(pair.substring(equals + 1), StandardCharsets.UTF_8);
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isOpen() {
|
||||
return connection.isOpen();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void send(String text) {
|
||||
connection.sendText(text).subscribe().with(ok -> { },
|
||||
error -> LOG.debugf("(RTMT) Could not send: %s", error.getMessage()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close(int code, String reason) {
|
||||
connection.close(new CloseReason(code, reason)).subscribe().with(ok -> { },
|
||||
error -> LOG.debugf("(RTMT) Could not close: %s", error.getMessage()));
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,256 @@
|
||||
package app.playmancala.rtmt;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.Executor;
|
||||
import java.util.function.BiConsumer;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
import org.jboss.logging.Logger;
|
||||
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.node.NullNode;
|
||||
|
||||
import app.playmancala.auth.InvalidTokenException;
|
||||
import app.playmancala.auth.TokenVerifier;
|
||||
import app.playmancala.loop.Cancellable;
|
||||
import app.playmancala.loop.Scheduler;
|
||||
import app.playmancala.model.AuthUser;
|
||||
import app.playmancala.model.Channels;
|
||||
import app.playmancala.model.Json;
|
||||
|
||||
// The WebSocket protocol: authentication, routing messages by channel, users with several sockets and the ping
|
||||
// keep-alive. Messages are JSON objects { channel, message }. Everything here runs on the game loop; the endpoint
|
||||
// forwards socket events to it, and tokens are checked on `verifier` and the outcome handled back on the loop.
|
||||
public class RtmtServer implements Rtmt {
|
||||
public static final long PING_INTERVAL_MS = 2_000;
|
||||
public static final long DEFAULT_AUTH_TIMEOUT_MS = 5_000;
|
||||
private static final Logger LOG = Logger.getLogger(RtmtServer.class);
|
||||
private static final Pattern LEGACY_USER_KEY_PATTERN =
|
||||
Pattern.compile("^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", Pattern.CASE_INSENSITIVE);
|
||||
|
||||
private record Envelope(String channel, Object message) {
|
||||
}
|
||||
|
||||
private static final class Client {
|
||||
final RtmtSocket socket;
|
||||
boolean alive = true;
|
||||
// Set once the socket is authenticated.
|
||||
String userId;
|
||||
Cancellable authTimeout;
|
||||
|
||||
Client(RtmtSocket socket) {
|
||||
this.socket = socket;
|
||||
}
|
||||
}
|
||||
|
||||
private final TokenVerifier tokenVerifier;
|
||||
private final boolean legacyUserKeyAuth;
|
||||
private final Scheduler loop;
|
||||
private final Executor verifier;
|
||||
private final long authTimeoutMs;
|
||||
|
||||
private final Map<String, BiConsumer<String, JsonNode>> messageChannels = new HashMap<>();
|
||||
// By socket id.
|
||||
private final Map<String, Client> clients = new HashMap<>();
|
||||
// A user can be connected from several devices or tabs at once.
|
||||
private final Map<String, Set<Client>> sockets = new HashMap<>();
|
||||
private final Map<String, AuthUser> users = new HashMap<>();
|
||||
private final List<BiConsumer<String, Boolean>> connectionChangeListeners = new ArrayList<>();
|
||||
private Cancellable pingTimer;
|
||||
|
||||
public RtmtServer(TokenVerifier tokenVerifier, boolean legacyUserKeyAuth, Scheduler loop, Executor verifier,
|
||||
long authTimeoutMs) {
|
||||
this.tokenVerifier = tokenVerifier;
|
||||
this.legacyUserKeyAuth = legacyUserKeyAuth;
|
||||
this.loop = loop;
|
||||
this.verifier = verifier;
|
||||
this.authTimeoutMs = authTimeoutMs;
|
||||
}
|
||||
|
||||
public void start() {
|
||||
stop();
|
||||
pingTimer = loop.scheduleAtFixedRate(this::ping, PING_INTERVAL_MS);
|
||||
}
|
||||
|
||||
public void stop() {
|
||||
if (pingTimer != null) pingTimer.cancel();
|
||||
pingTimer = null;
|
||||
}
|
||||
|
||||
public void onOpen(RtmtSocket socket) {
|
||||
LOG.debug("(RTMT) Client connected");
|
||||
Client client = new Client(socket);
|
||||
clients.put(socket.id(), client);
|
||||
|
||||
String legacyUserKey = socket.queryParam("userKey");
|
||||
if (legacyUserKey != null) {
|
||||
if (legacyUserKeyAuth && LEGACY_USER_KEY_PATTERN.matcher(legacyUserKey).matches()) {
|
||||
bindSocket(client, new AuthUser(legacyUserKey, null, true));
|
||||
} else {
|
||||
rejectSocket(client, "userKey login is not accepted");
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
client.authTimeout = loop.schedule(() -> {
|
||||
if (client.userId == null) rejectSocket(client, "authentication timeout");
|
||||
}, authTimeoutMs);
|
||||
}
|
||||
|
||||
public void onClose(RtmtSocket socket) {
|
||||
Client client = clients.remove(socket.id());
|
||||
if (client == null) return;
|
||||
LOG.debug("(RTMT) Client disconnected");
|
||||
if (client.authTimeout != null) client.authTimeout.cancel();
|
||||
unbindSocket(client);
|
||||
}
|
||||
|
||||
public void onText(RtmtSocket socket, String text) {
|
||||
Client client = clients.get(socket.id());
|
||||
if (client == null) return;
|
||||
JsonNode envelope;
|
||||
try {
|
||||
envelope = Json.parse(text);
|
||||
} catch (Exception e) {
|
||||
LOG.debug("(RTMT) Ignoring malformed message");
|
||||
return;
|
||||
}
|
||||
if (envelope == null || !envelope.path("channel").isTextual()) return;
|
||||
String channel = envelope.get("channel").asText();
|
||||
JsonNode message = envelope.has("message") ? envelope.get("message") : NullNode.getInstance();
|
||||
|
||||
if (channel.equals(Channels.PONG)) {
|
||||
client.alive = true;
|
||||
return;
|
||||
}
|
||||
if (channel.equals(Channels.AUTH)) {
|
||||
authenticate(client, message);
|
||||
return;
|
||||
}
|
||||
if (client.userId == null) {
|
||||
LOG.debugf("(RTMT) Ignoring message from unauthenticated client on channel %s", channel);
|
||||
return;
|
||||
}
|
||||
BiConsumer<String, JsonNode> callback = messageChannels.get(channel);
|
||||
if (callback != null) {
|
||||
callback.accept(client.userId, message);
|
||||
} else {
|
||||
LOG.debugf("(RTMT) Channel callback not found! %s", channel);
|
||||
}
|
||||
}
|
||||
|
||||
private void authenticate(Client client, JsonNode message) {
|
||||
JsonNode token = message.path("token");
|
||||
if (!token.isTextual()) {
|
||||
rejectSocket(client, "missing token");
|
||||
return;
|
||||
}
|
||||
verifier.execute(() -> {
|
||||
try {
|
||||
AuthUser user = tokenVerifier.verify(token.asText());
|
||||
loop.execute(() -> onAuthenticated(client, user));
|
||||
} catch (InvalidTokenException e) {
|
||||
loop.execute(() -> rejectSocket(client, "invalid token"));
|
||||
} catch (RuntimeException e) {
|
||||
LOG.error("(RTMT) Token verification failed", e);
|
||||
loop.execute(() -> rejectSocket(client, "authentication failed"));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private void onAuthenticated(Client client, AuthUser user) {
|
||||
if (!client.socket.isOpen() || clients.get(client.socket.id()) != client) return;
|
||||
if (client.userId != null && !client.userId.equals(user.id())) {
|
||||
rejectSocket(client, "token belongs to another user");
|
||||
return;
|
||||
}
|
||||
if (client.authTimeout != null) client.authTimeout.cancel();
|
||||
if (client.userId != null) {
|
||||
// Token refresh on an already authenticated socket.
|
||||
users.put(user.id(), user);
|
||||
} else {
|
||||
bindSocket(client, user);
|
||||
}
|
||||
send(client, Channels.ON_AUTH, Map.of("user", user));
|
||||
}
|
||||
|
||||
private void rejectSocket(Client client, String reason) {
|
||||
if (!client.socket.isOpen()) return;
|
||||
send(client, Channels.ON_AUTH_ERROR, Map.of("reason", reason));
|
||||
client.socket.close(Channels.CLOSE_AUTH_FAILED, reason);
|
||||
}
|
||||
|
||||
private void bindSocket(Client client, AuthUser user) {
|
||||
client.userId = user.id();
|
||||
users.put(user.id(), user);
|
||||
sockets.computeIfAbsent(user.id(), id -> new LinkedHashSet<>()).add(client);
|
||||
// Fired for every new socket so each device gets the current game state.
|
||||
List.copyOf(connectionChangeListeners).forEach(listener -> listener.accept(user.id(), true));
|
||||
}
|
||||
|
||||
private void unbindSocket(Client client) {
|
||||
String userId = client.userId;
|
||||
if (userId == null) return;
|
||||
Set<Client> userSockets = sockets.get(userId);
|
||||
if (userSockets == null || !userSockets.remove(client)) return;
|
||||
if (userSockets.isEmpty()) {
|
||||
sockets.remove(userId);
|
||||
users.remove(userId);
|
||||
List.copyOf(connectionChangeListeners).forEach(listener -> listener.accept(userId, false));
|
||||
}
|
||||
}
|
||||
|
||||
private void ping() {
|
||||
for (Client client : List.copyOf(clients.values())) {
|
||||
if (!client.alive) {
|
||||
LOG.debug("(RTMT) Closing a socket that did not answer the ping");
|
||||
client.socket.close(1001, "ping timeout");
|
||||
continue;
|
||||
}
|
||||
client.alive = false;
|
||||
send(client, Channels.PING, Map.of());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sendMessage(String clientId, String channel, Object message) {
|
||||
Set<Client> userSockets = sockets.get(clientId);
|
||||
if (userSockets == null) {
|
||||
LOG.debugf("(RTMT) Client not connected: %s", clientId);
|
||||
return;
|
||||
}
|
||||
String text = Json.write(new Envelope(channel, message));
|
||||
userSockets.forEach(client -> {
|
||||
if (client.socket.isOpen()) client.socket.send(text);
|
||||
});
|
||||
}
|
||||
|
||||
private void send(Client client, String channel, Object message) {
|
||||
if (client.socket.isOpen()) client.socket.send(Json.write(new Envelope(channel, message)));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void listenMessage(String channel, BiConsumer<String, JsonNode> callback) {
|
||||
messageChannels.put(channel, callback);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isClientOnline(String clientId) {
|
||||
return sockets.containsKey(clientId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public AuthUser getUser(String clientId) {
|
||||
return users.get(clientId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void listenOnClientConnectionChange(BiConsumer<String, Boolean> listener) {
|
||||
connectionChangeListeners.add(listener);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,15 @@
|
||||
package app.playmancala.rtmt;
|
||||
|
||||
// A client's WebSocket as RtmtServer sees it. Sends and closes do not block.
|
||||
public interface RtmtSocket {
|
||||
String id();
|
||||
|
||||
// A query parameter of the URL the client connected to, or null.
|
||||
String queryParam(String name);
|
||||
|
||||
boolean isOpen();
|
||||
|
||||
void send(String text);
|
||||
|
||||
void close(int code, String reason);
|
||||
}
|
||||
@ -35,7 +35,7 @@ quarkus.datasource.devservices.image-name=postgres:17.6
|
||||
# Valkey. Tests start one with Dev Services. Short timeouts, so callers fall back quickly while it is down.
|
||||
%dev,prod.quarkus.redis.hosts=${VALKEY_URL:redis://localhost:6379}
|
||||
quarkus.redis.timeout=2S
|
||||
quarkus.redis.tcp.connect-timeout=1S
|
||||
quarkus.redis.tcp.connection-timeout=1S
|
||||
quarkus.redis.devservices.image-name=valkey/valkey:9.1.2
|
||||
# Nothing listens on port 1; for the tests of an unreachable Valkey.
|
||||
%test.quarkus.redis.offline.hosts=redis://127.0.0.1:1
|
||||
|
||||
@ -0,0 +1,35 @@
|
||||
package app.playmancala.fakes;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
|
||||
import app.playmancala.live.Cache;
|
||||
import app.playmancala.model.Json;
|
||||
|
||||
public class InMemoryCache implements Cache {
|
||||
public final Map<String, String> values = new HashMap<>();
|
||||
|
||||
@Override
|
||||
public <T> Optional<T> get(String key, TypeReference<T> type) {
|
||||
String value = values.get(key);
|
||||
if (value == null) return Optional.empty();
|
||||
try {
|
||||
return Optional.of(Json.MAPPER.readValue(value, type));
|
||||
} catch (Exception e) {
|
||||
return Optional.empty();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void set(String key, Object value, long ttlSeconds) {
|
||||
values.put(key, Json.write(value));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void delete(String... keys) {
|
||||
for (String key : keys) values.remove(key);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,102 @@
|
||||
package app.playmancala.history;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertSame;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import app.playmancala.fakes.InMemoryCache;
|
||||
import app.playmancala.fakes.InMemoryGameHistoryRepository;
|
||||
import app.playmancala.model.EndReason;
|
||||
import app.playmancala.model.FinishedGameSummary;
|
||||
import app.playmancala.model.GameRecord;
|
||||
import app.playmancala.model.GameResult;
|
||||
import app.playmancala.model.PlayerSummary;
|
||||
|
||||
// The list behaviour the Node backend's HTTP API tests checked: the cache, unsaved games and a database that is down.
|
||||
class GameHistoryServiceTest {
|
||||
static final PlayerSummary ALICE = new PlayerSummary("kc:alice", "alice", false);
|
||||
static final PlayerSummary BOB = new PlayerSummary("kc:bob", "bob", false);
|
||||
static final PlayerSummary CAROL = new PlayerSummary("kc:carol", "carol", false);
|
||||
|
||||
InMemoryGameHistoryRepository repository;
|
||||
InMemoryCache cache;
|
||||
List<GameRecord> unsaved;
|
||||
GameHistoryService history;
|
||||
long endedAt = 1_000_000;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
repository = new InMemoryGameHistoryRepository();
|
||||
cache = new InMemoryCache();
|
||||
unsaved = new ArrayList<>();
|
||||
history = new GameHistoryService(repository, cache, () -> List.copyOf(unsaved));
|
||||
}
|
||||
|
||||
GameRecord record(PlayerSummary player1, PlayerSummary player2, int peakSpectators) {
|
||||
endedAt += 60_000;
|
||||
return new GameRecord(UUID.randomUUID().toString(), player1, player2, List.of(), GameResult.PLAYER1,
|
||||
EndReason.LEFT, null, endedAt - 1000, endedAt, 0, peakSpectators, 300_000L, List.of(), peakSpectators);
|
||||
}
|
||||
|
||||
static List<String> ids(List<FinishedGameSummary> games) {
|
||||
return games.stream().map(FinishedGameSummary::id).toList();
|
||||
}
|
||||
|
||||
@Test
|
||||
void returnsTheUsersLatestGamesAndShowsANewOneRightAwayDespiteTheCache() {
|
||||
GameRecord first = record(ALICE, BOB, 0);
|
||||
repository.save(first);
|
||||
repository.save(record(CAROL, BOB, 0));
|
||||
assertEquals(List.of(first.id()), ids(history.getRecentForUser(ALICE.userId(), 3)));
|
||||
|
||||
GameRecord second = record(BOB, ALICE, 0);
|
||||
repository.save(second);
|
||||
// Still cached.
|
||||
assertEquals(List.of(first.id()), ids(history.getRecentForUser(ALICE.userId(), 3)));
|
||||
history.onGameSaved(second);
|
||||
assertEquals(List.of(second.id()), ids(history.getRecentForUser(ALICE.userId(), 1)));
|
||||
}
|
||||
|
||||
@Test
|
||||
void returnsAnEmptyListWhileTheDatabaseIsDown() {
|
||||
repository.failing = true;
|
||||
assertEquals(List.of(), history.getRecentForUser(ALICE.userId(), 3));
|
||||
assertEquals(List.of(), history.getPopular(5));
|
||||
}
|
||||
|
||||
@Test
|
||||
void includesGamesThatEndedButAreNotSavedYet() {
|
||||
repository.failing = true;
|
||||
GameRecord game = record(ALICE, BOB, 0);
|
||||
unsaved.add(game);
|
||||
assertEquals(List.of(game.id()), ids(history.getRecentForUser(ALICE.userId(), 3)));
|
||||
assertEquals(List.of(), ids(history.getRecentForUser(CAROL.userId(), 3)));
|
||||
assertEquals(List.of(game.id()), ids(history.getPopular(5)));
|
||||
assertSame(game, history.get(game.id()));
|
||||
}
|
||||
|
||||
@Test
|
||||
void listsPopularGamesBySpectatorsThenNewestWithoutDuplicates() {
|
||||
GameRecord quiet = record(ALICE, BOB, 0);
|
||||
GameRecord watched = record(ALICE, BOB, 3);
|
||||
GameRecord newerQuiet = record(ALICE, BOB, 0);
|
||||
List.of(quiet, watched).forEach(repository::save);
|
||||
unsaved.add(newerQuiet);
|
||||
unsaved.add(watched); // saved meanwhile
|
||||
assertEquals(List.of(watched.id(), newerQuiet.id(), quiet.id()), ids(history.getPopular(5)));
|
||||
assertEquals(List.of(watched.id()), ids(history.getPopular(1)));
|
||||
}
|
||||
|
||||
@Test
|
||||
void readingASavedGameFailsWhileTheDatabaseIsDown() {
|
||||
repository.failing = true;
|
||||
assertThrows(IllegalStateException.class, () -> history.get(UUID.randomUUID().toString()));
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,221 @@
|
||||
package app.playmancala.http;
|
||||
|
||||
import static io.restassured.RestAssured.given;
|
||||
import static org.awaitility.Awaitility.await;
|
||||
import static org.hamcrest.Matchers.matchesPattern;
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
import java.net.URI;
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
|
||||
import app.playmancala.game.PublicIds;
|
||||
import app.playmancala.model.Channels;
|
||||
import app.playmancala.model.Json;
|
||||
import app.playmancala.support.TestClient;
|
||||
import io.quarkus.test.common.http.TestHTTPResource;
|
||||
import io.quarkus.test.junit.QuarkusTest;
|
||||
|
||||
// The WebSocket and the HTTP API end to end, with Postgres and Valkey in Docker and a fake token verifier.
|
||||
// Ported from the Node backend's tests/server/ExpressApp.test.ts and tests/rtmt/rtmt_websocket.test.ts.
|
||||
@QuarkusTest
|
||||
class ApiTest {
|
||||
@TestHTTPResource("/")
|
||||
URI base;
|
||||
|
||||
URI webSocketUri(String query) {
|
||||
return URI.create(base.toString().replaceFirst("^http", "ws") + query);
|
||||
}
|
||||
|
||||
static String unique(String name) {
|
||||
return "kc:" + name + "-" + UUID.randomUUID().toString().substring(0, 8);
|
||||
}
|
||||
|
||||
static JsonNode get(String path, String userId) {
|
||||
var request = given();
|
||||
if (userId != null) request = request.header("Authorization", "Bearer valid:" + userId);
|
||||
String body = request.when().get(path).then().statusCode(200).extract().asString();
|
||||
try {
|
||||
return Json.parse(body);
|
||||
} catch (Exception e) {
|
||||
throw new IllegalStateException(body, e);
|
||||
}
|
||||
}
|
||||
|
||||
static List<String> ids(JsonNode games) {
|
||||
return games.valueStream().map(game -> game.get("id").asText()).toList();
|
||||
}
|
||||
|
||||
// Both ask for a game and get paired; returns the on_game_start message.
|
||||
static JsonNode startGame(TestClient first, TestClient second) {
|
||||
first.send(Channels.NEW_GAME, Map.of());
|
||||
second.send(Channels.NEW_GAME, Map.of());
|
||||
JsonNode start = first.waitFor(Channels.ON_GAME_START);
|
||||
assertEquals(start.get("id"), second.waitFor(Channels.ON_GAME_START).get("id"));
|
||||
return start;
|
||||
}
|
||||
|
||||
@Test
|
||||
void answersOnTheRootPathAndIssuesLegacyUserKeys() {
|
||||
given().when().get("/").then().statusCode(200);
|
||||
String uuid = "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}";
|
||||
given().when().get("/register/").then().statusCode(200).body(matchesPattern(uuid));
|
||||
given().when().get("/register").then().statusCode(200).body(matchesPattern(uuid));
|
||||
}
|
||||
|
||||
@Test
|
||||
void allowsEveryOrigin() {
|
||||
given().header("Origin", "https://playmancala.app").when().get("/games/popular").then().statusCode(200)
|
||||
.header("Access-Control-Allow-Origin", matchesPattern("\\*|https://playmancala\\.app"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void aGameFromStartToTheSavedRecord() {
|
||||
String alice = unique("alice");
|
||||
String bob = unique("bob");
|
||||
String carol = unique("carol");
|
||||
try (TestClient aliceClient = TestClient.connectAs(webSocketUri(""), alice);
|
||||
TestClient bobClient = TestClient.connectAs(webSocketUri(""), bob);
|
||||
TestClient carolClient = TestClient.connectAs(webSocketUri(""), carol)) {
|
||||
JsonNode start = startGame(aliceClient, bobClient);
|
||||
String gameId = start.get("id").asText();
|
||||
assertEquals("initial", start.at("/mancalaGame/state").asText());
|
||||
assertEquals(300_000, start.at("/clock/initialMs").asLong());
|
||||
|
||||
// A spectator, seen by the API but without spectator ids.
|
||||
carolClient.send(Channels.LISTEN_GAME_EVENTS, gameId);
|
||||
await().atMost(Duration.ofSeconds(5)).until(() -> get("/game/" + gameId, null).get("spectatorCount").asInt() == 1);
|
||||
JsonNode live = get("/game/" + gameId, null);
|
||||
assertEquals(gameId, live.get("id").asText());
|
||||
assertFalse(live.has("spectatorIds"));
|
||||
assertTrue(ids(get("/games/popular?limit=20", null).get("games")).contains(gameId));
|
||||
assertEquals(gameId, get("/games/me/active", alice).at("/game/id").asText());
|
||||
assertTrue(get("/games/me/active", carol).get("game").isNull());
|
||||
|
||||
// A move reaches the players and the spectator.
|
||||
String turnPlayerId = start.at("/mancalaGame/turnPlayerId").asText();
|
||||
(turnPlayerId.equals(alice) ? aliceClient : bobClient).send(Channels.GAME_MOVE, Map.of("index", 0));
|
||||
for (TestClient client : List.of(aliceClient, bobClient, carolClient)) {
|
||||
assertEquals(1, client.waitFor(Channels.ON_GAME_UPDATE).at("/mancalaGame/history").size());
|
||||
}
|
||||
assertEquals(1, get("/game/" + gameId, null).at("/mancalaGame/history").size());
|
||||
|
||||
// Alice leaves: Bob wins, and the game is saved.
|
||||
aliceClient.send(Channels.LEAVE_GAME, Map.of());
|
||||
JsonNode end = carolClient.waitFor(Channels.ON_GAME_END);
|
||||
assertEquals("left", end.get("endReason").asText());
|
||||
assertEquals(bob, end.get("winnerId").asText());
|
||||
assertEquals(alice, bobClient.waitFor(Channels.ON_GAME_USER_LEAVE).asText());
|
||||
|
||||
JsonNode finished = get("/game/" + gameId, null);
|
||||
assertEquals("finished", finished.get("status").asText());
|
||||
assertEquals(1, finished.at("/record/moves").size());
|
||||
assertEquals(alice, finished.at("/record/leftUserId").asText());
|
||||
assertTrue(get("/games/me/active", alice).get("game").isNull());
|
||||
await().atMost(Duration.ofSeconds(10)).until(() -> ids(get("/games/me", alice).get("games")).contains(gameId));
|
||||
assertEquals(List.of(gameId), ids(get("/games/me?limit=1", bob).get("games")));
|
||||
assertFalse(get("/games/me", bob).get("games").get(0).has("moves"));
|
||||
assertTrue(ids(get("/games/popular?kind=finished&limit=20", null).get("games")).contains(gameId));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsANewGameWhileOneIsInProgress() {
|
||||
try (TestClient alice = TestClient.connectAs(webSocketUri(""), unique("alice"));
|
||||
TestClient bob = TestClient.connectAs(webSocketUri(""), unique("bob"))) {
|
||||
String gameId = startGame(alice, bob).get("id").asText();
|
||||
alice.send(Channels.NEW_GAME, Map.of());
|
||||
assertEquals(gameId, alice.waitFor(Channels.ON_NEW_GAME_REJECTED).get("gameId").asText());
|
||||
alice.send(Channels.LEAVE_GAME, Map.of());
|
||||
bob.waitFor(Channels.ON_GAME_END);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void aReconnectingPlayerGetsTheGameAgain() {
|
||||
String alice = unique("alice");
|
||||
try (TestClient aliceClient = TestClient.connectAs(webSocketUri(""), alice);
|
||||
TestClient bob = TestClient.connectAs(webSocketUri(""), unique("bob"))) {
|
||||
String gameId = startGame(aliceClient, bob).get("id").asText();
|
||||
aliceClient.close();
|
||||
JsonNode offline = bob.waitFor(Channels.ON_USER_CONNECTION_CHANGE, 2).getLast();
|
||||
assertTrue(offline.toString().contains("\"isOnline\":false"), offline.toString());
|
||||
|
||||
try (TestClient again = TestClient.connectAs(webSocketUri(""), alice)) {
|
||||
assertEquals(gameId, again.waitFor(Channels.ON_GAME_UPDATE).get("id").asText());
|
||||
again.send(Channels.LEAVE_GAME, Map.of());
|
||||
bob.waitFor(Channels.ON_GAME_END);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void sendsLiveGamePreviewsToSubscribers() {
|
||||
try (TestClient viewer = TestClient.connectAs(webSocketUri(""), unique("viewer"));
|
||||
TestClient alice = TestClient.connectAs(webSocketUri(""), unique("alice"));
|
||||
TestClient bob = TestClient.connectAs(webSocketUri(""), unique("bob"))) {
|
||||
viewer.send(Channels.SUBSCRIBE_LIVE_GAMES, Map.of());
|
||||
viewer.waitFor(Channels.ON_LIVE_GAMES);
|
||||
String gameId = startGame(alice, bob).get("id").asText();
|
||||
await().atMost(Duration.ofSeconds(5)).until(() -> viewer.messages(Channels.ON_LIVE_GAMES).stream()
|
||||
.anyMatch(push -> ids(push.get("games")).contains(gameId)));
|
||||
alice.send(Channels.LEAVE_GAME, Map.of());
|
||||
bob.waitFor(Channels.ON_GAME_END);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void requiresAValidTokenForTheUsersGames() {
|
||||
given().when().get("/games/me/active").then().statusCode(401);
|
||||
given().header("Authorization", "Bearer forged").when().get("/games/me/active").then().statusCode(401);
|
||||
given().when().get("/games/me").then().statusCode(401);
|
||||
}
|
||||
|
||||
@Test
|
||||
void answers404ForUnknownGames() {
|
||||
given().when().get("/game/unknown").then().statusCode(404);
|
||||
given().when().get("/game/" + UUID.randomUUID()).then().statusCode(404);
|
||||
}
|
||||
|
||||
@Test
|
||||
void closesASocketWithAnInvalidToken() throws Exception {
|
||||
TestClient client = new TestClient(webSocketUri(""));
|
||||
client.send(Channels.AUTH, Map.of("token", "forged"));
|
||||
assertEquals(Json.tree(Map.of("reason", "invalid token")), client.waitFor(Channels.ON_AUTH_ERROR));
|
||||
assertEquals(Channels.CLOSE_AUTH_FAILED, client.closed.get(5, TimeUnit.SECONDS));
|
||||
}
|
||||
|
||||
@Test
|
||||
void playsWithLegacyUserKeyClients() {
|
||||
String key1 = UUID.randomUUID().toString();
|
||||
String key2 = UUID.randomUUID().toString();
|
||||
try (TestClient player1 = new TestClient(webSocketUri("?userKey=" + key1));
|
||||
TestClient player2 = new TestClient(webSocketUri("?userKey=" + key2))) {
|
||||
String gameId = startGame(player1, player2).get("id").asText();
|
||||
JsonNode preview = get("/games/popular?limit=20", null).get("games").valueStream()
|
||||
.filter(game -> game.get("id").asText().equals(gameId)).findFirst().orElseThrow();
|
||||
assertFalse(preview.toString().contains(key1));
|
||||
assertTrue(preview.toString().contains(PublicIds.publicUserId(key1)));
|
||||
player1.send(Channels.LEAVE_GAME, Map.of());
|
||||
player2.waitFor(Channels.ON_GAME_END);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void parsesLimitsLikeParseInt() {
|
||||
assertEquals(3, GameResource.parseLimit(null, 3));
|
||||
assertEquals(3, GameResource.parseLimit("abc", 3));
|
||||
assertEquals(7, GameResource.parseLimit("7abc", 3));
|
||||
assertEquals(1, GameResource.parseLimit("-4", 3));
|
||||
assertEquals(20, GameResource.parseLimit("99999999999999999999", 3));
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,324 @@
|
||||
package app.playmancala.rtmt;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
|
||||
import app.playmancala.auth.InvalidTokenException;
|
||||
import app.playmancala.auth.TokenVerifier;
|
||||
import app.playmancala.loop.ManualScheduler;
|
||||
import app.playmancala.model.AuthUser;
|
||||
import app.playmancala.model.Channels;
|
||||
import app.playmancala.model.Json;
|
||||
|
||||
// Ported from the Node backend's tests/rtmt/rtmt_websocket.test.ts, with fake sockets instead of a server.
|
||||
class RtmtServerTest {
|
||||
static final String LEGACY_KEY = "0b9f6d1e-7c1a-4f7e-9a51-3f0c2b6d8e11";
|
||||
static final long AUTH_TIMEOUT_MS = 300;
|
||||
|
||||
// Accepts tokens of the form "valid:<id>".
|
||||
static final TokenVerifier FAKE_VERIFIER = token -> {
|
||||
if (!token.startsWith("valid:")) throw new InvalidTokenException("bad token");
|
||||
String id = token.substring("valid:".length());
|
||||
return new AuthUser(id, "name-of-" + id, false);
|
||||
};
|
||||
|
||||
final ManualScheduler loop = new ManualScheduler(0);
|
||||
// Token checks wait here until the test runs them, as they would on a worker thread.
|
||||
final List<Runnable> pendingChecks = new ArrayList<>();
|
||||
final List<String> connectionChanges = new ArrayList<>();
|
||||
RtmtServer server;
|
||||
int socketCount;
|
||||
|
||||
class FakeSocket implements RtmtSocket {
|
||||
final String id = "socket-" + ++socketCount;
|
||||
final String query;
|
||||
final List<JsonNode> received = new ArrayList<>();
|
||||
boolean open = true;
|
||||
Integer closeCode;
|
||||
|
||||
FakeSocket(String query) {
|
||||
this.query = query;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String id() {
|
||||
return id;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String queryParam(String name) {
|
||||
return query != null && query.startsWith(name + "=") ? query.substring(name.length() + 1) : null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isOpen() {
|
||||
return open;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void send(String text) {
|
||||
try {
|
||||
received.add(Json.parse(text));
|
||||
} catch (Exception e) {
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
}
|
||||
|
||||
// Closing also delivers the close event, as the endpoint would.
|
||||
@Override
|
||||
public void close(int code, String reason) {
|
||||
if (!open) return;
|
||||
open = false;
|
||||
closeCode = code;
|
||||
server.onClose(this);
|
||||
}
|
||||
|
||||
void sendText(String text) {
|
||||
server.onText(this, text);
|
||||
runChecks();
|
||||
}
|
||||
|
||||
void send(String channel, Object message) {
|
||||
sendText(Json.write(Map.of("channel", channel, "message", message)));
|
||||
}
|
||||
|
||||
List<JsonNode> messages(String channel) {
|
||||
return received.stream().filter(m -> m.get("channel").asText().equals(channel)).map(m -> m.get("message")).toList();
|
||||
}
|
||||
}
|
||||
|
||||
void start(boolean legacyUserKeyAuth) {
|
||||
server = new RtmtServer(FAKE_VERIFIER, legacyUserKeyAuth, loop, pendingChecks::add, AUTH_TIMEOUT_MS);
|
||||
server.listenOnClientConnectionChange((id, isOnline) -> connectionChanges.add(id + (isOnline ? " online" : " offline")));
|
||||
server.start();
|
||||
}
|
||||
|
||||
void runChecks() {
|
||||
List<Runnable> checks = List.copyOf(pendingChecks);
|
||||
pendingChecks.clear();
|
||||
checks.forEach(Runnable::run);
|
||||
}
|
||||
|
||||
FakeSocket connect(String query) {
|
||||
FakeSocket socket = new FakeSocket(query);
|
||||
server.onOpen(socket);
|
||||
return socket;
|
||||
}
|
||||
|
||||
FakeSocket connectAs(String id) {
|
||||
FakeSocket socket = connect(null);
|
||||
socket.send(Channels.AUTH, Map.of("token", "valid:" + id));
|
||||
assertEquals(1, socket.messages(Channels.ON_AUTH).size());
|
||||
return socket;
|
||||
}
|
||||
|
||||
void disconnect(FakeSocket socket) {
|
||||
socket.open = false;
|
||||
server.onClose(socket);
|
||||
}
|
||||
|
||||
@Test
|
||||
void authenticatesASocketAndRoutesItsMessagesAsThatUser() {
|
||||
start(false);
|
||||
List<String> received = new ArrayList<>();
|
||||
server.listenMessage("hello", (id, message) -> received.add(id + " " + message));
|
||||
|
||||
FakeSocket socket = connect(null);
|
||||
socket.send(Channels.AUTH, Map.of("token", "valid:alice"));
|
||||
assertEquals(List.of(Json.tree(Map.of("user", Map.of("id", "alice", "name", "name-of-alice", "isGuest", false)))),
|
||||
socket.messages(Channels.ON_AUTH));
|
||||
|
||||
socket.send("hello", Map.of("n", 1));
|
||||
assertEquals(List.of("alice {\"n\":1}"), received);
|
||||
assertTrue(server.isClientOnline("alice"));
|
||||
assertEquals("name-of-alice", server.getUser("alice").name());
|
||||
assertEquals(List.of("alice online"), connectionChanges);
|
||||
}
|
||||
|
||||
@Test
|
||||
void sendsMessagesAsChannelAndMessage() {
|
||||
start(false);
|
||||
FakeSocket socket = connectAs("alice");
|
||||
server.sendMessage("alice", "news", "text");
|
||||
assertEquals(Json.tree(Map.of("channel", "news", "message", "text")), socket.received.getLast());
|
||||
assertEquals(List.of("channel", "message"), socket.received.getLast().propertyStream().map(Map.Entry::getKey).toList());
|
||||
}
|
||||
|
||||
@Test
|
||||
void ignoresMessagesBeforeAuthentication() {
|
||||
start(false);
|
||||
List<String> received = new ArrayList<>();
|
||||
server.listenMessage("hello", (id, message) -> received.add(id));
|
||||
connect(null).send("hello", Map.of());
|
||||
assertEquals(List.of(), received);
|
||||
}
|
||||
|
||||
@Test
|
||||
void closesTheSocketOnAnInvalidToken() {
|
||||
start(false);
|
||||
FakeSocket socket = connect(null);
|
||||
socket.send(Channels.AUTH, Map.of("token", "forged"));
|
||||
assertEquals(List.of(Json.tree(Map.of("reason", "invalid token"))), socket.messages(Channels.ON_AUTH_ERROR));
|
||||
assertEquals(Channels.CLOSE_AUTH_FAILED, socket.closeCode);
|
||||
}
|
||||
|
||||
@Test
|
||||
void closesTheSocketWhenATokenIsMissing() {
|
||||
start(false);
|
||||
FakeSocket socket = connect(null);
|
||||
socket.send(Channels.AUTH, Map.of());
|
||||
assertEquals(List.of(Json.tree(Map.of("reason", "missing token"))), socket.messages(Channels.ON_AUTH_ERROR));
|
||||
assertEquals(Channels.CLOSE_AUTH_FAILED, socket.closeCode);
|
||||
}
|
||||
|
||||
@Test
|
||||
void closesTheSocketWhenNoTokenArrivesInTime() {
|
||||
start(false);
|
||||
FakeSocket socket = connect(null);
|
||||
loop.advance(AUTH_TIMEOUT_MS - 1);
|
||||
assertTrue(socket.open);
|
||||
loop.advance(1);
|
||||
assertEquals(List.of(Json.tree(Map.of("reason", "authentication timeout"))), socket.messages(Channels.ON_AUTH_ERROR));
|
||||
assertEquals(Channels.CLOSE_AUTH_FAILED, socket.closeCode);
|
||||
}
|
||||
|
||||
@Test
|
||||
void keepsAnAuthenticatedSocketPastTheTimeout() {
|
||||
start(false);
|
||||
FakeSocket socket = connectAs("alice");
|
||||
loop.advance(AUTH_TIMEOUT_MS * 2);
|
||||
assertNull(socket.closeCode);
|
||||
}
|
||||
|
||||
@Test
|
||||
void acceptsARefreshedTokenForTheSameUserButNotForAnotherUser() {
|
||||
start(false);
|
||||
FakeSocket socket = connectAs("alice");
|
||||
socket.send(Channels.AUTH, Map.of("token", "valid:alice"));
|
||||
assertEquals(2, socket.messages(Channels.ON_AUTH).size());
|
||||
assertEquals(List.of("alice online"), connectionChanges);
|
||||
|
||||
socket.send(Channels.AUTH, Map.of("token", "valid:mallory"));
|
||||
assertEquals(List.of(Json.tree(Map.of("reason", "token belongs to another user"))),
|
||||
socket.messages(Channels.ON_AUTH_ERROR));
|
||||
assertEquals(Channels.CLOSE_AUTH_FAILED, socket.closeCode);
|
||||
assertFalse(server.isClientOnline("alice"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void ignoresATokenCheckThatEndsAfterTheSocketClosed() {
|
||||
start(false);
|
||||
FakeSocket socket = connect(null);
|
||||
server.onText(socket, Json.write(Map.of("channel", Channels.AUTH, "message", Map.of("token", "valid:alice"))));
|
||||
disconnect(socket);
|
||||
runChecks();
|
||||
assertFalse(server.isClientOnline("alice"));
|
||||
assertEquals(List.of(), connectionChanges);
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsUserKeyConnectionsWhenLegacyAuthIsOff() {
|
||||
start(false);
|
||||
FakeSocket socket = connect("userKey=" + LEGACY_KEY);
|
||||
assertEquals(Channels.CLOSE_AUTH_FAILED, socket.closeCode);
|
||||
}
|
||||
|
||||
@Test
|
||||
void survivesMalformedMessages() {
|
||||
start(false);
|
||||
FakeSocket socket = connectAs("alice");
|
||||
socket.sendText("{not json");
|
||||
socket.sendText("null");
|
||||
socket.sendText("[1]");
|
||||
socket.sendText("{\"channel\":5}");
|
||||
socket.send(Channels.AUTH, Map.of("token", "valid:alice"));
|
||||
assertEquals(2, socket.messages(Channels.ON_AUTH).size());
|
||||
}
|
||||
|
||||
@Test
|
||||
void deliversMessagesToEverySocketOfTheUser() {
|
||||
start(false);
|
||||
FakeSocket phone = connectAs("alice");
|
||||
FakeSocket laptop = connectAs("alice");
|
||||
server.sendMessage("alice", "news", Map.of("n", 1));
|
||||
assertEquals(List.of(Json.tree(Map.of("n", 1))), phone.messages("news"));
|
||||
assertEquals(List.of(Json.tree(Map.of("n", 1))), laptop.messages("news"));
|
||||
assertEquals(List.of("alice online", "alice online"), connectionChanges);
|
||||
}
|
||||
|
||||
@Test
|
||||
void keepsTheUserOnlineUntilTheLastSocketCloses() {
|
||||
start(false);
|
||||
FakeSocket phone = connectAs("alice");
|
||||
FakeSocket laptop = connectAs("alice");
|
||||
|
||||
disconnect(phone);
|
||||
assertTrue(server.isClientOnline("alice"));
|
||||
assertFalse(connectionChanges.contains("alice offline"));
|
||||
|
||||
disconnect(laptop);
|
||||
assertFalse(server.isClientOnline("alice"));
|
||||
assertNull(server.getUser("alice"));
|
||||
assertTrue(connectionChanges.contains("alice offline"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void doesNotDropAReconnectedSocketWhenTheOldOneClosesLate() {
|
||||
start(false);
|
||||
FakeSocket oldSocket = connectAs("alice");
|
||||
FakeSocket newSocket = connectAs("alice");
|
||||
disconnect(oldSocket);
|
||||
server.sendMessage("alice", "news", Map.of());
|
||||
assertEquals(List.of(Json.tree(Map.of())), newSocket.messages("news"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void pingsEverySocketAndClosesThoseThatDoNotAnswer() {
|
||||
start(false);
|
||||
FakeSocket answering = connectAs("alice");
|
||||
FakeSocket silent = connectAs("bob");
|
||||
loop.advance(RtmtServer.PING_INTERVAL_MS);
|
||||
assertEquals(1, answering.messages(Channels.PING).size());
|
||||
assertEquals(Json.tree(Map.of()), silent.messages(Channels.PING).getFirst());
|
||||
|
||||
answering.send(Channels.PONG, Map.of());
|
||||
loop.advance(RtmtServer.PING_INTERVAL_MS);
|
||||
assertNull(answering.closeCode);
|
||||
assertEquals(1001, silent.closeCode);
|
||||
assertFalse(server.isClientOnline("bob"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void acceptsAUuidUserKey() {
|
||||
start(true);
|
||||
connect("userKey=" + LEGACY_KEY);
|
||||
assertTrue(server.isClientOnline(LEGACY_KEY));
|
||||
assertEquals(new AuthUser(LEGACY_KEY, null, true), server.getUser(LEGACY_KEY));
|
||||
}
|
||||
|
||||
@Test
|
||||
void aUserKeyCannotClaimAKeycloakUsersId() {
|
||||
start(true);
|
||||
FakeSocket socket = connect("userKey=kc:" + LEGACY_KEY);
|
||||
assertEquals(Channels.CLOSE_AUTH_FAILED, socket.closeCode);
|
||||
}
|
||||
|
||||
@Test
|
||||
void aLegacySocketCannotSwitchToATokenIdentity() {
|
||||
start(true);
|
||||
FakeSocket socket = connect("userKey=" + LEGACY_KEY);
|
||||
socket.send(Channels.AUTH, Map.of("token", "valid:kc:alice"));
|
||||
assertEquals(Channels.CLOSE_AUTH_FAILED, socket.closeCode);
|
||||
assertFalse(server.isClientOnline(LEGACY_KEY));
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,21 @@
|
||||
package app.playmancala.support;
|
||||
|
||||
import app.playmancala.auth.InvalidTokenException;
|
||||
import app.playmancala.auth.TokenVerifier;
|
||||
import app.playmancala.model.AuthUser;
|
||||
import jakarta.annotation.Priority;
|
||||
import jakarta.enterprise.inject.Alternative;
|
||||
import jakarta.inject.Singleton;
|
||||
|
||||
// Replaces Keycloak in @QuarkusTests: accepts tokens of the form "valid:<user id>".
|
||||
@Alternative
|
||||
@Priority(1)
|
||||
@Singleton
|
||||
public class FakeTokenVerifier implements TokenVerifier {
|
||||
@Override
|
||||
public AuthUser verify(String token) throws InvalidTokenException {
|
||||
if (!token.startsWith("valid:")) throw new InvalidTokenException("bad token");
|
||||
String id = token.substring("valid:".length());
|
||||
return new AuthUser(id, "name-of-" + id, false);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,96 @@
|
||||
package app.playmancala.support;
|
||||
|
||||
import static org.awaitility.Awaitility.await;
|
||||
|
||||
import java.net.URI;
|
||||
import java.net.http.HttpClient;
|
||||
import java.net.http.WebSocket;
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.CompletionStage;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
|
||||
import app.playmancala.model.Channels;
|
||||
import app.playmancala.model.Json;
|
||||
|
||||
// A WebSocket client speaking the game protocol; answers pings like the real clients.
|
||||
public class TestClient implements WebSocket.Listener, AutoCloseable {
|
||||
public final List<JsonNode> messages = new CopyOnWriteArrayList<>();
|
||||
public final CompletableFuture<Integer> closed = new CompletableFuture<>();
|
||||
private final StringBuilder partial = new StringBuilder();
|
||||
private final WebSocket webSocket;
|
||||
|
||||
public TestClient(URI uri) {
|
||||
webSocket = HttpClient.newHttpClient().newWebSocketBuilder().buildAsync(uri, this).join();
|
||||
}
|
||||
|
||||
// Connects and logs in as the user, with a token the FakeTokenVerifier accepts.
|
||||
public static TestClient connectAs(URI uri, String userId) {
|
||||
TestClient client = new TestClient(uri);
|
||||
client.send(Channels.AUTH, Map.of("token", "valid:" + userId));
|
||||
client.waitFor(Channels.ON_AUTH);
|
||||
return client;
|
||||
}
|
||||
|
||||
public void send(String channel, Object message) {
|
||||
webSocket.sendText(Json.write(Map.of("channel", channel, "message", message)), true).join();
|
||||
}
|
||||
|
||||
public List<JsonNode> messages(String channel) {
|
||||
return messages.stream().filter(m -> m.get("channel").asText().equals(channel)).map(m -> m.get("message")).toList();
|
||||
}
|
||||
|
||||
// The first message on the channel, waiting up to 10 seconds for it.
|
||||
public JsonNode waitFor(String channel) {
|
||||
return waitFor(channel, 1).getFirst();
|
||||
}
|
||||
|
||||
public List<JsonNode> waitFor(String channel, int count) {
|
||||
await().atMost(Duration.ofSeconds(10)).until(() -> messages(channel).size() >= count);
|
||||
return messages(channel);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onOpen(WebSocket webSocket) {
|
||||
webSocket.request(1);
|
||||
}
|
||||
|
||||
@Override
|
||||
public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
|
||||
partial.append(data);
|
||||
if (last) {
|
||||
try {
|
||||
JsonNode message = Json.parse(partial.toString());
|
||||
if (message.get("channel").asText().equals(Channels.PING)) {
|
||||
webSocket.sendText(Json.write(Map.of("channel", Channels.PONG, "message", Map.of())), true);
|
||||
}
|
||||
messages.add(message);
|
||||
} catch (Exception e) {
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
partial.setLength(0);
|
||||
}
|
||||
webSocket.request(1);
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
|
||||
closed.complete(statusCode);
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(WebSocket webSocket, Throwable error) {
|
||||
closed.completeExceptionally(error);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
if (!webSocket.isOutputClosed()) webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "").join();
|
||||
}
|
||||
}
|
||||
@ -12,7 +12,7 @@ Work through the tasks in order; each ends with its tests passing and a commit.
|
||||
unit tests.
|
||||
- [x] 5. **Storage**: migrations and their runner, `PostgresGameHistoryRepository`, `GameHistoryService`,
|
||||
`ValkeyLiveGameStore`, `ValkeyCache`; integration tests with Dev Services.
|
||||
- [ ] 6. **WebSocket and HTTP**: `RtmtServer` with unit tests, `RtmtEndpoint`, the HTTP resources, `Backend`
|
||||
- [x] 6. **WebSocket and HTTP**: `RtmtServer` with unit tests, `RtmtEndpoint`, the HTTP resources, `Backend`
|
||||
wiring and startup, configuration; `@QuarkusTest`s for the API and the WebSocket.
|
||||
- [ ] 7. **Replace the Node backend**: move the project to `apps/backend`, Dockerfile, `.dockerignore`, compose,
|
||||
remove the Node sources; build the image.
|
||||
|
||||
Loading…
Reference in New Issue
Block a user