feat: Replace websocket with asynchronous webrtc

README.md

@@ -1,9 +1,13 @@
 # Gooroomie
 
-Broadcasting test with websocket.
+> Webcam to browser.
+
+Broadcasting application with webRTC. 
+
 
 ## TO-DO
-- [ ] Replace image with video streaming.
+- [x] Replace image with video streaming.
+- [ ] Add audio streaming.
 - [ ] Add authentication.
 - [ ] Split server and client code.
 - [ ] Provision at aws with fully repeatable IaC.

app/Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.10.10-slim-buster
+FROM python:3.11.3-slim-buster
 
 # Install gcc.
 RUN apt-get update \
@@ -14,4 +14,4 @@ RUN pip install poetry --upgrade
 RUN poetry config virtualenvs.create false
 RUN poetry install --no-interaction --no-root
 
-ENTRYPOINT ["gunicorn", "-c", "python:gunicorn_config", "-k", "uvicorn.workers.UvicornWorker", "main:app"]
+ENTRYPOINT ["gunicorn", "-c", "python:gunicorn_config", "-k", "uvicorn.workers.UvicornWorker", "app.api:create_app"]

app/applications/http/api.py

@@ -0,0 +1,23 @@
+from fastapi import FastAPI
+
+from containers.container import Container
+
+from .pages.router import pages_router
+from .webrtc.router import web_rtc_router
+
+
+def create_app() -> FastAPI:
+    app = FastAPI()
+    app.container = Container()
+
+    # event handler
+    app.add_event_handler(
+        "shutdown",
+        app.container.webrtc_peer_connection_manager().close_all_peer_connections(),
+    )
+
+    # routing
+    app.include_router(pages_router)
+    app.include_router(web_rtc_router)
+
+    return app

app/applications/http/pages/responses.py

@@ -0,0 +1,5 @@
+from starlette.responses import Response
+
+
+class JavascriptResponse(Response):
+    media_type = "application/javascript"

app/applications/http/pages/router.py

@@ -0,0 +1,36 @@
+from dependency_injector.wiring import inject, Provide
+from fastapi import APIRouter, Depends, Request
+from starlette.responses import HTMLResponse
+
+from .responses import JavascriptResponse
+
+pages_router = APIRouter()
+
+
+@pages_router.get("/", response_class=HTMLResponse)
+@inject
+async def index(
+    request: Request,
+    templates=Depends(Provide["templates"]),
+):
+    return templates.TemplateResponse(
+        name="index.html",
+        context={
+            "request": request,
+            # add context here.
+        },
+    )
+
+
+@pages_router.get("/client.js", response_class=JavascriptResponse)
+@inject
+async def client_js(
+    request: Request,
+    templates=Depends(Provide["templates"]),
+):
+    return templates.TemplateResponse(
+        name="client.js",
+        context={
+            "request": request,
+        },
+    )

app/applications/http/webrtc/router.py

@@ -0,0 +1,23 @@
+from typing import TYPE_CHECKING
+
+from dependency_injector.wiring import inject, Provide
+from fastapi import APIRouter, Depends
+
+from .schemas import Description
+
+if TYPE_CHECKING:
+    from handlers.webrtc.peer_connection_manager import WebRTCPeerConnectionManager
+
+web_rtc_router = APIRouter(prefix="/webrtc/v1")
+
+
+@web_rtc_router.post("/offers")
+@inject
+async def offers(
+    body: Description,
+    webrtc_peer_connection_manager: "WebRTCPeerConnectionManager" = Depends(Provide["webrtc_peer_connection_manager"]),
+):
+    answer = await webrtc_peer_connection_manager.create_answer(
+        offer=body.to_rtc_session_description(),
+    )
+    return Description.from_rtc_session_description(answer)

app/applications/http/webrtc/schemas.py

@@ -0,0 +1,20 @@
+from aiortc import RTCSessionDescription
+from pydantic import BaseModel
+
+
+class Description(BaseModel):
+    sdp: str
+    type: str
+
+    @classmethod
+    def from_rtc_session_description(cls, rtc_session_description: RTCSessionDescription):
+        return cls(
+            sdp=rtc_session_description.sdp,
+            type=rtc_session_description.type,
+        )
+
+    def to_rtc_session_description(self):
+        return RTCSessionDescription(
+            sdp=self.sdp,
+            type=self.type,
+        )

app/config.yml

@@ -0,0 +1,4 @@
+webrtc:
+  peer_connection_manager:
+    audio_codec: ${WEBRTC_PEER_CONNECTION_MANAGER_AUDIO_CODEC:audio/opus}
+    video_codec: ${WEBRTC_PEER_CONNECTION_MANAGER_VIDEO_CODEC:video/H264}

app/containers/container.py

