update api, implement example

Author: Gawor270

examples/selective_subscription/__init__.py

@@ -0,0 +1 @@
+"""Selective subscription demo package."""

examples/selective_subscription/main.py

@@ -0,0 +1,36 @@
+from contextlib import asynccontextmanager
+
+import uvicorn
+from selective_subscription.app import app, room_service
+from selective_subscription.config import HOST, PORT
+from selective_subscription.notification_handler import NotificationHandler
+from selective_subscription.selective_subscription.worker import async_worker
+
+
+@asynccontextmanager
+async def lifespan(app):
+    """Application lifespan manager."""
+    async with async_worker() as worker:
+        notification_handler = NotificationHandler(room_service)
+        worker.run_in_background(notification_handler.start())
+        
+        print(f"Selective subscription demo started on http://{HOST}:{PORT}")
+        print("Available endpoints:")
+        print("  POST /api/peers - Create a new peer")
+        print("  GET /api/rooms/{room_name}/peers - Get available peers")
+        print("  POST /api/subscriptions - Toggle subscription")
+        
+        yield
+
+
+app.router.lifespan_context = lifespan
+
+
+if __name__ == "__main__":
+    uvicorn.run(
+        "main:app",
+        host=HOST,
+        port=PORT,
+        reload=True,
+        log_level="info"
+    )

examples/selective_subscription/pyproject.toml

@@ -0,0 +1,19 @@
+[project]
+name = "selective-subscription-demo"
+version = "0.1.0"
+description = "Selective subscription demo using Fishjam Python SDK"
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+    "starlette>=0.35.0",
+    "uvicorn>=0.25.0",
+    "jinja2>=3.1.0",
+    "python-multipart>=0.0.6",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["selective_subscription"]

examples/selective_subscription/selective_subscription/__init__.py

@@ -0,0 +1 @@
+"""Selective subscription demo package."""

examples/selective_subscription/selective_subscription/app.py

@@ -0,0 +1,152 @@
+import json
+from typing import Dict, Any
+
+from starlette.applications import Starlette
+from starlette.middleware import Middleware
+from starlette.middleware.cors import CORSMiddleware
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+from starlette.routing import Route
+from starlette.templating import Jinja2Templates
+
+from .room_service import RoomService
+
+
+# Initialize services
+room_service = RoomService()
+templates = Jinja2Templates(directory="templates")
+
+
+async def create_peer(request: Request) -> Response:
+    """Create a new peer in a room."""
+    try:
+        body = await request.json()
+        room_name = body.get("room_name")
+        peer_name = body.get("peer_name")
+
+        if not room_name or not peer_name:
+            return JSONResponse(
+                {"error": "room_name and peer_name are required"},
+                status_code=400
+            )
+
+        peer, token = room_service.create_peer()
+
+        return JSONResponse({
+            "peer_id": peer.id,
+            "token": token,
+            "room_name": room_name,
+            "peer_name": peer_name
+        })
+    
+    except Exception as e:
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+
+async def get_available_peers(request: Request) -> Response:
+    """Get peers available for subscription."""
+    room_name = request.path_params.get("room_name")
+    peer_id = request.query_params.get("peer_id")
+    
+    if not room_name:
+        return JSONResponse({"error": "room_name is required"}, status_code=400)
+    
+    peers = room_service.get_available_peers(room_name, peer_id)
+    
+    return JSONResponse({
+        "peers": [
+            {
+                "id": peer.id,
+                "metadata": peer.metadata if peer.metadata else {},
+                "tracks": [
+                    {
+                        "id": track["id"],
+                        "type": track.get("type", "unknown")
+                    }
+                    for track in peer.tracks
+                ]
+            }
+            for peer in peers
+        ]
+    })
+
+
+async def toggle_subscription(request: Request) -> Response:
+    """Toggle subscription to a peer's tracks."""
+    try:
+        body = await request.json()
+        peer_id = body.get("peer_id")
+        target_peer_id = body.get("target_peer_id")
+        
+        if not peer_id or not target_peer_id:
+            return JSONResponse(
+                {"error": "peer_id and target_peer_id are required"},
+                status_code=400
+            )
+        
+        subscribed = room_service.toggle_subscription(peer_id, target_peer_id)
+        
+        return JSONResponse({
+            "subscribed": subscribed,
+            "peer_id": peer_id,
+            "target_peer_id": target_peer_id
+        })
+    
+    except Exception as e:
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+
+async def get_subscription_status(request: Request) -> Response:
+    """Get current subscription status for a peer."""
+    peer_id = request.path_params.get("peer_id")
+    
+    if not peer_id:
+        return JSONResponse({"error": "peer_id is required"}, status_code=400)
+    
+    session = room_service.get_peer_session(peer_id)
+    
+    if not session:
+        return JSONResponse({"error": "Peer not found"}, status_code=404)
+    
+    return JSONResponse({
+        "peer_id": peer_id,
+        "subscribed_peers": list(session.subscribed_peers)
+    })
+
+
+async def health_check(request: Request) -> Response:
+    """Health check endpoint."""
+    return JSONResponse({"status": "OK"})
+
+
+async def serve_index(request: Request) -> Response:
+    """Serve the main HTML interface."""
+    return templates.TemplateResponse("index.html", {"request": request})
+
+
+# Define routes
+routes = [
+    Route("/", serve_index, methods=["GET"]),
+    Route("/health", health_check, methods=["GET"]),
+    Route("/api/peers", create_peer, methods=["POST"]),
+    Route("/api/rooms/{room_name}/peers", get_available_peers, methods=["GET"]),
+    Route("/api/subscriptions", toggle_subscription, methods=["POST"]),
+    Route("/api/peers/{peer_id}/subscriptions", get_subscription_status, methods=["GET"]),
+]
+
+# Define middleware
+middleware = [
+    Middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+]
+
+# Create application
+app = Starlette(
+    routes=routes,
+    middleware=middleware,
+)

