[#37] chore : merge pull request #54 from Game-as-a-Service/feature/b…

…ackend/socketio/create-game

Hi @wingtkw ,
As I  described above
I will merge the PR first , you are still welcome to leave comment at here,
or create another issue / discussion thread for any question about learning my code at here

backend/app/adapter/api.py

@@ -6,7 +6,6 @@
 from fastapi.responses import JSONResponse
 from hypercorn.config import Config
 from hypercorn.asyncio import serve
-from app.config import LOG_FILE_PATH
 
 from app.dto import (
     CreateGameReqDto,
@@ -22,9 +21,10 @@
     SwitchInvestigatorUseCase,
     UpdateGameDifficultyUseCase,
 )
-from app.config import REST_HOST, REST_PORT
+from app.config import LOG_FILE_PATH, REST_HOST, REST_PORT
 from app.domain import GameError
-from app.adapter.repository import get_repository
+from app.adapter.event_emitter import SocketIoEventEmitter
+from app.adapter.repository import get_game_repository
 from app.adapter.presenter import read_investigator_presenter, create_game_presenter
 
 _logger = logging.getLogger(__name__)
@@ -54,7 +54,9 @@ def __init__(self, e: GameError):
 @_router.post("/games", response_model=CreateGameRespDto)
 async def create_game(req: CreateGameReqDto):
     uc = CreateGameUseCase(
-        repository=shared_context["repository"], settings=shared_context["settings"]
+        repository=shared_context["repository"],
+        evt_emitter=shared_context["evt_emit"],
+        settings=shared_context["settings"],
     )
     response = await uc.execute(req.players, create_game_presenter)
     return response
@@ -96,11 +98,14 @@ async def update_game_difficulty(game_id: str, req: UpdateDifficultyDto):
 async def lifetime_server_context(app: FastAPI):
     # TODO, parameters should be in separate python module or `json` , `toml` file
     settings = {"host": "localhost:8081"}
-    shared_context["repository"] = get_repository()
+    shared_context["repository"] = get_game_repository()
     shared_context["settings"] = settings
+    shared_context["evt_emit"] = SocketIoEventEmitter()
     yield
     ## TODO, de-initialize the context if necessary,
     # e.g. close database connections
+    evt_emit = shared_context.pop("evt_emit")
+    await evt_emit.deinit()
     shared_context.clear()
 
 

backend/app/adapter/event_emitter.py

@@ -0,0 +1,38 @@
+import logging
+import socketio
+
+from app.config import LOG_FILE_PATH, RTC_HOST, RTC_PORT
+from app.constant import RealTimeCommConst as RtcConst
+from app.domain import Game
+
+_logger = logging.getLogger(__name__)
+_logger.setLevel(logging.WARNING)
+_logger.addHandler(logging.FileHandler(LOG_FILE_PATH["REST"], mode="a"))
+
+
+class AbsEventEmitter:
+    async def create_game(self, game: Game):
+        raise NotImplementedError("AbsEventEmitter.create_game")
+
+
+class SocketIoEventEmitter(AbsEventEmitter):
+    def __init__(self):
+        self._url = "http://%s:%s" % (RTC_HOST, RTC_PORT)
+        self._namespace = RtcConst.NAMESPACE
+        self._client = socketio.AsyncSimpleClient(logger=True)
+
+    async def create_game(self, game: Game):
+        data = {"gameID": game.id, "players": [p.id for p in game.players]}
+        try:
+            if not self._client.connected:
+                await self._client.connect(self._url, namespace=self._namespace)
+            await self._client.emit(RtcConst.EVENTS.NEW_ROOM.value, data=data)
+        except Exception as e:
+            _logger.error("send new-room event: %s", e)
+            # TODO, reconnect if connection inactive
+
+    async def deinit(self):
+        try:
+            await self._client.disconnect()
+        except Exception as e:
+            _logger.error("error on deinit: %s", e)

backend/app/adapter/repository/__init__.py

@@ -1,18 +1,35 @@
+from typing import Optional
+
 from app.domain import Game
+from app.dto import RtcRoomMsgData
 
 
-class AbstractRepository:
+class AbstractGameRepository:
     async def save(self, game: Game):
-        raise NotImplementedError("AbstractRepository.save")
+        raise NotImplementedError("AbstractGameRepository.save")
 
     async def get_game(self, game_id: str) -> Game:
-        raise NotImplementedError("AbstractRepository.get_game")
+        raise NotImplementedError("AbstractGameRepository.get_game")
+
+
+class AbstractRtcRoomRepository:
+    async def save(self, data: RtcRoomMsgData):
+        raise NotImplementedError("AbstractRtcRoomRepository.save")
+
+    async def get(self, game_id: str) -> Optional[RtcRoomMsgData]:
+        raise NotImplementedError("AbstractRtcRoomRepository.get")
 
 
-def get_repository():
+def get_game_repository():
     # TODO
     # - make it configurable at runtime
     # - set max limit of concurrent and ongoing games to save
-    from .in_mem import InMemoryRepository
+    from .in_mem import InMemoryGameRepository
+
+    return InMemoryGameRepository()
+
+
+def get_rtc_room_repository():
+    from .in_mem import InMemoryRtcRoomRepository
 
-    return InMemoryRepository()
+    return InMemoryRtcRoomRepository()

backend/app/adapter/repository/in_mem.py

@@ -2,10 +2,11 @@
 from typing import Optional
 
 from app.domain import Game
-from app.adapter.repository import AbstractRepository
+from app.adapter.repository import AbstractGameRepository, AbstractRtcRoomRepository
+from app.dto import RtcRoomMsgData
 
 
-class InMemoryRepository(AbstractRepository):
+class InMemoryGameRepository(AbstractGameRepository):
     def __init__(self):
         self._lock = asyncio.Lock()
         self._games = {}
@@ -19,3 +20,17 @@ async def save(self, game: Game):
     async def get_game(self, game_id: str) -> Optional[Game]:
         async with self._lock:
             return self._games.get(game_id)
+
+
+class InMemoryRtcRoomRepository(AbstractRtcRoomRepository):
+    def __init__(self):
+        self._lock = asyncio.Lock()
+        self._rooms = {}
+
+    async def save(self, data: RtcRoomMsgData):
+        async with self._lock:
+            self._rooms[data.gameID] = data
+
+    async def get(self, game_id: str) -> Optional[RtcRoomMsgData]:
+        async with self._lock:
+            return self._rooms.get(game_id)

backend/app/adapter/sio_srv.py

@@ -6,33 +6,25 @@
 import socketio
 from hypercorn.config import Config
 from hypercorn.asyncio import serve
-from pydantic import BaseModel, ConfigDict, ValidationError
+from pydantic import ValidationError
 
 from app.constant import RealTimeCommConst as RtcConst
 from app.config import RTC_HOST, RTC_PORT, LOG_FILE_PATH
+from app.adapter.repository import get_rtc_room_repository
+from app.dto import RtcRoomMsgData, ChatMsgData, RtcInitMsgData
+from app.domain import GameError, GameErrorCodes, GameFuncCodes
 
 srv = socketio.AsyncServer(async_mode="asgi")
-# currently the logger is configured in simple way, if someone needs to run it
-# in production environment, maybe they can switch to more advanced architecture
-# e.g. centralized logging architecture, ELK stack, EFK stack ...etc
+# currently the logger is configured in simple way, anyone who needs to run it
+# in production environment can switch to more advanced architecture e.g.
+# centralized logging architecture, ELK stack, EFK stack ...etc
 _logger = logging.getLogger(__name__)
 _logger.setLevel(logging.WARNING)
 _logger.addHandler(logging.FileHandler(LOG_FILE_PATH["RTC"], mode="a"))
 
-
-class ChatMsgData(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-    msg: str
-    nickname: str
-    gameID: str
-    client: str  ## client session ID
-
-
-class RtcInitMsgData(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-    nickname: str
-    gameID: str
-    client: str
+# there is no use-case layer in this socketio server, all the game logic should
+# be implemented in the http sserver
+_repo = get_rtc_room_repository()
 
 
 @srv.on(RtcConst.EVENTS.CHAT.value, namespace=RtcConst.NAMESPACE)
@@ -53,22 +45,46 @@ async def _forward_chat_msg(sid, data: Dict):
         )
 
 
+async def check_room_exist(repo, data: RtcInitMsgData):
+    fetched_room = await repo.get(data.gameID)
+    if fetched_room is None or fetched_room.gameID != data.gameID:
+        ecode = GameErrorCodes.GAME_NOT_FOUND
+    elif data.player.id not in fetched_room.players:
+        ecode = GameErrorCodes.INVALID_PLAYER
+    else:
+        ecode = None
+    if ecode:
+        raise GameError(ecode, fn_code=GameFuncCodes.RTC_ENDPOINT)
+
+
+## TODO, error handling decorator
 @srv.on(RtcConst.EVENTS.INIT.value, namespace=RtcConst.NAMESPACE)
 async def init_communication(sid, data: Dict):
+    _logger.debug("init-rtc, raw-req-data: %s", data)
+    error = None
     try:
-        RtcInitMsgData(**data)
-        await srv.enter_room(sid, room=data["gameID"], namespace=RtcConst.NAMESPACE)
+        validated = RtcInitMsgData(**data)
+        await check_room_exist(_repo, data=validated)
+        await srv.enter_room(sid, room=validated.gameID, namespace=RtcConst.NAMESPACE)
         data["succeed"] = True
         await srv.emit(
             RtcConst.EVENTS.INIT.value,
             data,
             namespace=RtcConst.NAMESPACE,
-            room=data["gameID"],
+            room=validated.gameID,
         )
     except ValidationError as e:
         _logger.error("%s", e)
         error = e.errors(include_url=False, include_input=False)
         error["succeed"] = False
+    except GameError as e:
+        _logger.error("req-room-ID:%s , error:%s", validated.gameID, e)
+        error = {
+            "succeed": False,
+            "code": e.error_code.value[0],
+            "func": e.func_code.value,
+        }
+    if error:
         await srv.emit(
             RtcConst.EVENTS.INIT.value,
             namespace=RtcConst.NAMESPACE,
@@ -79,20 +95,30 @@ async def init_communication(sid, data: Dict):
 
 @srv.on(RtcConst.EVENTS.DEINIT.value, namespace=RtcConst.NAMESPACE)
 async def deinit_communication(sid, data: Dict):
+    error = None
     try:
-        RtcInitMsgData(**data)
-        await srv.leave_room(sid, room=data["gameID"], namespace=RtcConst.NAMESPACE)
+        validated = RtcInitMsgData(**data)
+        await check_room_exist(_repo, data=validated)
+        await srv.leave_room(sid, room=validated.gameID, namespace=RtcConst.NAMESPACE)
         data["succeed"] = True
         await srv.emit(
             RtcConst.EVENTS.DEINIT.value,
             data,
             namespace=RtcConst.NAMESPACE,
-            room=data["gameID"],
+            room=validated.gameID,
         )
     except ValidationError as e:
         _logger.error("%s", e)
         error = e.errors(include_url=False, include_input=False)
-        data["succeed"] = False
+        error["succeed"] = False
+    except GameError as e:
+        _logger.error("req-room-ID:%s , error:%s", validated.gameID, e)
+        error = {
+            "succeed": False,
+            "code": e.error_code.value[0],
+            "func": e.func_code.value,
+        }
+    if error:
         await srv.emit(
             RtcConst.EVENTS.DEINIT.value,
             namespace=RtcConst.NAMESPACE,
@@ -101,6 +127,15 @@ async def deinit_communication(sid, data: Dict):
         )
 
 
+@srv.on(RtcConst.EVENTS.NEW_ROOM.value, namespace=RtcConst.NAMESPACE)
+async def _new_game_room(sid, data: Dict):
+    try:  # TODO, ensure this event is sent by authorized http server
+        validated = RtcRoomMsgData(**data)
+        await _repo.save(validated)
+    except ValidationError as e:
+        _logger.error("%s", e)
+
+
 def gen_srv_task(host: str):
     cfg = Config()
     cfg.bind = [host]

backend/app/constant.py

@@ -9,6 +9,7 @@ class GameRtcEvent(Enum):
     INIT = "init"
     DEINIT = "deinit"
     CHAT = "chat"
+    NEW_ROOM = "new_room"
 
 
 class RealTimeCommConst:

backend/app/domain/game.py

@@ -26,14 +26,16 @@ class GameFuncCodes(Enum):
     SWITCH_CHARACTER = 1003
     CLEAR_CHARACTER_SELECTION = 1004
     UPDATE_DIFFICULTY = 1005
-    USE_CASE_EXECUTE = 1099  ## TODO, rename this
+    ## TODO, rename the following members
+    USE_CASE_EXECUTE = 1099
+    RTC_ENDPOINT = 1098  ## for real-time communication like socket.io server endpoint
 
 
 class GameError(Exception):
     def __init__(
-        self, e_code: GameFuncCodes, fn_code: GameFuncCodes, msg: Optional[str] = None
+        self, e_code: GameErrorCodes, fn_code: GameFuncCodes, msg: Optional[str] = None
     ):
-        self.error_code: GameFuncCodes = e_code
+        self.error_code: GameErrorCodes = e_code
         self.func_code: GameFuncCodes = fn_code
         self.message = msg
 

backend/app/dto.py

@@ -1,6 +1,6 @@
 from enum import Enum
 from typing import List
-from pydantic import BaseModel, RootModel
+from pydantic import BaseModel, RootModel, ConfigDict
 
 # data transfer objects (DTO) in the application
 # TODO, determine module path
@@ -63,3 +63,24 @@ class UpdateDifficultyDto(BaseModel):
 
 class UpdateCommonRespDto(BaseModel):
     message: str
+
+
+class RtcRoomMsgData(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    gameID: str
+    players: List[str]
+
+
+class ChatMsgData(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    msg: str
+    nickname: str  # TODO, replace with PlayerDto
+    gameID: str
+    client: str  ## client session ID
+
+
+class RtcInitMsgData(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    player: PlayerDto
+    gameID: str
+    client: str  # client session ID

backend/app/usecase/config_game.py

@@ -12,8 +12,9 @@
 
 
 class AbstractUseCase:
-    def __init__(self, repository, settings: Dict = None):
+    def __init__(self, repository, evt_emitter=None, settings: Dict = None):
         self.repository = repository
+        self.evt_emitter = evt_emitter
         self.settings = settings
 
 
@@ -29,6 +30,8 @@ async def execute(
         game.add_players(data)
         self.rand_select_investigator(game)
         await self.repository.save(game)
+        if self.evt_emitter:
+            await self.evt_emitter.create_game(game)
         return presenter(self.settings, game.id)
 
     def rand_select_investigator(self, game: Game):