apologiesserver.manager

State manager.

Python’s asyncio is primarily meant for use in single-threaded code, but there is still concurrent execution happening any time we hit a yield from or await.

We want to minimize the risk of unexpected behavior when there are conflicting requests. For instance, if we simultaneously get a request to start a game and to quit a game, we want to make sure that one operation completes entirely before the next one starts. This means that we need thread synchronization whenever state is updated.

I’ve chosen to synchronize all state upate operations behind a single transaction boundary (a single lock). This is easier to follow and easier to write (correctly) than tracking individual locks at a more granular level, like at the player or the game level. The state will never be locked for all that long, because state update operations are all done in-memory and are quite fast. The slow stuff like network requests all happen outside the lock, whether we’re processing a request or executing a scheduled task.

The design would be different if we were using a database to save state, but this seems like the best compromise for the simple in-memory design that we’re using now.

The transaction boundary is handled by a single lock on the StateManager object. Callers must ensure that they get that lock before reading or modifying state in any way. Other than that, none of the objects defined in this module are thread-safe, or even thread-aware. There are no asynchronous methods or await calls. This simplifies the implementation and avoids confusion.

Module Contents

apologiesserver.manager.log
class apologiesserver.manager.TrackedWebsocket

The state that is tracked for a websocket within the state manager.

websocket :websockets.legacy.server.WebSocketServerProtocol
registration_date :apologiesserver.interface.DateTime
last_active_date :apologiesserver.interface.DateTime
activity_state :apologiesserver.interface.ActivityState
player_ids :ordered_set.OrderedSet[str]
mark_active() → None

Mark the websocket as active.

mark_idle() → None

Mark the websocket as idle.

mark_inactive() → None

Mark the websocket as inactive.

class apologiesserver.manager.TrackedPlayer

The state that is tracked for a player within the state manager.

player_id :str
handle :str
websocket :apologiesserver.interface.Optional[websockets.legacy.server.WebSocketServerProtocol]
registration_date :apologiesserver.interface.DateTime
last_active_date :apologiesserver.interface.DateTime
activity_state :apologiesserver.interface.ActivityState
connection_state :apologiesserver.interface.ConnectionState
player_state :apologiesserver.interface.PlayerState
game_id :apologiesserver.interface.Optional[str]
static for_context(player_id: str, websocket: websockets.legacy.server.WebSocketServerProtocol, handle: str) → TrackedPlayer

Create a tracked player based on provided context.

to_registered_player() → apologiesserver.interface.RegisteredPlayer

Convert this TrackedPlayer to a RegisteredPlayer.

mark_active() → None

Mark the player as active.

mark_idle() → None

Mark the player as idle.

mark_inactive() → None

Mark the player as inactive.

mark_joined(game: TrackedGame) → None

Mark that the player has joined a game.

mark_playing() → None

Mark that the player is playing a game.

mark_quit() → None

Mark that the player has quit a game.

mark_disconnected() → None

Mark the player as disconnected.

class apologiesserver.manager.CurrentTurn
handle :str
color :apologiesserver.interface.PlayerColor
view :apologiesserver.interface.PlayerView
movelist :apologiesserver.interface.List[apologiesserver.interface.Move]
movedict :apologiesserver.interface.Dict[str, apologiesserver.interface.Move]
draw_again(engine: apologies.Engine) → CurrentTurn
static next_player(engine: apologies.Engine) → CurrentTurn
static for_handle(engine: apologies.Engine, handle: str, color: apologiesserver.interface.PlayerColor) → CurrentTurn
class apologiesserver.manager.TrackedEngine

Wrapper over an Apologies game engine, to manage game play state for TrackedGame.

start_game(mode: apologiesserver.interface.GameMode, handles: apologiesserver.interface.List[str]) → apologiesserver.interface.Dict[str, apologiesserver.interface.PlayerColor]

Start the game, returning a map from handle to assigned color.