examples/selective_subscription/selective_subscription/config.py

@@ -0,0 +1,7 @@
+import os
+
+FISHJAM_ID = os.getenv("FISHJAM_ID", "")
+FISHJAM_TOKEN = os.environ["FISHJAM_MANAGEMENT_TOKEN"]
+FISHJAM_URL = os.getenv("FISHJAM_URL", "http://localhost:5002")
+HOST = os.getenv("HOST", "localhost")
+PORT = int(os.getenv("PORT", "8000"))

examples/selective_subscription/selective_subscription/notification_handler.py

@@ -0,0 +1,60 @@
+from typing import Dict, Set
+import asyncio
+
+from .config import FISHJAM_ID, FISHJAM_TOKEN, FISHJAM_URL
+from .room_service import RoomService
+from fishjam._ws_notifier import FishjamNotifier
+from fishjam.events import (
+    ServerMessagePeerType,
+    ServerMessagePeerConnected,
+    ServerMessagePeerDisconnected,
+    ServerMessageTrackAdded,
+    ServerMessageTrackRemoved,
+)
+from fishjam.events.allowed_notifications import AllowedNotification
+
+
+class NotificationHandler:
+    """Handles Fishjam server notifications for selective subscription.
+    """
+    
+    def __init__(self, room_service: RoomService):
+        self.room_service = room_service
+        self._notifier = FishjamNotifier(FISHJAM_ID, FISHJAM_TOKEN)
+        @self._notifier.on_server_notification
+        async def _(notification: AllowedNotification):
+            match notification:
+                case ServerMessagePeerConnected(
+                    peer_type=ServerMessagePeerType.PEER_TYPE_WEBRTC,
+                ):
+                    await handle_peer_connected(notification)
+                case ServerMessagePeerDisconnected(
+                    peer_type=ServerMessagePeerType.PEER_TYPE_WEBRTC,
+                ):
+                    await handle_peer_disconnected(notification)
+                case ServerMessageTrackAdded(
+                    peer_type=ServerMessagePeerType.PEER_TYPE_WEBRTC,
+                ):
+                    await handle_track_added(notification)
+                case ServerMessageTrackRemoved(
+                    peer_type=ServerMessagePeerType.PEER_TYPE_WEBRTC,
+                ):
+                    await handle_track_removed(notification)
+
+        async def handle_peer_connected(notification: ServerMessagePeerConnected):
+            print(f"Peer connected: {notification.peer_id}")
+
+        async def handle_peer_disconnected(notification: ServerMessagePeerDisconnected):
+            print(f"Peer disconnected: {notification.peer_id}")
+
+        async def handle_track_added(notification: ServerMessageTrackAdded):
+            print(f"Track added: {notification.track}")
+
+        async def handle_track_removed(notification: ServerMessageTrackRemoved):
+            print(f"Track removed: {notification.track}")
+    
+    async def start(self) -> None:
+        """Long-running coroutine that connects the notifier and processes messages."""
+        await self._notifier.connect()
+            
+

