Add viewer token (#27)

* Add generating viewer token to API

* Update openapi

* Enable broadcasting rooms in fishjam container

* Fix tests

* Rename room type broadcaster to livestream

Author: Karolk99

docker-compose-test.yaml

@@ -17,6 +17,10 @@ services:
       FJ_SECRET_KEY_BASE: "super-secret-key"
       FJ_SIP_IP: "127.0.0.1"
       FJ_COMPONENTS_USED: "rtsp file hls recording sip"
+      FJ_BROADCASTING_ENABLED: "true"
+      FJ_BROADCASTER_URL: "http://broadcaster:4000" 
+      FJ_BROADCASTER_TOKEN: "broadcaster_token" 
+      FJ_BROADCASTER_WHIP_TOKEN: "whip_token"
     ports:
       - "5002:5002"
       - "49999:49999"

examples/room_manager/room_service.py

@@ -81,7 +81,6 @@ def __find_or_create_room(
         options = RoomOptions(
             max_peers=self.config.max_peers,
             webhook_url=self.config.webhook_url,
-            peerless_purge_timeout=self.config.peerless_purge_timeout,
             room_type=room_type.value if room_type else "full_feature",
         )
 

fishjam/_openapi_client/api/viewer/generate_token.py

@@ -1,11 +1,12 @@
 from http import HTTPStatus
-from typing import Any, Dict, Optional, Union, cast
+from typing import Any, Dict, Optional, Union
 
 import httpx
 
 from ... import errors
 from ...client import AuthenticatedClient, Client
 from ...models.error import Error
+from ...models.viewer_token import ViewerToken
 from ...types import Response
 
 
@@ -22,9 +23,10 @@ def _get_kwargs(
 
 def _parse_response(
     *, client: Union[AuthenticatedClient, Client], response: httpx.Response
-) -> Optional[Union[Error, str]]:
+) -> Optional[Union[Error, ViewerToken]]:
     if response.status_code == HTTPStatus.CREATED:
-        response_201 = cast(str, response.json())
+        response_201 = ViewerToken.from_dict(response.json())
+
         return response_201
     if response.status_code == HTTPStatus.BAD_REQUEST:
         response_400 = Error.from_dict(response.json())
@@ -50,7 +52,7 @@ def _parse_response(
 
 def _build_response(
     *, client: Union[AuthenticatedClient, Client], response: httpx.Response
-) -> Response[Union[Error, str]]:
+) -> Response[Union[Error, ViewerToken]]:
     return Response(
         status_code=HTTPStatus(response.status_code),
         content=response.content,
@@ -63,8 +65,8 @@ def sync_detailed(
     room_id: str,
     *,
     client: Union[AuthenticatedClient, Client],
-) -> Response[Union[Error, str]]:
-    """Generate token for single viewer
+) -> Response[Union[Error, ViewerToken]]:
+    """Generate single broadcaster access token
 
     Args:
         room_id (str):
@@ -74,7 +76,7 @@ def sync_detailed(
         httpx.TimeoutException: If the request takes longer than Client.timeout.
 
     Returns:
-        Response[Union[Error, str]]
+        Response[Union[Error, ViewerToken]]
     """
 
     kwargs = _get_kwargs(
@@ -92,8 +94,8 @@ def sync(
     room_id: str,
     *,
     client: Union[AuthenticatedClient, Client],
-) -> Optional[Union[Error, str]]:
-    """Generate token for single viewer
+) -> Optional[Union[Error, ViewerToken]]:
+    """Generate single broadcaster access token
 
     Args:
         room_id (str):
@@ -103,7 +105,7 @@ def sync(
         httpx.TimeoutException: If the request takes longer than Client.timeout.
 
     Returns:
-        Union[Error, str]
+        Union[Error, ViewerToken]
     """
 
     return sync_detailed(
@@ -116,8 +118,8 @@ async def asyncio_detailed(
     room_id: str,
     *,
     client: Union[AuthenticatedClient, Client],
-) -> Response[Union[Error, str]]:
-    """Generate token for single viewer
+) -> Response[Union[Error, ViewerToken]]:
+    """Generate single broadcaster access token
 
     Args:
         room_id (str):
@@ -127,7 +129,7 @@ async def asyncio_detailed(
         httpx.TimeoutException: If the request takes longer than Client.timeout.
 
     Returns:
-        Response[Union[Error, str]]
+        Response[Union[Error, ViewerToken]]
     """
 
     kwargs = _get_kwargs(
@@ -143,8 +145,8 @@ async def asyncio(
     room_id: str,
     *,
     client: Union[AuthenticatedClient, Client],
-) -> Optional[Union[Error, str]]:
-    """Generate token for single viewer
+) -> Optional[Union[Error, ViewerToken]]:
+    """Generate single broadcaster access token
 
     Args:
         room_id (str):
@@ -154,7 +156,7 @@ async def asyncio(
         httpx.TimeoutException: If the request takes longer than Client.timeout.
 
     Returns:
-        Union[Error, str]
+        Union[Error, ViewerToken]
     """
 
     return (

fishjam/_openapi_client/models/__init__.py

@@ -68,6 +68,7 @@
 from .track_type import TrackType
 from .user import User
 from .user_listing_response import UserListingResponse
+from .viewer_token import ViewerToken
 
 __all__ = (
     "AddComponentJsonBody",
@@ -130,4 +131,5 @@
     "TrackType",
     "User",
     "UserListingResponse",
+    "ViewerToken",
 )

fishjam/_openapi_client/models/room_config.py

@@ -16,10 +16,6 @@ class RoomConfig:
 
     max_peers: Union[Unset, None, int] = UNSET
     """Maximum amount of peers allowed into the room"""
-    peer_disconnected_timeout: Union[Unset, None, int] = UNSET
-    """Duration (in seconds) after which the peer will be removed if it is disconnected. If not provided, this feature is disabled."""
-    peerless_purge_timeout: Union[Unset, None, int] = UNSET
-    """Duration (in seconds) after which the room will be removed if no peers are connected. If not provided, this feature is disabled."""
     room_type: Union[Unset, RoomConfigRoomType] = RoomConfigRoomType.FULL_FEATURE
     """The use-case of the room. If not provided, this defaults to full_feature."""
     video_codec: Union[Unset, None, RoomConfigVideoCodec] = UNSET
@@ -32,8 +28,6 @@ class RoomConfig:
     def to_dict(self) -> Dict[str, Any]:
         """@private"""
         max_peers = self.max_peers
-        peer_disconnected_timeout = self.peer_disconnected_timeout
-        peerless_purge_timeout = self.peerless_purge_timeout
         room_type: Union[Unset, str] = UNSET
         if not isinstance(self.room_type, Unset):
             room_type = self.room_type.value
@@ -49,10 +43,6 @@ def to_dict(self) -> Dict[str, Any]:
         field_dict.update({})
         if max_peers is not UNSET:
             field_dict["maxPeers"] = max_peers
-        if peer_disconnected_timeout is not UNSET:
-            field_dict["peerDisconnectedTimeout"] = peer_disconnected_timeout
-        if peerless_purge_timeout is not UNSET:
-            field_dict["peerlessPurgeTimeout"] = peerless_purge_timeout
         if room_type is not UNSET:
             field_dict["roomType"] = room_type
         if video_codec is not UNSET:
@@ -68,10 +58,6 @@ def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
         d = src_dict.copy()
         max_peers = d.pop("maxPeers", UNSET)
 
-        peer_disconnected_timeout = d.pop("peerDisconnectedTimeout", UNSET)
-
-        peerless_purge_timeout = d.pop("peerlessPurgeTimeout", UNSET)
-
         _room_type = d.pop("roomType", UNSET)
         room_type: Union[Unset, RoomConfigRoomType]
         if isinstance(_room_type, Unset):
@@ -92,8 +78,6 @@ def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
 
         room_config = cls(
             max_peers=max_peers,
-            peer_disconnected_timeout=peer_disconnected_timeout,
-            peerless_purge_timeout=peerless_purge_timeout,
             room_type=room_type,
             video_codec=video_codec,
             webhook_url=webhook_url,

fishjam/_openapi_client/models/viewer_token.py

@@ -0,0 +1,60 @@
+from typing import Any, Dict, List, Type, TypeVar
+
+from attrs import define as _attrs_define
+from attrs import field as _attrs_field
+
+T = TypeVar("T", bound="ViewerToken")
+
+
+@_attrs_define
+class ViewerToken:
+    """Token for authorizing broadcaster viewer connection"""
+
+    token: str
+    """None"""
+    additional_properties: Dict[str, Any] = _attrs_field(init=False, factory=dict)
+    """@private"""
+
+    def to_dict(self) -> Dict[str, Any]:
+        """@private"""
+        token = self.token
+
+        field_dict: Dict[str, Any] = {}
+        field_dict.update(self.additional_properties)
+        field_dict.update(
+            {
+                "token": token,
+            }
+        )
+
+        return field_dict
+
+    @classmethod
+    def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
+        """@private"""
+        d = src_dict.copy()
+        token = d.pop("token")
+
+        viewer_token = cls(
+            token=token,
+        )
+
+        viewer_token.additional_properties = d
+        return viewer_token
+
+    @property
+    def additional_keys(self) -> List[str]:
+        """@private"""
+        return list(self.additional_properties.keys())
+
+    def __getitem__(self, key: str) -> Any:
+        return self.additional_properties[key]
+
+    def __setitem__(self, key: str, value: Any) -> None:
+        self.additional_properties[key] = value
+
+    def __delitem__(self, key: str) -> None:
+        del self.additional_properties[key]
+
+    def __contains__(self, key: str) -> bool:
+        return key in self.additional_properties

fishjam/api/_fishjam_client.py

@@ -12,6 +12,7 @@
 from fishjam._openapi_client.api.room import get_all_rooms as room_get_all_rooms
 from fishjam._openapi_client.api.room import get_room as room_get_room
 from fishjam._openapi_client.api.room import refresh_token as room_refresh_token
+from fishjam._openapi_client.api.viewer import generate_token as viewer_generate_token
 from fishjam._openapi_client.models import (
     AddPeerJsonBody,
     Peer,
@@ -23,6 +24,7 @@
     RoomCreateDetailsResponse,
     RoomDetailsResponse,
     RoomsListingResponse,
+    ViewerToken,
 )
 from fishjam._openapi_client.models.peer_options_web_rtc_metadata import (
     PeerOptionsWebRTCMetadata,
@@ -49,21 +51,13 @@ class RoomOptions:
 
     max_peers: int | None = None
     """Maximum amount of peers allowed into the room"""
-    peer_disconnected_timeout: int | None = None
-    """
-    Duration (in seconds) after which the peer will be removed if it is disconnected.
-    If not provided, this feature is disabled.
-    """
-    peerless_purge_timeout: int | None = None
-    """
-    Duration (in seconds) after which the room will be removed 
-    if no peers are connected. If not provided, this feature is disabled.
-    """
     video_codec: Literal["h264", "vp8"] | None = None
     """Enforces video codec for each peer in the room"""
     webhook_url: str | None = None
     """URL where Fishjam notifications will be sent"""
-    room_type: Literal["full_feature", "audio_only", "broadcaster"] = "full_feature"
+    room_type: Literal[
+        "full_feature", "audio_only", "broadcaster", "livestream"
+    ] = "full_feature"
     """The use-case of the room. If not provided, this defaults to full_feature."""
 
 
@@ -124,10 +118,11 @@ def create_room(self, options: RoomOptions | None = None) -> Room:
         if options.video_codec:
             codec = RoomConfigVideoCodec(options.video_codec)
 
+        if options.room_type == "livestream":
+            options.room_type = "broadcaster"
+
         config = RoomConfig(
             max_peers=options.max_peers,
-            peer_disconnected_timeout=options.peer_disconnected_timeout,
-            peerless_purge_timeout=options.peerless_purge_timeout,
             video_codec=codec,
             webhook_url=options.webhook_url,
             room_type=RoomConfigRoomType(options.room_type),
@@ -177,6 +172,15 @@ def refresh_peer_token(self, room_id: str, peer_id: str) -> str:
 
         return response.data.token
 
+    def create_livestream_viewer_token(self, room_id: str) -> str:
+        """Generates viewer token for livestream rooms"""
+
+        response = cast(
+            ViewerToken, self._request(viewer_generate_token, room_id=room_id)
+        )
+
+        return response.token
+
     def __parse_peer_metadata(self, metadata: dict | None) -> PeerOptionsWebRTCMetadata:
         peer_metadata = PeerOptionsWebRTCMetadata()
 

poetry_scripts.py

@@ -35,7 +35,7 @@ def run_tests():
 
 
 def run_local_test():
-    check_exit_code('poetry run pytest -m "not file_component_sources"')
+    check_exit_code('poetry run pytest -m "not file_component_sources" -vv')
 
 
 def run_formatter():

protos

@@ -154,45 +154,6 @@ async def test_peer_connected_disconnected(
         for event in event_checks:
             self.assert_event(event)
 
-    @pytest.mark.asyncio
-    async def test_peer_connected_disconnected_deleted(
-        self, room_api: FishjamClient, notifier: FishjamNotifier
-    ):
-        event_checks = [
-            ServerMessageRoomCreated,
-            ServerMessagePeerAdded,
-            ServerMessagePeerConnected,
-            ServerMessagePeerDisconnected,
-            ServerMessagePeerDeleted,
-            ServerMessageRoomDeleted,
-        ]
-
-        assert_task = asyncio.create_task(assert_events(notifier, event_checks.copy()))
-
-        notifier_task = asyncio.create_task(notifier.connect())
-        await notifier.wait_ready()
-
-        options = RoomOptions(
-            webhook_url=WEBHOOK_URL,
-            peerless_purge_timeout=2,
-            peer_disconnected_timeout=1,
-        )
-        room = room_api.create_room(options=options)
-
-        _peer, token = room_api.create_peer(room.id)
-
-        peer_socket = PeerSocket(fishjam_url=FISHJAM_URL, auto_close=True)
-        peer_task = asyncio.create_task(peer_socket.connect(token))
-
-        await peer_socket.wait_ready()
-
-        await assert_task
-        await cancel(peer_task)
-        await cancel(notifier_task)
-
-        for event in event_checks:
-            self.assert_event(event)
-
     @pytest.mark.asyncio
     async def test_peer_connected_room_deleted(
         self, room_api: FishjamClient, notifier: FishjamNotifier