stop_game() → None

Stop the game after the game is completed or has been cancelled.

get_next_turn() → str

Get the handle of the player that should play the next turn.

Get the legal moves for the player at this stage in the game.

get_player_view(handle: str) → apologiesserver.interface.PlayerView

Get the player’s view of the game state.

get_recent_history(max_entries: int) → apologiesserver.interface.List[apologiesserver.interface.History]

Return up to a certain number of game history entries.

is_move_pending(handle: str) → bool

Whether a move is pending for the player with the passed-in handle.

Whether the passed-in move id is a legal move for the player.

execute_move(handle: str, move_id: str) → Tuple[bool, apologiesserver.interface.Optional[str], apologiesserver.interface.Optional[str]]

Execute a player’s move, returning whether the game was completed (and the winner and a comment if so).

class apologiesserver.manager.TrackedGame

The state that is tracked for a game within the state manager.

game_id :str
advertiser_handle :str
name :str
mode :apologiesserver.interface.GameMode
players :int
visibility :apologiesserver.interface.Visibility
invited_handles :apologiesserver.interface.List[str]
advertised_date :apologiesserver.interface.DateTime
last_active_date :apologiesserver.interface.DateTime
started_date :apologiesserver.interface.Optional[apologiesserver.interface.DateTime]
completed_date :apologiesserver.interface.Optional[apologiesserver.interface.DateTime]
game_state :apologiesserver.interface.GameState
activity_state :apologiesserver.interface.ActivityState
cancelled_reason :apologiesserver.interface.Optional[apologiesserver.interface.CancelledReason]
completed_comment :apologiesserver.interface.Optional[str]
game_players :apologiesserver.interface.Dict[str, apologiesserver.interface.GamePlayer]
static for_context(advertiser_handle: str, game_id: str, context: apologiesserver.interface.AdvertiseGameContext) → TrackedGame

Create a tracked game based on provided context.

property completed → bool

Whether the game is completed.

to_advertised_game() → apologiesserver.interface.AdvertisedGame

Convert this tracked game to an AdvertisedGame.

get_game_players() → apologiesserver.interface.List[apologiesserver.interface.GamePlayer]

Get a list of game players.

get_available_players() → apologiesserver.interface.List[apologiesserver.interface.GamePlayer]

Get the players that are still available to play the game.

get_available_player_count() → int

Get the number of players that are still available to play the game.

get_next_turn() → Tuple[str, apologiesserver.interface.PlayerType]

Get the next turn to be played.

Get the legal moves for the player at this stage in the game.

get_player_view(handle: str) → apologiesserver.interface.PlayerView

Get the player’s view of the game state.

get_recent_history(max_entries: int) → apologiesserver.interface.List[apologiesserver.interface.History]

Return up to a certain number of game history entries.

is_available(handle: str) → bool

Whether the game is available to be joined by the passed-in player.

is_advertised() → bool

Whether a game is currently being advertised.

is_playing() → bool

Whether a game is being played.

is_in_progress() → bool

Whether a game is in-progress, meaning it is advertised or being played.

is_fully_joined() → bool

Whether the number of requested players have joined the game.

is_viable() → bool

Whether the game is viable.

is_move_pending(handle: str) → bool

Whether a move is pending for the player with the passed-in handle.

Whether the passed-in move id is a legal move for the player.

mark_active() → None

Mark the game as active.

mark_idle() → None

Mark the game as idle.

mark_inactive() → None

Mark the game as inactive.

mark_joined(handle: str) → None

Mark that a player has joined a game.

mark_started() → None

Mark the game as started.

mark_completed(comment: apologiesserver.interface.Optional[str]) → None

Mark the game as completed.

mark_cancelled(reason: apologiesserver.interface.CancelledReason, comment: apologiesserver.interface.Optional[str] = None) → None

Mark the game as cancelled.

mark_quit(handle: str) → None