examples/selective_subscription/selective_subscription/room_service.py

@@ -0,0 +1,70 @@
+from dataclasses import dataclass
+from typing import Dict, List, Optional
+
+from fishjam import FishjamClient, Peer, PeerOptions, Room, RoomOptions
+from fishjam.errors import NotFoundError
+
+from .config import FISHJAM_ID, FISHJAM_TOKEN, FISHJAM_URL
+
+
+class RoomService:
+    """Service for managing rooms and peer subscriptions."""
+    
+    def __init__(self):
+        self.fishjam = FishjamClient(
+            FISHJAM_ID,
+            FISHJAM_TOKEN,
+        )
+        self.room = self.fishjam.create_room(RoomOptions(
+            max_peers=10,
+            room_type="conference"
+        ))
+    
+    def get_or_create_room(self) -> Room:
+        """Get existing room or create a new one."""
+        if self.room:
+            try:
+                room = self.fishjam.get_room(self.room.id)
+                return room
+            except NotFoundError:
+                pass
+
+        return self.fishjam.create_room()
+    
+    def create_peer(self) -> tuple[Peer, str]:
+        """Create a peer with manual subscription mode."""
+        room = self.get_or_create_room()
+
+        options = PeerOptions(
+            subscribe_mode="manual",
+        )
+        
+        peer, token = self.fishjam.create_peer(room.id, options)
+        return peer, token
+
+    
+    def subscibe_peer(self, peer_id: str, target_peer_id: str):
+        """Subscribe a peer to all tracks of another peer."""
+        room = self.get_or_create_room()
+
+        self.fishjam.subscribe_peer(room.id, peer_id, target_peer_id)
+    
+    def subscribe_tracks(self, peer_id: str, track_ids: List[str]):
+        """Subscribe a peer to specific tracks."""
+        room = self.get_or_create_room()
+
+        self.fishjam.subscribe_tracks(room.id, peer_id, track_ids)
+
+    def get_peer_session(self, peer_id: str):
+        """Return a lightweight session-like object for example endpoints."""
+        room = self.get_or_create_room()
+        for p in room.peers:
+            if p.id == peer_id:
+                # create a simple object that has subscribed_peers attribute
+                class _Session:
+                    def __init__(self):
+                        self.subscribed_peers: set[str] = set()
+
+                return _Session()
+
+        return None

examples/selective_subscription/selective_subscription/worker.py

@@ -0,0 +1,31 @@
+from asyncio import Task, TaskGroup
+from contextlib import asynccontextmanager
+from typing import Any, Coroutine
+
+
+class BackgroundWorker:
+    def __init__(self, tg: TaskGroup) -> None:
+        self._tg = tg
+        self._tasks: set[Task[None]] = set()
+
+    def run_in_background(self, coro: Coroutine[Any, Any, None]):
+        task = self._tg.create_task(coro)
+        task.add_done_callback(self._remove_task)
+        self._tasks.add(task)
+        return task
+
+    def _remove_task(self, task: Task[None]):
+        self._tasks.discard(task)
+
+    def cleanup(self):
+        for task in self._tasks:
+            task.cancel()
+        self._tasks = set()
+
+
+@asynccontextmanager
+async def async_worker():
+    async with TaskGroup() as tg:
+        worker = BackgroundWorker(tg)
+        yield worker
+        worker.cleanup()