Fixed incorrect player disconnection #8, refactored managers communications

tfkfan · tfkfan · commit fd8a8eac70f4 · 2025-07-14T15:04:37.000+03:00
diff --git a/example/src/main/resources/application.yaml b/example/src/main/resources/application.yaml
@@ -6,4 +6,4 @@ room:
   loopRate: 50
   initDelay: 1
   startDelay: 5000
-  endDelay: 12000
+  endDelay: 120000
diff --git a/orbital-core/src/main/java/io/github/tfkfan/orbital/core/OrbitalBuilderImpl.java b/orbital-core/src/main/java/io/github/tfkfan/orbital/core/OrbitalBuilderImpl.java
@@ -55,11 +55,15 @@ public OrbitalBuilderImpl withConfig(OrbitalConfig orbitalConfig) {
         return this;
     }
 
-    public OrbitalBuilderImpl withConfig(String fullPath) {
-        configSupplier = () -> loadLocalConfig(fullPath).map(cnf -> new Pair<>(cnf, cnf.mapTo(OrbitalConfig.class)));
+    public <C extends OrbitalConfig> OrbitalBuilderImpl withConfig(String fullPath, Class<C> configClazz) {
+        configSupplier = () -> loadLocalConfig(fullPath).map(cnf -> new Pair<>(cnf, cnf.mapTo(configClazz)));
         return this;
     }
 
+    public OrbitalBuilderImpl withConfig(String fullPath) {
+        return withConfig(fullPath, OrbitalConfig.class);
+    }
+
     public OrbitalBuilderImpl withLocalConfig() {
         return withConfig(DEFAULT_CONFIG_PATH);
     }
diff --git a/orbital-core/src/main/java/io/github/tfkfan/orbital/core/configuration/Fields.java b/orbital-core/src/main/java/io/github/tfkfan/orbital/core/configuration/Fields.java
@@ -7,6 +7,7 @@ public interface Fields {
     String roomType = "roomType";
     String sessions = "sessions";
     String sessionId = "sessionId";
+    String success = "success";
     String admin = "admin";
     String initialData = "initialData";
     String type = "type";
diff --git a/orbital-core/src/main/java/io/github/tfkfan/orbital/core/manager/BaseGatewayManager.java b/orbital-core/src/main/java/io/github/tfkfan/orbital/core/manager/BaseGatewayManager.java
@@ -1,12 +1,14 @@
 package io.github.tfkfan.orbital.core.manager;
 
+import io.vertx.core.Vertx;
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
 public abstract class BaseGatewayManager extends BaseManager implements GatewayManager {
     protected final MatchmakerManager matchmakerManager;
 
-    public BaseGatewayManager(MatchmakerManager matchmakerManager) {
+    protected BaseGatewayManager(Vertx vertx,MatchmakerManager matchmakerManager) {
+        super(vertx);
         this.matchmakerManager = matchmakerManager;
     }
 }
diff --git a/orbital-core/src/main/java/io/github/tfkfan/orbital/core/manager/BaseManager.java b/orbital-core/src/main/java/io/github/tfkfan/orbital/core/manager/BaseManager.java
@@ -1,4 +1,56 @@
 package io.github.tfkfan.orbital.core.manager;
 
+import io.github.tfkfan.orbital.core.configuration.Constants;
+import io.github.tfkfan.orbital.core.configuration.Fields;
+import io.github.tfkfan.orbital.core.room.RoomType;
+import io.github.tfkfan.orbital.core.shared.ActionType;
+import io.vertx.core.Future;
+import io.vertx.core.Vertx;
+import io.vertx.core.eventbus.EventBus;
+import io.vertx.core.eventbus.Message;
+import io.vertx.core.json.JsonArray;
+import io.vertx.core.json.JsonObject;
+
+import java.util.Objects;
+import java.util.UUID;
+
+import static io.github.tfkfan.orbital.core.verticle.BaseVerticle.defaults;
+
 public abstract class BaseManager implements Manager {
+    protected final Vertx vertx;
+    protected final EventBus eventBus;
+
+    protected BaseManager(Vertx vertx) {
+        this.vertx = vertx;
+        eventBus = vertx.eventBus();
+    }
+
+    public Future<Message<JsonObject>> requestRoomManagementEvent(final UUID roomId,
+                                                                  final ActionType actionType,
+                                                                  final RoomType roomType,
+                                                                  final JsonArray userSessions) {
+        return eventBus.request(Constants.ROOM_VERTICAL_CHANNEL, makeData(roomId, actionType, roomType, userSessions), defaults());
+    }
+
+    public void publishRoomManagementEvent(final UUID roomId,
+                                           final ActionType actionType,
+                                           final RoomType roomType,
+                                           final JsonArray userSessions) {
+        eventBus.publish(Constants.ROOM_VERTICAL_CHANNEL, makeData(roomId, actionType, roomType, userSessions), defaults());
+    }
+
+    private JsonObject makeData(final UUID roomId,
+                                final ActionType actionType,
+                                final RoomType roomType,
+                                final JsonArray userSessions) {
+        final JsonObject data = new JsonObject()
+                .put(Fields.action, actionType)
+                .put(Fields.roomId, roomId.toString())
+                .put(Fields.sessions, userSessions);
+
+        if (Objects.nonNull(roomType))
+            data.put(Fields.roomType, roomType.toString());
+
+        return data;
+    }
 }
diff --git a/orbital-core/src/main/java/io/github/tfkfan/orbital/core/manager/BaseMatchmakerManager.java b/orbital-core/src/main/java/io/github/tfkfan/orbital/core/manager/BaseMatchmakerManager.java
@@ -6,34 +6,35 @@
 import io.github.tfkfan.orbital.core.configuration.props.RoomConfig;
 import io.github.tfkfan.orbital.core.event.*;
 import io.github.tfkfan.orbital.core.network.message.Message;
+import io.github.tfkfan.orbital.core.room.GameRoom;
 import io.github.tfkfan.orbital.core.room.RoomType;
 import io.github.tfkfan.orbital.core.route.MessageRoute;
 import io.github.tfkfan.orbital.core.session.GatewaySession;
 import io.github.tfkfan.orbital.core.shared.ActionType;
 import io.github.tfkfan.orbital.core.shared.UniqueQueue;
+import io.vertx.core.Future;
 import io.vertx.core.Vertx;
-import io.vertx.core.eventbus.DeliveryOptions;
 import io.vertx.core.eventbus.EventBus;
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 import java.util.UUID;
 import java.util.stream.Collectors;
 
+import static io.github.tfkfan.orbital.core.verticle.BaseVerticle.defaults;
+
 @Slf4j
 public abstract class BaseMatchmakerManager extends BaseManager implements MatchmakerManager {
-    protected final Vertx vertx;
-    protected final EventBus eventBus;
     protected final RoomConfig roomConfig;
     private final UniqueQueue<GameRoomJoinEvent> playersQueue = new UniqueQueue<>();
 
-    public BaseMatchmakerManager(RoomConfig roomConfig) {
+    public BaseMatchmakerManager(Vertx vertx, RoomConfig roomConfig) {
+        super(vertx);
         this.roomConfig = roomConfig;
-        vertx = Vertx.currentContext().owner();
-        eventBus = vertx.eventBus();
         initConsumers();
     }
 
@@ -47,6 +48,9 @@ protected void initConsumers() {
     @Override
     public void onDisconnect(GatewaySession data) {
         playersQueue.removeIf(e -> e.getSession().getId().equals(data.getId()));
+        publishRoomManagementEvent(data.getRoomKey(), ActionType.PLAYER_DISCONNECT, null, new JsonArray(
+                Collections.singletonList(new JsonObject().put(Fields.sessionId, data.getId()))
+        ));
     }
 
     @Override
@@ -66,7 +70,7 @@ protected void onJoin(GameRoomJoinEvent joinEvent) {
     }
 
     protected void handleJoinTraining(final GameRoomJoinEvent joinEvent) {
-        newRoomEvent(newRoomId(), RoomType.TRAINING, Collections.singletonList(joinEvent));
+        requestRoomManagementCreateEvent(newRoomId(), RoomType.TRAINING, Collections.singletonList(joinEvent));
     }
 
     protected void handleJoin(final RoomType roomType, final GameRoomJoinEvent joinEvent) {
@@ -77,27 +81,20 @@ protected void handleJoin(final RoomType roomType, final GameRoomJoinEvent joinE
         if (playersQueue.size() < roomConfig.getMaxPlayers())
             return;
 
-        newRoomEvent(newRoomId(), roomType, playersQueue.chunk(roomConfig.getMaxPlayers()));
+        requestRoomManagementCreateEvent(newRoomId(), roomType, playersQueue.chunk(roomConfig.getMaxPlayers()));
     }
 
     private UUID newRoomId() {
         return UUID.randomUUID();
     }
 
-    protected void newRoomEvent(final UUID roomId, final RoomType roomType, final List<GameRoomJoinEvent> userSessions) {
-        eventBus.sender(Constants.ROOM_VERTICAL_CHANNEL, new DeliveryOptions()
-                        .setLocalOnly(true)
-                        .setSendTimeout(1000))
-                .write(new JsonObject()
-                        .put(Fields.action, ActionType.NEW_ROOM)
-                        .put(Fields.roomId, roomId.toString())
-                        .put(Fields.roomType, roomType.toString())
-                        .put(Fields.sessions, new JsonArray(userSessions.stream().map(e ->
-                                        new JsonObject()
-                                                .put(Fields.sessionId, e.getSession().getId())
-                                                .put(Fields.admin, e.getSession().isAdmin())
-                                                .put(Fields.initialData, e.getData()))
-                                .collect(Collectors.toList()))))
+    protected void requestRoomManagementCreateEvent(final UUID roomId, final RoomType roomType, List<GameRoomJoinEvent> userSessions) {
+        requestRoomManagementEvent(roomId, ActionType.NEW_ROOM, roomType, new JsonArray(userSessions.stream().map(e ->
+                        new JsonObject()
+                                .put(Fields.sessionId, e.getSession().getId())
+                                .put(Fields.admin, e.getSession().isAdmin())
+                                .put(Fields.initialData, e.getData()))
+                .collect(Collectors.toList())))
                 .onSuccess(t -> userSessions.forEach(it -> it.getSession().setRoomKey(roomId)))
                 .onFailure(t -> {
                     log.error("Room verticle {} failed to join", roomId, t);
diff --git a/orbital-core/src/main/java/io/github/tfkfan/orbital/core/manager/WebSocketManager.java b/orbital-core/src/main/java/io/github/tfkfan/orbital/core/manager/WebSocketManager.java
@@ -3,5 +3,5 @@
 import io.vertx.core.Handler;
 import io.vertx.core.http.ServerWebSocket;
 
-public interface WebSocketManager extends GatewayManager, Handler<ServerWebSocket>,Manager  {
+public interface WebSocketManager extends GatewayManager, Handler<ServerWebSocket>, Manager {
 }
diff --git a/orbital-core/src/main/java/io/github/tfkfan/orbital/core/manager/impl/GameManagerImpl.java b/orbital-core/src/main/java/io/github/tfkfan/orbital/core/manager/impl/GameManagerImpl.java
@@ -55,22 +55,38 @@ public GameManagerImpl(String verticleId,
     }
 
     protected void onMessage(Message<JsonObject> message) {
-        log.info("Received a message to room {} : {}", verticleId, message.body().encode());
+        log.info("Received a message at room verticle {} : {}", verticleId, message.body().encode());
         final JsonObject json = message.body();
         final String rawAction = json.getString(Fields.action);
-        if (rawAction != null) {
-            ActionType actionType = ActionType.valueOf(rawAction);
-            if (actionType.equals(ActionType.NEW_ROOM)) {
+        if (rawAction != null)
+            onRoom(ActionType.valueOf(rawAction), message, json);
+    }
+
+    protected void onRoom(ActionType actionType, Message<JsonObject> message, JsonObject json) {
+        switch (actionType) {
+            case NEW_ROOM -> {
                 final S gameState = gameStateFactory.get();
                 final RoomType roomType = RoomType.valueOf(json.getString(Fields.roomType));
                 final UUID roomId = UUID.fromString(json.getString(Fields.roomId));
-                final GameRoom room = createRoom(roomId, roomType, gameState, json.getJsonArray(Fields.sessions));
+                final GameRoom room = onNewRoom(roomId, roomType, gameState, json.getJsonArray(Fields.sessions));
                 room.onStart();
             }
+            case PLAYER_DISCONNECT -> {
+                final List<String> sessionsIds = json.getJsonArray(Fields.sessions).stream().map(it -> ((JsonObject) it).getString(Fields.sessionId)).toList();
+                sessionsIds.forEach(sessionId -> {
+                    try {
+                        final PlayerSession session = playerSessionsMap.remove(sessionId);
+                        if (session != null)
+                            session.getPlayer().getGameRoom().onDisconnect(session);
+                    } catch (Exception ignored) {
+                    }
+                });
+            }
         }
+        message.reply(new JsonObject().put(Fields.success, true));
     }
 
-    protected GameRoom createRoom(final UUID roomId, final RoomType roomType, final S gameState, JsonArray playersSessions) {
+    protected GameRoom onNewRoom(final UUID roomId, final RoomType roomType, final S gameState, JsonArray playersSessions) {
         validatePlayersCount(roomType, playersSessions, roomId);
 
         final GameRoom room = gameRoomFactory.createGameRoom(verticleId, roomId, roomType,
diff --git a/orbital-core/src/main/java/io/github/tfkfan/orbital/core/manager/impl/MatchmakerManagerImpl.java b/orbital-core/src/main/java/io/github/tfkfan/orbital/core/manager/impl/MatchmakerManagerImpl.java
@@ -2,11 +2,12 @@
 
 import io.github.tfkfan.orbital.core.configuration.props.RoomConfig;
 import io.github.tfkfan.orbital.core.manager.BaseMatchmakerManager;
+import io.vertx.core.Vertx;
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
 public class MatchmakerManagerImpl extends BaseMatchmakerManager {
     public MatchmakerManagerImpl(RoomConfig roomConfig) {
-        super(roomConfig);
+        super(Vertx.currentContext().owner(), roomConfig);
     }
 }
diff --git a/orbital-core/src/main/java/io/github/tfkfan/orbital/core/manager/impl/WebSocketManagerImpl.java b/orbital-core/src/main/java/io/github/tfkfan/orbital/core/manager/impl/WebSocketManagerImpl.java
@@ -20,11 +20,10 @@
 
 @Slf4j
 public class WebSocketManagerImpl extends BaseGatewayManager implements WebSocketManager {
-    protected final Vertx vertx = Vertx.currentContext().owner();
     protected final RouteProcessor routeProcessor;
 
     public WebSocketManagerImpl(MatchmakerManager matchmakerManager) {
-        super(matchmakerManager);
+        super(Vertx.currentContext().owner(), matchmakerManager);
         this.routeProcessor = new RouteProcessor(matchmakerManager);
     }
 
@@ -74,7 +73,7 @@ private void onMessage(GatewaySession userSession, JsonObject message) {
 
     @Override
     public void broadcast(MessageType type, String message) {
-        log.error("broadcast not implemented yet");
+        broadcast(type.getType(), message);
     }
 
     @Override
@@ -89,6 +88,7 @@ public void broadcast(int messageType, Object content) {
 
     @Override
     public void broadcast(Function<Session, JsonObject> messageFunction) {
+
     }
 
     @Override
diff --git a/orbital-core/src/main/java/io/github/tfkfan/orbital/core/room/AbstractGameRoom.java b/orbital-core/src/main/java/io/github/tfkfan/orbital/core/room/AbstractGameRoom.java
@@ -126,6 +126,8 @@ protected void onPlayerInitRequest(PlayerSession playerSession, InitPlayerEvent
     @Override
     public <E extends Event> void addEventListener(EventListener<E> listener, Class<E> clazz, boolean startCheck) {
         consumerList.add(vertx.eventBus().localConsumer(GameRoom.constructEventListenerConsumer(gameRoomId, clazz), event -> {
+            log.trace("Event received: {}", event.address());
+
             final String playerSessionId = event.headers().get(Fields.sessionId);
             if (!sessions.containsKey(playerSessionId) || startCheck && !started)
                 return;
@@ -142,28 +144,34 @@ public <E extends Event> void addEventListener(EventListener<E> listener, Class<
 
     @Override
     public void onCreate() {
+        log.trace("Room {} create call", key());
+
         vertx.eventBus().publish(Constants.MATCHMAKER_ROOM_CREATE_CHANNEL, new JsonObject()
                 .put(Fields.roomId, key().toString()));
         gameRoomMetricsRegistrar.register();
         gameManager.onCreate(this);
+
         log.debug("Room {} has been created", key());
     }
 
     @Override
     public void onStart() {
+        log.trace("Room {} start call", key());
+
         schedule(config.getEndDelay() + config.getStartDelay(), (t) -> onBattleEnd());
         schedule(config.getStartDelay(), (t) -> onBattleStart());
         broadcast(MessageTypes.GAME_ROOM_START, new GameRoomInfoPack(
                 OffsetDateTime.now().plus(config.getStartDelay(), ChronoUnit.MILLIS).toInstant().toEpochMilli()
         ));
 
         gameManager.onStart(this);
-        log.debug("Room {} has been started", key());
+        log.debug("Room {} just started", key());
     }
 
     @Override
     public void onBattleStart() {
-        log.debug("Room {} battle has been started", key());
+        log.trace("Room {} battle start call", key());
+
         started = true;
         gameManager.onBattleStart(this);
 
@@ -178,12 +186,14 @@ public void onBattleStart() {
 
     @Override
     public void onBattleEnd() {
-        log.debug("Room {}. Battle has been ended", key());
+        log.trace("Room {} onBattleEnd call", key());
         gameManager.onBattleEnd(this);
     }
 
     @Override
     public void onDestroy() {
+        log.trace("Room {} destroy call", key());
+
         gameRoomMetricsRegistrar.unregister();
 
         sessions().forEach(playerSession -> state.removePlayer(playerSession.getPlayer()));
@@ -202,18 +212,24 @@ public void onDestroy() {
 
     @Override
     public void onJoin(PlayerSession playerSession) {
+        log.trace("Player session {} joined at room {}", playerSession.getId(), key());
+
         this.sessions.put(playerSession.getId(), playerSession);
         playerSession.send(MessageTypes.GAME_ROOM_JOIN_SUCCESS, new GameSettingsPack(config.getLoopRate()));
     }
 
     @Override
     public void onRejoin(PlayerSession playerSession, UUID reconnectKey) {
+        log.trace("Player session {} rejoined at room {}. Reconnect key {}", playerSession.getId(), key(), reconnectKey);
+
         sessions.put(playerSession.getId(), playerSession);
         playerSession.send(MessageTypes.GAME_ROOM_JOIN_SUCCESS, new GameSettingsPack(config.getLoopRate()));
     }
 
     @Override
     public PlayerSession onDisconnect(PlayerSession playerSession) {
+        log.trace("Player session {} disconnected at room {}", playerSession.getId(), key());
+
         state.removePlayer(playerSession.getPlayer());
         return sessions.remove(playerSession.getId());
     }
diff --git a/orbital-core/src/main/java/io/github/tfkfan/orbital/core/shared/ActionType.java b/orbital-core/src/main/java/io/github/tfkfan/orbital/core/shared/ActionType.java
@@ -1,5 +1,5 @@
 package io.github.tfkfan.orbital.core.shared;
 
 public enum ActionType {
-    NEW_ROOM
+    NEW_ROOM, CLOSE_ROOM, PLAYER_DISCONNECT
 }
diff --git a/orbital-core/src/main/java/io/github/tfkfan/orbital/core/verticle/BaseVerticle.java b/orbital-core/src/main/java/io/github/tfkfan/orbital/core/verticle/BaseVerticle.java

Original file line number	Diff line number	Diff line change
`@@ -55,11 +55,15 @@ public OrbitalBuilderImpl withConfig(OrbitalConfig orbitalConfig) {`
`55`	`55`	`return this;`
`56`	`56`	`}`
`57`	`57`
`58`		`- public OrbitalBuilderImpl withConfig(String fullPath) {`
`59`		`- configSupplier = () -> loadLocalConfig(fullPath).map(cnf -> new Pair<>(cnf, cnf.mapTo(OrbitalConfig.class)));`
	`58`	`+ public <C extends OrbitalConfig> OrbitalBuilderImpl withConfig(String fullPath, Class<C> configClazz) {`
	`59`	`+ configSupplier = () -> loadLocalConfig(fullPath).map(cnf -> new Pair<>(cnf, cnf.mapTo(configClazz)));`
`60`	`60`	`return this;`
`61`	`61`	`}`
`62`	`62`
	`63`	`+ public OrbitalBuilderImpl withConfig(String fullPath) {`
	`64`	`+ return withConfig(fullPath, OrbitalConfig.class);`
	`65`	`+ }`
	`66`	`+`
`63`	`67`	`public OrbitalBuilderImpl withLocalConfig() {`
`64`	`68`	`return withConfig(DEFAULT_CONFIG_PATH);`
`65`	`69`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,12 +1,14 @@`
`1`	`1`	`package io.github.tfkfan.orbital.core.manager;`
`2`	`2`
	`3`	`+import io.vertx.core.Vertx;`
`3`	`4`	`import lombok.extern.slf4j.Slf4j;`
`4`	`5`
`5`	`6`	`@Slf4j`
`6`	`7`	`public abstract class BaseGatewayManager extends BaseManager implements GatewayManager {`
`7`	`8`	`protected final MatchmakerManager matchmakerManager;`
`8`	`9`
`9`		`- public BaseGatewayManager(MatchmakerManager matchmakerManager) {`
	`10`	`+ protected BaseGatewayManager(Vertx vertx,MatchmakerManager matchmakerManager) {`
	`11`	`+ super(vertx);`
`10`	`12`	`this.matchmakerManager = matchmakerManager;`
`11`	`13`	`}`
`12`	`14`	`}`
Original file line number	Diff line number	Diff line change
`@@ -3,5 +3,5 @@`
`3`	`3`	`import io.vertx.core.Handler;`
`4`	`4`	`import io.vertx.core.http.ServerWebSocket;`
`5`	`5`
`6`		`-public interface WebSocketManager extends GatewayManager, Handler<ServerWebSocket>,Manager {`
	`6`	`+public interface WebSocketManager extends GatewayManager, Handler<ServerWebSocket>, Manager {`
`7`	`7`	`}`
Original file line number	Diff line number	Diff line change
`@@ -2,11 +2,12 @@`
`2`	`2`
`3`	`3`	`import io.github.tfkfan.orbital.core.configuration.props.RoomConfig;`
`4`	`4`	`import io.github.tfkfan.orbital.core.manager.BaseMatchmakerManager;`
	`5`	`+import io.vertx.core.Vertx;`
`5`	`6`	`import lombok.extern.slf4j.Slf4j;`
`6`	`7`
`7`	`8`	`@Slf4j`
`8`	`9`	`public class MatchmakerManagerImpl extends BaseMatchmakerManager {`
`9`	`10`	`public MatchmakerManagerImpl(RoomConfig roomConfig) {`
`10`		`- super(roomConfig);`
	`11`	`+ super(Vertx.currentContext().owner(), roomConfig);`
`11`	`12`	`}`
`12`	`13`	`}`