package app.playmancala.game; import java.util.Comparator; import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Set; import app.playmancala.loop.Cancellable; import app.playmancala.loop.Scheduler; import app.playmancala.model.Channels; import app.playmancala.model.Json; import app.playmancala.model.LiveGameSummary; import app.playmancala.rtmt.Rtmt; // Live games for the home page: most spectated first, then most recently active. Subscribers get the top games // when they subscribe and again whenever they change, at most once per second. Subscribing does not make anyone a // spectator. Only touched on the game loop. public class LiveGames { public static final int LIVE_GAMES_PUSH_COUNT = 5; private static final long PUSH_INTERVAL_MS = 1_000; private final Rtmt rtmt; private final GameStore gameStore; private final Scheduler scheduler; private final Set subscribers = new LinkedHashSet<>(); private String lastPushed = ""; private long lastPushAt; private Cancellable pushTimer; public LiveGames(Rtmt rtmt, GameStore gameStore, GameManager gameManager, Scheduler scheduler) { this.rtmt = rtmt; this.gameStore = gameStore; this.scheduler = scheduler; rtmt.listenMessage(Channels.SUBSCRIBE_LIVE_GAMES, (userId, message) -> subscribe(userId)); rtmt.listenMessage(Channels.UNSUBSCRIBE_LIVE_GAMES, (userId, message) -> subscribers.remove(userId)); rtmt.listenOnClientConnectionChange((userId, isOnline) -> { if (!isOnline) subscribers.remove(userId); }); gameManager.onLiveGamesChanged(this::schedulePush); } public static LiveGameSummary toLiveGameSummary(BackendGame game) { var mancalaGame = game.mancalaGame; return new LiveGameSummary(game.id, GameSerialization.publicSummary(game, mancalaGame.getPlayer1Id()), GameSerialization.publicSummary(game, mancalaGame.getPlayer2Id()), GameSerialization.boxed(mancalaGame.getBoard().getStoneArray()), PublicIds.publicUserId(mancalaGame.getTurnPlayerId()), game.spectatorIds.size(), mancalaGame.getHistory().size(), game.startedAt); } private static long lastActivity(BackendGame game) { return game.moveTimes.isEmpty() ? game.startedAt : game.moveTimes.getLast(); } public List getTop(int limit) { return gameStore.getAll().stream() .sorted(Comparator.comparingInt(game -> game.spectatorIds.size()) .thenComparingLong(LiveGames::lastActivity).reversed()) .limit(limit) .map(LiveGames::toLiveGameSummary) .toList(); } public void dispose() { if (pushTimer != null) pushTimer.cancel(); } private void subscribe(String userId) { subscribers.add(userId); rtmt.sendMessage(userId, Channels.ON_LIVE_GAMES, Map.of("games", getTop(LIVE_GAMES_PUSH_COUNT))); } private void schedulePush() { if (pushTimer != null) return; long wait = Math.max(0, lastPushAt + PUSH_INTERVAL_MS - scheduler.now()); pushTimer = scheduler.schedule(() -> { pushTimer = null; push(); }, wait); } private void push() { List games = getTop(LIVE_GAMES_PUSH_COUNT); String serialized = Json.write(games); if (serialized.equals(lastPushed)) return; lastPushed = serialized; lastPushAt = scheduler.now(); Map message = Map.of("games", games); List.copyOf(subscribers).forEach(userId -> rtmt.sendMessage(userId, Channels.ON_LIVE_GAMES, message)); } }