Mark that the player has quit a game.

execute_move(handle: str, move_id: str) → Tuple[bool, apologiesserver.interface.Optional[str], apologiesserver.interface.Optional[str]]

Execute a player’s move, returning an indication of whether the game was completed (and a comment if so).

class apologiesserver.manager.StateManager

Manages system state.

lock :asyncio.Lock
mark_active(player: TrackedPlayer) → None

Mark a player and its associated websocket as active.

track_websocket(websocket: websockets.legacy.server.WebSocketServerProtocol) → None

Track a connected websocket.

delete_websocket(websocket: websockets.legacy.server.WebSocketServerProtocol) → None

Delete a websocket, so it is no longer tracked.

get_websocket_count() → int

Return the number of connected websockets in the system.

lookup_all_websockets() → apologiesserver.interface.List[websockets.legacy.server.WebSocketServerProtocol]

Return a list of websockets for all tracked players.

lookup_players_for_websocket(websocket: websockets.legacy.server.WebSocketServerProtocol) → apologiesserver.interface.List[TrackedPlayer]

Look up the players associated with a websocket, if any.

track_game(player: TrackedPlayer, advertised: apologiesserver.interface.AdvertiseGameContext) → TrackedGame

Track a newly-advertised game.

delete_game(game: TrackedGame) → None

Delete a tracked game, so it is no longer tracked.

get_total_game_count() → int

Return the total number of games in the system.

get_in_progress_game_count() → int

Return the number of in-progress games in the system.

lookup_game(game_id: apologiesserver.interface.Optional[str] = None, player: apologiesserver.interface.Optional[TrackedPlayer] = None) → apologiesserver.interface.Optional[TrackedGame]

Look up a game by id, returning None if the game is not found.

lookup_all_games() → apologiesserver.interface.List[TrackedGame]

Return a list of all tracked games.

lookup_in_progress_games() → apologiesserver.interface.List[TrackedGame]

Return a list of all in-progress games.

lookup_game_players(game: TrackedGame) → apologiesserver.interface.List[TrackedPlayer]

Lookup the players that are currently playing a game.

lookup_available_games(player: TrackedPlayer) → apologiesserver.interface.List[TrackedGame]

Return a list of games the passed-in player may join.

track_player(websocket: websockets.legacy.server.WebSocketServerProtocol, handle: str) → TrackedPlayer

Track a newly-registered player.

retrack_player(player: TrackedPlayer, websocket: websockets.legacy.server.WebSocketServerProtocol) → None

Re-track and existing player, associating it with a different websocket.

delete_player(player: TrackedPlayer) → None

Delete a tracked player, so it is no longer tracked.

get_registered_player_count() → int

Return the number of registered players in the system.

lookup_player(player_id: apologiesserver.interface.Optional[str] = None, handle: apologiesserver.interface.Optional[str] = None) → apologiesserver.interface.Optional[TrackedPlayer]

Look up a player by either player id or handle.

lookup_all_players() → apologiesserver.interface.List[TrackedPlayer]

Return a list of all tracked players.

lookup_websocket_activity() → apologiesserver.interface.List[Tuple[TrackedWebsocket, apologiesserver.interface.DateTime, int]]

Look up the last active date and number of registered players for all websockets.

lookup_player_activity() → apologiesserver.interface.List[Tuple[TrackedPlayer, apologiesserver.interface.DateTime, apologiesserver.interface.ConnectionState]]

Look up the last active date and connection state for all players.

lookup_game_activity() → apologiesserver.interface.List[Tuple[TrackedGame, apologiesserver.interface.DateTime]]

Look up the last active date for all games.

lookup_game_completion() → apologiesserver.interface.List[Tuple[TrackedGame, apologiesserver.interface.Optional[apologiesserver.interface.DateTime]]]

Look up the completed date for all completed games.

apologiesserver.manager.manager() → StateManager

Return the state manager.