daniel156161/snake-python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

remove stroing of the Game Board State into Redis or Memory

Browse Source
This commit is contained in:
Daniel Dolezal
2026-04-08 08:36:54 +02:00
parent f6e19e18e6
commit a62501cf22
11 changed files with 13 additions and 296 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/services/game_runtime.py
+6 -33
View File
@@ -1,4 +1,3 @@
from typing import Protocol, cast
import time
from server.metrics import MetricsCollector
@@ -6,18 +5,9 @@ from server.GameBoard import GameBoard
from snakes import SnakeBuilder
class GameStateStoreLike(Protocol):
async def save(self, game_id: str, game_board: GameBoard) -> None: ...
async def load(self, game_id: str) -> object | None: ...
async def delete(self, game_id: str) -> None: ...
class GameRuntimeService:
def __init__(self, game_state_store:GameStateStoreLike, snake_type:str, game_state_local_cache:bool, stale_game_timeout_sec:int):
self.game_state_store = game_state_store
def __init__(self, snake_type:str, stale_game_timeout_sec:int):
self.snake_type = snake_type
self.game_state_local_cache = game_state_local_cache
self.stale_game_timeout_sec = stale_game_timeout_sec
self.metrics_collector = None
@@ -41,43 +31,27 @@ class GameRuntimeService:
)
await new_game_board.start_game(game_state)
if self.game_state_local_cache:
self.running_games[game_id] = new_game_board
await self.game_state_store.save(game_id, new_game_board)
self.running_games[game_id] = new_game_board
self.game_move_counts[game_id] = 0
self.game_last_seen_unix[game_id] = int(time.time())
if self.metrics_collector is not None:
await self.metrics_collector.record_game_started(len(self.game_last_seen_unix))
return new_game_board
async def persist_game_board(self, game_id:str, game_board:GameBoard) -> None:
if self.game_state_local_cache:
self.running_games[game_id] = game_board
await self.game_state_store.save(game_id, game_board)
async def delete_game_board(self, game_state:dict) -> None:
game_id = game_state['game']['id']
self.running_games.pop(game_id, None)
self.game_move_counts.pop(game_id, None)
self.game_last_seen_unix.pop(game_id, None)
await self.game_state_store.delete(game_id)
async def get_game_board(self, game_state:dict, end:bool=False) -> GameBoard:
game_id = game_state['game']['id']
game_board:GameBoard
if self.game_state_local_cache and game_id in self.running_games:
if game_id in self.running_games:
game_board = self.running_games[game_id]
else:
persisted_board = await self.game_state_store.load(game_id)
if persisted_board is not None:
game_board = cast(GameBoard, persisted_board)
if self.game_state_local_cache:
self.running_games[game_id] = game_board
else:
game_board = await self.create_game_board(game_state)
if self.metrics_collector is not None:
await self.metrics_collector.record_game_autocreated()
game_board = await self.create_game_board(game_state)
if self.metrics_collector is not None:
await self.metrics_collector.record_game_autocreated()
if not end:
self.game_move_counts[game_id] = self.game_move_counts.get(game_id, 0) + 1
@@ -86,7 +60,6 @@ class GameRuntimeService:
game_board.read_game_data(game_state)
if end:
game_board.end_game(game_state)
await self.persist_game_board(game_id, game_board)
return game_board