almost working with dashboard

Author: Gawor270

examples/selective_subscription/main.py

@@ -4,21 +4,20 @@
 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
+from 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")
+        print("  POST /api/subscribe_peer - subscribe to all tracks of a peer")
+        print("  POST /api/subscribe_tracks - subscribe to specific tracks")
 
         yield
 

examples/selective_subscription/selective_subscription/app.py

@@ -1,4 +1,5 @@
 import json
+from pathlib import Path
 from typing import Dict, Any
 
 from starlette.applications import Starlette
@@ -12,13 +13,11 @@
 from .room_service import RoomService
 
 
-# Initialize services
 room_service = RoomService()
-templates = Jinja2Templates(directory="templates")
+templates = Jinja2Templates(directory=str(Path(__file__).resolve().parent.parent / "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")
@@ -42,99 +41,60 @@ async def create_peer(request: Request) -> Response:
     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."""
+async def subscribe_peer(request: Request) -> Response:
     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
-        })
+
+        room_service.subscibe_peer(peer_id, target_peer_id)
+
+        return JSONResponse({"status": "subscribed"})
 
     except Exception as e:
         return JSONResponse({"error": str(e)}, status_code=500)
+    
+async def subscribe_tracks(request: Request) -> Response:
+    try:
+        body = await request.json()
+        peer_id = body.get("peer_id")
+        track_ids = body.get("track_ids")
 
+        if not peer_id or not track_ids:
+            return JSONResponse(
+                {"error": "peer_id and track_ids are required"},
+                status_code=400
+            )
 
-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)
-    })
+        room_service.subscribe_tracks(peer_id, track_ids)
 
+        return JSONResponse({"status": "subscribed"})
+    
+    except Exception as e:
+        return JSONResponse({"error": str(e)}, status_code=500)
 
 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"]),
+    Route("/api/subscribe_peer", subscribe_peer, methods=["POST"]),
+    Route("/api/subscribe_tracks", subscribe_tracks, methods=["POST"]),
 ]
 
-# Define middleware
 middleware = [
     Middleware(
         CORSMiddleware,
@@ -145,7 +105,6 @@ async def serve_index(request: Request) -> Response:
     )
 ]
 
-# Create application
 app = Starlette(
     routes=routes,
     middleware=middleware,

examples/selective_subscription/selective_subscription/config.py

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

examples/selective_subscription/selective_subscription/notification_handler.py

@@ -1,7 +1,4 @@
-from typing import Dict, Set
-import asyncio
-
-from .config import FISHJAM_ID, FISHJAM_TOKEN, FISHJAM_URL
+from .config import FISHJAM_ID, FISHJAM_TOKEN
 from .room_service import RoomService
 from fishjam._ws_notifier import FishjamNotifier
 from fishjam.events import (
@@ -15,8 +12,6 @@
 
 
 class NotificationHandler:
-    """Handles Fishjam server notifications for selective subscription.
-    """
 
     def __init__(self, room_service: RoomService):
         self.room_service = room_service

examples/selective_subscription/selective_subscription/room_service.py

@@ -1,14 +1,12 @@
-from dataclasses import dataclass
-from typing import Dict, List, Optional
+from typing import List
 
 from fishjam import FishjamClient, Peer, PeerOptions, Room, RoomOptions
 from fishjam.errors import NotFoundError
 
-from .config import FISHJAM_ID, FISHJAM_TOKEN, FISHJAM_URL
+from .config import FISHJAM_ID, FISHJAM_TOKEN
 
 
 class RoomService:
-    """Service for managing rooms and peer subscriptions."""
 
     def __init__(self):
         self.fishjam = FishjamClient(
@@ -21,7 +19,6 @@ def __init__(self):
         ))
 
     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)
@@ -32,7 +29,6 @@ def get_or_create_room(self) -> Room:
         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(
@@ -44,23 +40,19 @@ def create_peer(self) -> tuple[Peer, str]:
 
 
     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()