[RTC-365] Server notifications

.gitignore

@@ -158,3 +158,6 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+# Temporary files
+/tmp

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "protos"]
+	path = protos
+	url = https://github.com/jellyfish-dev/protos.git

README.md

@@ -2,7 +2,7 @@
 
 [![CircleCI](https://dl.circleci.com/status-badge/img/gh/jellyfish-dev/python-server-sdk/tree/main.svg?style=svg)](https://dl.circleci.com/status-badge/redirect/gh/jellyfish-dev/python-server-sdk/tree/main)
 
-Python server SDK for the [Jellyfish](https://github.com/jellyfish-dev/jellyfish) media server.
+Python server SDK for the [Jellyfish Media Server](https://github.com/jellyfish-dev/jellyfish).
 
 Read the docs [here](https://jellyfish-dev.github.io/python-server-sdk/jellyfish.html)
 
@@ -15,15 +15,22 @@ pip install git+https://github.com/jellyfish-dev/python-server-sdk
 
 ## Usage
 
-First create a `RoomApi` instance, providing the jellyfish server address and api token
+The SDK exports two main classes for interacting with Jellyfish server:
+`RoomApi` and `Notifier`.
+
+`RoomApi` wraps http REST api calls, while `Notifier` is responsible for receiving real-time updates from the server.
+
+#### RoomApi
+
+Create a `RoomApi` instance, providing the jellyfish server address and api token
 
 ```python
 from jellyfish import RoomApi
 
-room_api = RoomApi(server_address='http://localhost:5002', server_api_token='development')
+room_api = RoomApi(server_address='localhost:5002', server_api_token='development')
 ```
 
-You can use it to interact with Jellyfish managing rooms, peers and components
+You can use it to interact with Jellyfish, manage rooms, peers and components
 
 ```python
 # Create a room
@@ -43,6 +50,60 @@ component_hls = room_api.add_component(room.id, options=ComponentOptionsHLS())
 # Component(actual_instance=ComponentHLS(id='c0dfab50-cafd-438d-985e-7b8f97ae55e3', metadata=ComponentMetadataHLS(low_latency=False, playable=False), type='hls'))
 ```
 
+#### Notifier
+
+Create `Notifier` instance
+```python
+from jellyfish import Notifier
+
+notifier = Notifier(server_address='localhost:5002', server_api_token='development')
+```
+
+Then define handlers for incoming messages
+```python
+@notifier.on_server_notification
+def handle_notification(server_notification):
+    print(f'Received a notification: {notification}')
+
+@notifier.on_metrics
+def handle_metrics(metrics_report):
+    print(f'Received WebRTC metrics: {metrics_report.metrics}')
+```
+
+After that you can start the notifier
+```python
+async def test_notifier():
+    notifier_task = asyncio.create_task(notifier.connect())
+
+    # Wait for notifier to be ready to receive messages
+    await notifier.wait_ready()
+
+    # Create a room to trigger a server notification
+    room_api = RoomApi()
+    room_api.create_room()
+
+    await notifier_task
+
+asyncio.run(test_notifier())
+
+# Received a notification: ServerMessageRoomCreated(room_id='69a3fd1a-6a4d-47bc-ae54-0c72b0d05e29')
+# Received WebRTC metrics: ServerMessageMetricsReport(metrics='{}')
+```
+
+## Testing
+
+You can test the SDK against a local instance of Jellyfish by running
+```console
+pytest
+```
+
+Make sure to use the default configuration for Jellyfish
+
+Alternatively you can test using Docker
+```console
+docker-compose -f docker-compose-test.yaml run test
+```
+
 ## Copyright and License
 
 Copyright 2023, [Software Mansion](https://swmansion.com/?utm_source=git&utm_medium=readme&utm_campaign=jellyfish)

compile_proto.sh

@@ -0,0 +1,20 @@
+#!/bin/bash
+
+# Terminate on errors
+set -e
+
+
+printf "Synchronising submodules... "
+git submodule sync --recursive >> /dev/null
+git submodule update --recursive --remote --init >> /dev/null
+printf "DONE\n\n"
+
+server_file="./protos/jellyfish/server_notifications.proto"
+printf "Compiling: file $server_file"
+protoc -I . --python_betterproto_out=./jellyfish/events/_protos $server_file
+printf "\tDONE\n"
+
+peer_file="./protos/jellyfish/peer_notifications.proto"
+printf "Compiling: file $peer_file"
+protoc -I . --python_betterproto_out=./tests/support/protos $peer_file
+printf "\tDONE\n"

dev-requirements.txt

@@ -1,9 +1,12 @@
 aenum==3.1.15
+betterproto==1.2.5
 pydantic==1.10.12
 pytest==7.1.3
+pytest-asyncio==0.21.1
 python_dateutil==2.8.2
 setuptools==67.6.1
 typing_extensions==4.7.1
 urllib3 >= 1.25.3, < 2
 pylint==2.17.5
 pdoc==14.1.0
+websockets==11.0.3

docker-compose-test.yaml

@@ -13,15 +13,14 @@ services:
       timeout: 2s
       start_period: 30s
     environment:
-      VIRTUAL_HOST: "jellyfish"
-      USE_INTEGRATED_TURN: "true"
-      INTEGRATED_TURN_IP: "${INTEGRATED_TURN_IP:-127.0.0.1}"
-      INTEGRATED_TURN_LISTEN_IP: "0.0.0.0"
-      INTEGRATED_TURN_PORT_RANGE: "50000-50050"
-      INTEGRATED_TCP_TURN_PORT: "49999"
-      SERVER_API_TOKEN: "development"
-      PORT: 5002
-      SECRET_KEY_BASE: "super-secret-key"
+      JF_HOST: "jellyfish"
+      JF_INTEGRATED_TURN_IP: "${INTEGRATED_TURN_IP:-127.0.0.1}"
+      JF_INTEGRATED_TURN_LISTEN_IP: "0.0.0.0"
+      JF_INTEGRATED_TURN_PORT_RANGE: "50000-50050"
+      JF_INTEGRATED_TCP_TURN_PORT: "49999"
+      JF_SERVER_API_TOKEN: "development"
+      JF_PORT: 5002
+      JF_SECRET_KEY_BASE: "super-secret-key"
     ports:
       - "5002:5002"
       - "49999:49999"
@@ -31,9 +30,9 @@ services:
 
   test:
     image: python:3.8-alpine3.18
-    command: sh -c "cd app/ && pip install -e . && pytest"
+    command: sh -c "cd app/ && pip install -r dev-requirements.txt && pytest -s"
     environment:
-      - DOCKER_TEST=TRUE
+      DOCKER_TEST: "TRUE"
     volumes:
       - .:/app
     networks:

jellyfish/__init__.py

@@ -6,17 +6,27 @@
 
 from pydantic.error_wrappers import ValidationError
 
+# API
+from jellyfish._room_api import RoomApi
+from jellyfish._notifier import Notifier
+
+# Models
 from jellyfish._openapi_client import (
     Room, RoomConfig, Peer, Component, ComponentHLS, ComponentRTSP, ComponentOptions,
     ComponentOptionsRTSP, ComponentOptionsHLS, PeerOptionsWebRTC)
 
+# Server Messages
+from jellyfish import events
+
+# Exceptions
 from jellyfish._openapi_client.exceptions import (
     UnauthorizedException, NotFoundException, BadRequestException)
 
-from jellyfish._room_api import RoomApi
 
-__all__ = ['RoomApi', 'Room', 'Peer', 'Component', 'ComponentHLS', 'ComponentRTSP',
-           'ComponentOptionsHLS', 'RoomConfig', 'ComponentOptions', 'ComponentOptionsRTSP',
-           'PeerOptionsWebRTC', 'UnauthorizedException', 'NotFoundException', 'BadRequestException']
+__all__ = [
+    'RoomApi', 'Notifier', 'Room', 'Peer', 'Component', 'ComponentHLS', 'ComponentRTSP',
+    'ComponentOptionsHLS', 'RoomConfig', 'ComponentOptions', 'ComponentOptionsRTSP',
+    'PeerOptionsWebRTC', 'events', 'UnauthorizedException', 'NotFoundException',
+    'BadRequestException']
 
 __docformat__ = "restructuredtext"

jellyfish/_notifier.py

@@ -0,0 +1,133 @@
+'''
+Notifier listening to WebSocket events
+'''
+
+import asyncio
+from typing import Callable, Any
+
+import betterproto
+
+from websockets import client
+from websockets.exceptions import ConnectionClosed
+
+from jellyfish.events import (
+    ServerMessage, ServerMessageAuthRequest, ServerMessageAuthenticated,
+    ServerMessageSubscribeRequest, ServerMessageEventType, ServerMessageSubscribeResponse,
+    ServerMessageMetricsReport)
+
+
+class Notifier:
+    '''
+    Allows for receiving WebSocket messages from Jellyfish.
+    '''
+
+    def __init__(self,
+                 server_address: str = 'localhost:5002', server_api_token: str = 'development'):
+        self._server_address = server_address
+        self._server_api_token = server_api_token
+        self._websocket = None
+        self._ready = False
+
+        self._ready_event: asyncio.Event = None
+
+        self._notification_handler: Callable = None
+        self._metrics_handler: Callable = None
+
+    def on_server_notification(self, handler: Callable[[Any], None]):
+        '''
+        Decorator used for defining handler for ServerNotifications
+        i.e. all messages other than `ServerMessageMetricsReport`.
+        '''
+        self._notification_handler = handler
+        return handler
+
+    def on_metrics(self, handler: Callable[[ServerMessageMetricsReport], None]):
+        '''
+        Decorator used for defining handler for `ServerMessageMetricsReport`.
+        '''
+        self._metrics_handler = handler
+        return handler
+
+    async def connect(self):
+        '''
+        A coroutine which connects Notifier to Jellyfish and listens for all incoming
+        messages from the Jellyfish.
+
+        It runs until the connection isn't closed.
+
+        The incoming messages are handled by the functions defined using the
+        `on_server_notification` and `on_metrics` decorators.
+
+        The handlers have to be defined before calling `connect`,
+        otherwise the messages won't be received.
+        '''
+        address = f'ws://{self._server_address}/socket/server/websocket'
+        async with client.connect(address) as websocket:
+            try:
+                self._websocket = websocket
+                await self._authenticate()
+
+                if self._notification_handler:
+                    await self._subscribe_event(
+                        event=ServerMessageEventType.EVENT_TYPE_SERVER_NOTIFICATION)
+
+                if self._metrics_handler:
+                    await self._subscribe_event(event=ServerMessageEventType.EVENT_TYPE_METRICS)
+
+                self._ready = True
+                if self._ready_event:
+                    self._ready_event.set()
+
+                await self._receive_loop()
+            finally:
+                self._websocket = None
+
+    async def wait_ready(self) -> True:
+        '''
+        Waits until the notifier is connected and authenticated to Jellyfish.
+
+        If already connected, returns `True` immediately.
+        '''
+        if self._ready:
+            return True
+
+        if self._ready_event is None:
+            self._ready_event = asyncio.Event()
+
+        await self._ready_event.wait()
+
+    async def _authenticate(self):
+        msg = ServerMessage(auth_request=ServerMessageAuthRequest(token=self._server_api_token))
+        await self._websocket.send(bytes(msg))
+
+        try:
+            message = await self._websocket.recv()
+        except ConnectionClosed as exception:
+            if 'invalid token' in str(exception):
+                raise RuntimeError('Invalid server_api_token') from exception
+            raise
+
+        message = ServerMessage().parse(message)
+
+        _type, message = betterproto.which_one_of(message, 'content')
+        assert isinstance(message, ServerMessageAuthenticated)
+
+    async def _receive_loop(self):
+        while True:
+            message = await self._websocket.recv()
+            message = ServerMessage().parse(message)
+            _which, message = betterproto.which_one_of(message, 'content')
+
+            if isinstance(message, ServerMessageMetricsReport):
+                self._metrics_handler(message)
+            else:
+                self._notification_handler(message)
+
+    async def _subscribe_event(self, event: ServerMessageEventType):
+        request = ServerMessage(subscribe_request=ServerMessageSubscribeRequest(event))
+
+        await self._websocket.send(bytes(request))
+        message = await self._websocket.recv()
+        message = ServerMessage().parse(message)
+        _which, message = betterproto.which_one_of(message, "content")
+        assert isinstance(message, ServerMessageSubscribeResponse)

jellyfish/_room_api.py

@@ -1,6 +1,6 @@
-"""
-    RoomApi used to manage rooms
-"""
+'''
+RoomApi used to manage rooms
+'''
 
 from typing import Union, Literal
 
@@ -20,7 +20,7 @@ def __init__(self,
         Create RoomApi instance, providing the jellyfish address and api token.
         '''
         self._configuration = jellyfish_api.Configuration(
-            host=server_address,
+            host=f'http://{server_address}',
             access_token=server_api_token
         )
 

jellyfish/events/__init__.py

@@ -0,0 +1,21 @@
+'''
+Server events being sent Jellyfish
+'''
+
+# Private messages
+from jellyfish.events._protos.jellyfish import (
+    ServerMessage, ServerMessageAuthenticated, ServerMessageAuthRequest, ServerMessageEventType,
+    ServerMessageSubscribeResponse, ServerMessageSubscribeRequest)
+
+# Exported messages
+from jellyfish.events._protos.jellyfish import (
+    ServerMessageComponentCrashed, ServerMessageHlsPlayable,
+    ServerMessageMetricsReport, ServerMessagePeerCrashed, ServerMessagePeerConnected,
+    ServerMessagePeerDisconnected, ServerMessageRoomCrashed, ServerMessageRoomDeleted,
+    ServerMessageRoomCreated)
+
+__all__ = [
+    'ServerMessageComponentCrashed', 'ServerMessageHlsPlayable',
+    'ServerMessageMetricsReport', 'ServerMessagePeerCrashed', 'ServerMessagePeerConnected',
+    'ServerMessagePeerDisconnected', 'ServerMessageRoomCrashed', 'ServerMessageRoomDeleted',
+    'ServerMessageRoomCreated']