@@ -0,0 +1,24 @@
+from aiortc.contrib.media import MediaRelay
+from dependency_injector import containers, providers
+from starlette.templating import Jinja2Templates
+
+from handlers.webcam_streamer import WebcamStreamer
+from handlers.webrtc.peer_connection_manager import WebRTCPeerConnectionManager
+
+
+class Container(containers.DeclarativeContainer):
+    wiring_config = containers.WiringConfiguration(packages=["applications"])
+    config = providers.Configuration(yaml_files=["config.yml"])
+
+    video_media_relay = providers.Singleton(MediaRelay)
+    webcam_streamer = providers.Singleton(
+        WebcamStreamer,
+        video_media_relay=video_media_relay,
+    )
+    webrtc_peer_connection_manager = providers.Singleton(
+        WebRTCPeerConnectionManager,
+        webcam_streamer=webcam_streamer,
+        audio_codec=config.webrtc.peer_connection_manager.audio_codec,
+        video_codec=config.webrtc.peer_connection_manager.video_codec,
+    )
+    templates = providers.Resource(Jinja2Templates, directory="templates")

app/gunicorn_config.py

@@ -1,6 +1,6 @@
 workers = 1
-worker_class = 'sync'
+worker_class = "sync"
 timeout = 30
 threads = 1
 
-bind = '0.0.0.0:8080'
+bind = "0.0.0.0:8080"

app/handlers/webcam_streamer.py

@@ -0,0 +1,20 @@
+import platform
+
+from aiortc import MediaStreamTrack
+from aiortc.contrib.media import MediaPlayer, MediaRelay
+
+
+class WebcamStreamer:
+    def __init__(self, video_media_relay: MediaRelay):
+        self.video_media_relay = video_media_relay
+
+    def create_local_tracks(self, decode) -> tuple[MediaStreamTrack | None, MediaStreamTrack | None]:
+        options = dict(framerate="30", video_size="640x480")
+        match platform.system():
+            case "Darwin":
+                webcam = MediaPlayer("default:none", format="avfoundation", options=options)
+            case "Windows":
+                webcam = MediaPlayer("video=Integrated Camera", format="dshow", options=options)
+            case _:
+                webcam = MediaPlayer("/dev/video0", format="v4l2", options=options)
+        return None, self.video_media_relay.subscribe(webcam.video)

app/handlers/webrtc/__init__.py

@@ -0,0 +1 @@
+

app/handlers/webrtc/peer_connection.py

@@ -0,0 +1,12 @@
+import uuid
+from typing import Optional
+
+from aiortc import RTCConfiguration, RTCPeerConnection
+
+
+class PeerConnection(RTCPeerConnection):
+    connection_id: str
+
+    def __init__(self, configuration: Optional[RTCConfiguration] = None) -> None:
+        super().__init__(configuration)
+        self.connection_id = uuid.uuid4().__str__()

app/handlers/webrtc/peer_connection_manager.py

@@ -0,0 +1,57 @@
+import asyncio
+from typing import TYPE_CHECKING
+
+from aiortc import RTCRtpSender, RTCSessionDescription
+
+from handlers.webrtc.peer_connection import PeerConnection
+
+if TYPE_CHECKING:
+    from handlers.webcam_streamer import WebcamStreamer
+
+
+class WebRTCPeerConnectionManager:
+    def __init__(self, webcam_streamer: "WebcamStreamer", audio_codec: str, video_codec: str):
+        self.peer_connections: set[PeerConnection] = set()
+        self.webcam_streamer = webcam_streamer
+        self.audio_codec = audio_codec
+        self.video_codec = video_codec
+
+    def _create_new_peer_connection(self) -> PeerConnection:
+        peer_connection = PeerConnection()
+        self.peer_connections.add(peer_connection)
+
+        @peer_connection.on("connectionstatechange")
+        async def on_connection_state_change():
+            if peer_connection.connectionState == "failed":
+                await peer_connection.close()
+                self.peer_connections.remove(peer_connection)
+
+        return peer_connection
+
+    def _force_codec(self, peer_connection, sender, forced_codec):
+        kind = forced_codec.split("/")[0]
+        codecs = RTCRtpSender.getCapabilities(kind).codecs
+        transceiver = next(t for t in peer_connection.getTransceivers() if t.sender == sender)
+        transceiver.setCodecPreferences([codec for codec in codecs if codec.mimeType == forced_codec])
+
+    def add_track(self, peer_connection: PeerConnection):
+        audio_track, video_track = self.webcam_streamer.create_local_tracks(False)
+        if audio_track:
+            audio_sender = peer_connection.addTrack(audio_track)
+            self._force_codec(peer_connection, audio_sender, self.audio_codec)
+
+        if video_track:
+            video_sender = peer_connection.addTrack(video_track)
+            self._force_codec(peer_connection, video_sender, self.video_codec)
+
+    async def create_answer(self, offer) -> RTCSessionDescription:
+        pc = self._create_new_peer_connection()
+        self.add_track(pc)
+        await pc.setRemoteDescription(offer)
+        answer = await pc.createAnswer()
+        await pc.setLocalDescription(answer)
+        return pc.localDescription
+
+    async def close_all_peer_connections(self):
+        await asyncio.gather(*[pc.close() for pc in self.peer_connections])
+        self.peer_connections.clear()