Display player count (online and in-game)

battleship/client/__init__.py

@@ -3,13 +3,13 @@
     ClientError,
     ConnectionImpossible,
     RequestFailed,
-    SessionSubscription,
     Unauthorized,
 )
 from battleship.client.credentials import (
     CredentialsProvider,
     filesystem_credentials_provider,
 )
+from battleship.client.subscriptions import PlayerSubscription, SessionSubscription
 
 __all__ = [
     "Client",
@@ -20,4 +20,5 @@
     "Unauthorized",
     "ConnectionImpossible",
     "ClientError",
+    "PlayerSubscription",
 ]

battleship/client/client.py

@@ -1,7 +1,7 @@
 import asyncio
 import json as json_
 from asyncio import Task
-from typing import Any, AsyncIterator, Callable, Collection, Coroutine, Optional, cast
+from typing import Any, AsyncIterator, Callable, Collection, Optional
 from urllib.parse import urlparse
 
 import httpx
@@ -15,6 +15,7 @@
 
 from battleship.client.auth import IDTokenAuth
 from battleship.client.credentials import Credentials, CredentialsProvider
+from battleship.client.subscriptions import PlayerSubscription, SessionSubscription
 from battleship.shared.compat import async_timeout as timeout
 from battleship.shared.events import (
     ClientEvent,
@@ -26,6 +27,7 @@
     Action,
     IDToken,
     LoginData,
+    PlayerCount,
     PlayerStatistics,
     Session,
     SessionID,
@@ -70,23 +72,6 @@ def done(self) -> None:
         self._event.set()
 
 
-class SessionSubscription:
-    def __init__(self) -> None:
-        self._ee = AsyncIOEventEmitter()
-
-    def on_add(self, callback: Callable[[Session], Coroutine[Any, Any, Any]]) -> None:
-        self._ee.add_listener(Action.ADD, callback)
-
-    def on_remove(self, callback: Callable[[SessionID], Coroutine[Any, Any, Any]]) -> None:
-        self._ee.add_listener(Action.REMOVE, callback)
-
-    def on_start(self, callback: Callable[[SessionID], Coroutine[Any, Any, Any]]) -> None:
-        self._ee.add_listener(Action.START, callback)
-
-    def emit(self, event: str, *args: Any, **kwargs: Any) -> None:
-        self._ee.emit(event, *args, **kwargs)
-
-
 class Client:
     """
     Provides a convenient interface to the server API and realtime events.
@@ -280,9 +265,9 @@ async def fetch_statistics(self) -> PlayerStatistics:
         response = await self._request("GET", f"/statistics/{self.nickname}")
         return PlayerStatistics(**response.json())
 
-    async def fetch_clients_online(self) -> int:
-        response = await self._request("GET", "/clients/online")
-        return cast(int, response.json())
+    async def fetch_players_online(self) -> PlayerCount:
+        response = await self._request("GET", "/players/online")
+        return PlayerCount(**response.json())
 
     async def sessions_subscribe(self) -> SessionSubscription:
         subscription = SessionSubscription()
@@ -306,6 +291,22 @@ async def publish_update(payload: dict) -> None:  # type: ignore[type-arg]
     async def sessions_unsubscribe(self) -> None:
         await self._request("POST", url="/sessions/unsubscribe")
 
+    async def players_subscribe(self) -> PlayerSubscription:
+        subscription = PlayerSubscription()
+
+        def publish_update(payload: dict[str, Any]) -> None:
+            count = payload["count"]
+            event = payload["event"]
+
+            subscription.emit(event, count=count)
+
+        self._emitter.add_listener(ServerEvent.PLAYERS_UPDATE, publish_update)
+        await self._request("POST", url="/players/subscribe")
+        return subscription
+
+    async def players_unsubscribe(self) -> None:
+        await self._request("POST", url="/players/unsubscribe")
+
     def add_listener(self, event: str, handler: Callable[..., Any]) -> None:
         self._emitter.add_listener(event, handler)
 

battleship/client/subscriptions.py

@@ -0,0 +1,41 @@
+from typing import Any, Callable, Coroutine, TypeAlias
+
+from pyee.asyncio import AsyncIOEventEmitter
+
+from battleship.shared.models import Action, Session, SessionID
+
+ClientCount: TypeAlias = int
+SessionCallback = Callable[[Session], Coroutine[Any, Any, Any]]
+SessionIDCallback = Callable[[SessionID], Coroutine[Any, Any, Any]]
+ClientCallback = Callable[[ClientCount], Coroutine[Any, Any, Any]]
+
+
+class SessionSubscription:
+    def __init__(self) -> None:
+        self._ee = AsyncIOEventEmitter()
+
+    def on_add(self, callback: SessionCallback) -> None:
+        self._ee.add_listener(Action.ADD, callback)
+
+    def on_remove(self, callback: SessionIDCallback) -> None:
+        self._ee.add_listener(Action.REMOVE, callback)
+
+    def on_start(self, callback: SessionIDCallback) -> None:
+        self._ee.add_listener(Action.START, callback)
+
+    def emit(self, event: str, *args: Any, **kwargs: Any) -> None:
+        self._ee.emit(event, *args, **kwargs)
+
+
+class PlayerSubscription:
+    def __init__(self) -> None:
+        self._ee = AsyncIOEventEmitter()
+
+    def on_online_changed(self, callback: ClientCallback) -> None:
+        self._ee.add_listener("online_changed", callback)
+
+    def on_ingame_changed(self, callback: ClientCallback) -> None:
+        self._ee.add_listener("ingame_changed", callback)
+
+    def emit(self, event: str, *args: Any, **kwargs: Any) -> None:
+        self._ee.emit(event, *args, **kwargs)

battleship/server/app.py

@@ -10,8 +10,8 @@
 from battleship.server.auth import Auth0AuthManager, AuthManager
 from battleship.server.config import Config, get_config
 from battleship.server.handlers import (
-    ClientSubscriptionHandler,
     GameHandler,
+    PlayerSubscriptionHandler,
     SessionSubscriptionHandler,
 )
 from battleship.server.pubsub import (
@@ -69,7 +69,7 @@ def create_app() -> Application:
     app.services.add_singleton(OutgoingChannel, OutgoingRedisChannel)
     app.services.add_singleton(SessionSubscriptionHandler)
     app.services.add_singleton(GameHandler)
-    app.services.add_singleton(ClientSubscriptionHandler)
+    app.services.add_singleton(PlayerSubscriptionHandler)
 
     app.use_authentication().add(
         JWTBearerAuthentication(

battleship/server/handlers.py

@@ -68,6 +68,10 @@ def __init__(self, out_channel: OutgoingChannel, session_repository: SessionRepo
         self._out = out_channel
         self._sessions = session_repository
 
+    @property
+    def cb_namespace(self) -> str:
+        return self.__class__.__name__.lower()
+
     def make_session_observer(self, client_id: str) -> Listener:
         async def session_observer(session_id: str, action: Action) -> None:
             payload: dict[str, Any] = dict(action=action)
@@ -89,38 +93,72 @@ async def session_observer(session_id: str, action: Action) -> None:
         return session_observer
 
     def subscribe(self, client_id: str) -> None:
-        self._sessions.subscribe(client_id, self.make_session_observer(client_id))
+        callback_id = f"{self.cb_namespace}:{client_id}"
+        self._sessions.subscribe(callback_id, self.make_session_observer(client_id))
 
     def unsubscribe(self, client_id: str) -> None:
-        self._sessions.unsubscribe(client_id)
+        callback_id = f"{self.cb_namespace}:{client_id}"
+        self._sessions.unsubscribe(callback_id)
 
 
-class ClientSubscriptionHandler:
-    def __init__(self, out_channel: OutgoingChannel, client_repository: ClientRepository):
+class PlayerSubscriptionHandler:
+    def __init__(
+        self,
+        out_channel: OutgoingChannel,
+        client_repository: ClientRepository,
+        session_repository: SessionRepository,
+    ):
         self._out = out_channel
         self._clients = client_repository
+        self._sessions = session_repository
+
+    @property
+    def cb_namespace(self) -> str:
+        return self.__class__.__name__.lower()
 
     def make_client_observer(self, client_id: str) -> Listener:
         async def client_observer(_: str, action: Action) -> None:
-            payload: dict[str, Any] = dict(action=action)
-
             if action not in (Action.ADD, Action.REMOVE):
                 return
 
-            payload.update(count=await self._clients.count())
+            payload = dict(event="online_changed", count=await self._clients.count())
 
             await self._out.publish(
                 client_id,
                 EventMessage(
-                    kind=ServerEvent.CLIENTS_UPDATE,
+                    kind=ServerEvent.PLAYERS_UPDATE,
                     payload=payload,
                 ),
             )
 
         return client_observer
 
+    def make_session_observer(self, client_id: str) -> Listener:
+        async def session_observer(_: str, action: Action) -> None:
+            if action not in (Action.START, Action.REMOVE):
+                return
+
+            sessions = await self._sessions.list()
+            started_sessions = [s for s in sessions if s.started]
+            players_ingame = len(started_sessions) * 2
+            payload = dict(event="ingame_changed", count=players_ingame)
+
+            await self._out.publish(
+                client_id,
+                EventMessage(
+                    kind=ServerEvent.PLAYERS_UPDATE,
+                    payload=payload,
+                ),
+            )
+
+        return session_observer
+
     def subscribe(self, client_id: str) -> None:
-        self._clients.subscribe(client_id, self.make_client_observer(client_id))
+        callback_id = f"{self.cb_namespace}:{client_id}"
+        self._clients.subscribe(callback_id, self.make_client_observer(client_id))
+        self._sessions.subscribe(callback_id, self.make_session_observer(client_id))
 
     def unsubscribe(self, client_id: str) -> None:
-        self._clients.unsubscribe(client_id)
+        callback_id = f"{self.cb_namespace}:{client_id}"
+        self._clients.unsubscribe(callback_id)
+        self._sessions.unsubscribe(callback_id)

battleship/server/routes.py

@@ -17,8 +17,8 @@
 
 from battleship.server.auth import AuthManager, InvalidSignup, WrongCredentials
 from battleship.server.handlers import (
-    ClientSubscriptionHandler,
     GameHandler,
+    PlayerSubscriptionHandler,
     SessionSubscriptionHandler,
 )
 from battleship.server.pubsub import IncomingChannel, OutgoingChannel
@@ -33,6 +33,7 @@
     IDToken,
     LoginCredentials,
     LoginData,
+    PlayerCount,
     RefreshToken,
     Session,
     SessionCreate,
@@ -50,7 +51,7 @@ async def ws(
     in_channel: IncomingChannel,
     out_channel: OutgoingChannel,
     session_subscription_handler: SessionSubscriptionHandler,
-    client_subscription_handler: ClientSubscriptionHandler,
+    player_subscription_handler: PlayerSubscriptionHandler,
     session_repository: SessionRepository,
     game_handler: GameHandler,
 ) -> None:
@@ -65,7 +66,7 @@ async def ws(
     await connection.listen()
     logger.debug(f"{connection} disconnected.")
     session_subscription_handler.unsubscribe(client.id)
-    client_subscription_handler.unsubscribe(client.id)
+    player_subscription_handler.unsubscribe(client.id)
 
     current_session = await session_repository.get_for_client(client.id)
 
@@ -140,29 +141,35 @@ async def join_session(
     game_handler.start_new_game(host, guest, session)
 
 
-@router.get("/clients/online")
-async def get_players_online(client_repository: ClientRepository) -> int:
-    return await client_repository.count()
+@router.get("/players/online")
+async def get_players_online(
+    client_repository: ClientRepository, session_repository: SessionRepository
+) -> PlayerCount:
+    players = await client_repository.count()
+    sessions = await session_repository.list()
+    started_sessions = [s for s in sessions if s.started]
+    players_ingame = len(started_sessions) * 2
+    return PlayerCount(total=players, ingame=players_ingame)
 
 
-@router.post("/clients/subscribe")
-async def subscribe_to_client_count_updates(
+@router.post("/players/subscribe")
+async def subscribe_to_player_count_updates(
     identity: Identity,
     client_repository: ClientRepository,
-    client_subscription_handler: ClientSubscriptionHandler,
+    player_subscription_handler: PlayerSubscriptionHandler,
 ) -> None:
     client = await client_repository.get(identity.claims["sub"])
-    client_subscription_handler.subscribe(client.id)
+    player_subscription_handler.subscribe(client.id)
 
 
-@router.post("/clients/unsubscribe")
-async def unsubscribe_from_client_count_updates(
+@router.post("/players/unsubscribe")
+async def unsubscribe_from_player_count_updates(
     identity: Identity,
     client_repository: ClientRepository,
-    client_subscription_handler: ClientSubscriptionHandler,
+    player_subscription_handler: PlayerSubscriptionHandler,
 ) -> None:
     client = await client_repository.get(identity.claims["sub"])
-    client_subscription_handler.unsubscribe(client.id)
+    player_subscription_handler.unsubscribe(client.id)
 
 
 @allow_anonymous()

battleship/shared/events.py

@@ -25,7 +25,7 @@ class ServerEvent(StrEnum):
     SALVO = auto()
     GAME_ENDED = auto()
     GAME_CANCELLED = auto()
-    CLIENTS_UPDATE = auto()
+    PLAYERS_UPDATE = auto()
 
 
 Event: TypeAlias = ServerEvent | ClientEvent

battleship/shared/models.py

@@ -230,5 +230,10 @@ class Client(BaseModel):
     guest: bool
 
 
+class PlayerCount(BaseModel):
+    total: int
+    ingame: int
+
+
 def make_session_id() -> SessionID:
     return f"session_{secrets.token_urlsafe(8)}"