Adding support for renegotiation.

examples/aiortc/video-transform/client/typescript/src/smallWebRTCTransport.ts

@@ -27,6 +27,9 @@ export class SmallWebRTCTransport {
 
     private pc: RTCPeerConnection | null = null;
     private dc: RTCDataChannel | null = null;
+    private audioCodec: string | null = null;
+    private videoCodec: string | null = null;
+    private pc_id: string | null = null;
 
     constructor(callbacks: SmallWebRTCTransportCallbacks) {
         this._callbacks = callbacks
@@ -78,7 +81,7 @@ export class SmallWebRTCTransport {
         return pc;
     }
 
-    private async negotiate(audioCodec: string, videoCodec: string): Promise<void> {
+    private async negotiate(): Promise<void> {
         if (!this.pc) {
             return Promise.reject('Peer connection is not initialized');
         }
@@ -107,22 +110,25 @@ export class SmallWebRTCTransport {
             let codec: string;
 
             // Filter audio codec
-            if (audioCodec !== 'default') {
+            if (this.audioCodec && this.audioCodec !== 'default') {
                 // @ts-ignore
-                offerSdp.sdp = this.sdpFilterCodec('audio', audioCodec, offerSdp.sdp);
+                offerSdp.sdp = this.sdpFilterCodec('audio', this.audioCodec, offerSdp.sdp);
             }
 
             // Filter video codec
-            if (videoCodec !== 'default') {
+            if (this.videoCodec && this.videoCodec !== 'default') {
                 // @ts-ignore
-                offerSdp.sdp = this.sdpFilterCodec('video', videoCodec, offerSdp.sdp);
+                offerSdp.sdp = this.sdpFilterCodec('video', this.videoCodec, offerSdp.sdp);
             }
 
+            this.log(`Will create offer for peerId: ${this.pc_id}`)
+
             // Send offer to server
             const response = await fetch('/api/offer', {
                 body: JSON.stringify({
                     sdp: offerSdp.sdp,
                     type: offerSdp.type,
+                    pc_id: this.pc_id
                 }),
                 headers: {
                     'Content-Type': 'application/json',
@@ -132,6 +138,8 @@ export class SmallWebRTCTransport {
 
             const answer: RTCSessionDescriptionInit = await response.json();
             // @ts-ignore
+            this.pc_id = answer.pc_id
+            // @ts-ignore
             this.log(`Received answer for peer connection id ${answer.pc_id}`)
             await this.pc!.setRemoteDescription(answer);
         } catch (e) {
@@ -164,7 +172,9 @@ export class SmallWebRTCTransport {
         this.addInitialTransceivers();
         this.dc = this.createDataChannel('chat', { ordered: true });
         await this.addUserMedias(audioDevice, videoDevice);
-        await this.negotiate(audioCodec, videoCodec);
+        this.audioCodec = audioCodec
+        this.videoCodec = videoCodec
+        await this.negotiate();
     }
 
     private async addUserMedias(audioDevice: string|undefined, videoDevice:string|undefined): Promise<void> {
@@ -195,7 +205,7 @@ export class SmallWebRTCTransport {
 
             // Check if it's a signalling message
             if (messageObj.type === SIGNALLING_TYPE) {
-                this.handleSignallingMessage(messageObj as SignallingMessageObject); // Delegate to handleSignallingMessage
+                void this.handleSignallingMessage(messageObj as SignallingMessageObject); // Delegate to handleSignallingMessage
             } else {
                 // implement to handle the other messages in the future
             }
@@ -205,15 +215,14 @@ export class SmallWebRTCTransport {
     }
 
     // Method to handle signalling messages specifically
-    handleSignallingMessage(messageObj: SignallingMessageObject): void {
+    async handleSignallingMessage(messageObj: SignallingMessageObject): Promise<void> {
         // Cast the object to the correct type after verification
         const signallingMessage = messageObj as SignallingMessageObject;
 
         // Handle different signalling message types
         switch (signallingMessage.message) {
             case SignallingMessage.RENEGOTIATE:
-                this.log("Handling renegotiation...");
+                await this.negotiate()
-                // TODO: implement it
                 break;
 
             default:

examples/aiortc/video-transform/server/server.py

@@ -2,6 +2,7 @@ import argparse
 import asyncio
 import logging
 from contextlib import asynccontextmanager
+from typing import Dict
 
 import uvicorn
 from aiortc_bot import run_bot
@@ -17,33 +18,42 @@ logger = logging.getLogger("pc")
 
 app = FastAPI()
 
-
+# Store connections by pc_id
-pcs = set()
+pcs_map: Dict[str, SmallWebRTCConnection] = {}
 
 
 @app.post("/api/offer")
 async def offer(request: dict, background_tasks: BackgroundTasks):
-    pipecat_connection = SmallWebRTCConnection()
+    pc_id = request.get("pc_id")
-    await pipecat_connection.initialize(sdp=request["sdp"], type=request["type"])
 
-    pcs.add(pipecat_connection)
+    if pc_id and pc_id in pcs_map:
+        pipecat_connection = pcs_map[pc_id]
+        logger.info(f"Reusing existing connection for pc_id: {pc_id}")
+        await pipecat_connection.renegotiate(sdp=request["sdp"], type=request["type"])
+    else:
+        pipecat_connection = SmallWebRTCConnection()
+        await pipecat_connection.initialize(sdp=request["sdp"], type=request["type"])
 
-    @pipecat_connection.on("closed")
+        @pipecat_connection.on("closed")
-    async def handle_disconnected():
+        async def handle_disconnected(webrtc_connection: SmallWebRTCConnection):
-        logger.info("Discarding the peer connection.")
+            logger.info(f"Discarding peer connection for pc_id: {webrtc_connection.pc_id}")
-        pcs.discard(pipecat_connection)
+            pcs_map.pop(webrtc_connection.pc_id, None)
 
-    background_tasks.add_task(run_bot, pipecat_connection)
+        background_tasks.add_task(run_bot, pipecat_connection)
 
-    return pipecat_connection.get_answer()
+    answer = pipecat_connection.get_answer()
+    # Updating the peer connection inside the map
+    pcs_map[answer["pc_id"]] = pipecat_connection
+
+    return answer
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     yield  # Run app
-    coros = [pc.close() for pc in pcs]
+    coros = [pc.close() for pc in pcs_map.values()]
     await asyncio.gather(*coros)
-    pcs.clear()
+    pcs_map.clear()
 
 
 if __name__ == "__main__":

src/pipecat/transports/network/small_webrtc.py

@@ -156,17 +156,17 @@ class SmallWebRTCClient:
         self._pipecat_resampler = AudioResampler("s16", "mono", 16000)
 
         @self._webrtcConnection.on("connected")
-        async def on_connected():
+        async def on_connected(connection: SmallWebRTCConnection):
             logger.info("Peer connection established.")
             await self._handle_client_connected()
 
         @self._webrtcConnection.on("disconnected")
-        async def on_disconnected():
+        async def on_disconnected(connection: SmallWebRTCConnection):
             logger.info("Peer connection lost.")
             await self._handle_client_disconnected()
 
         @self._webrtcConnection.on("closed")
-        async def on_closed():
+        async def on_closed(connection: SmallWebRTCConnection):
             logger.info("Client connection closed.")
             await self._handle_client_closed()
 
@@ -188,9 +188,8 @@ class SmallWebRTCClient:
                 frame = await asyncio.wait_for(self._video_input_track.recv(), timeout=1.0)
             except asyncio.TimeoutError:
                 logger.warning("Timeout: No video frame received within the specified time.")
-                # TODO maybe we should ask to renegotiate in this case. Need to test.
-                # self._webrtcConnection.renegotiate()
                 frame = None
+                self._webrtcConnection.ask_to_renegotiate()
 
             if frame is None or not isinstance(frame, VideoFrame):
                 # If no valid frame, sleep for a bit

src/pipecat/transports/network/webrtc_connection.py

@@ -1,3 +1,4 @@
+import asyncio
 import json
 import uuid
 from enum import Enum
@@ -24,6 +25,7 @@ class SmallWebRTCConnection(EventEmitter):
         self._setup_listeners()
         self._tracks = set()
         self._data_channel = None
+        self._renegotiation_in_progress = False
 
     def _setup_listeners(self):
         @self.pc.on("datachannel")
@@ -41,7 +43,7 @@ class SmallWebRTCConnection(EventEmitter):
         @self.pc.on("connectionstatechange")
         async def on_connectionstatechange():
             logger.info(f"Connection state is {self.pc.connectionState}")
-            await self.emit(self.pc.connectionState)
+            await self.emit(self.pc.connectionState, self)
             if self.pc.connectionState == "failed":
                 await self.close()
 
@@ -57,7 +59,7 @@ class SmallWebRTCConnection(EventEmitter):
                 self._tracks.discard(track)
                 await self.emit("track-ended", track)
 
-    async def initialize(self, sdp: str, type: str):
+    async def _create_answer(self, sdp: str, type: str):
         offer = RTCSessionDescription(sdp=sdp, type=type)
         await self.pc.setRemoteDescription(offer)
 
@@ -67,11 +69,26 @@ class SmallWebRTCConnection(EventEmitter):
 
         self.answer = await self.pc.createAnswer()
 
-        return self.pc
+    async def initialize(self, sdp: str, type: str):
+        await self._create_answer(sdp, type)
 
     async def connect(self):
         await self.pc.setLocalDescription(self.answer)
 
+    async def renegotiate(self, sdp: str, type: str):
+        logger.info(f"Renegotiating {self.pc_id}")
+        await self._create_answer(sdp, type)
+        await self.pc.setLocalDescription(self.answer)
+
+        # TODO maybe we should refactor to receive a message from the client side when the renegotiation is completed.
+        # or look at the peer connection listeners
+        # but this is good enough for now for testing.
+        async def delayed_task():
+            await asyncio.sleep(2)
+            self._renegotiation_in_progress = False
+
+        asyncio.create_task(delayed_task())
+
     def force_transceivers_to_send_recv(self):
         for transceiver in self.pc.getTransceivers():
             transceiver.direction = "sendrecv"
@@ -149,7 +166,11 @@ class SmallWebRTCConnection(EventEmitter):
             json_message = json.dumps(message)
             self._data_channel.send(json_message)
 
-    def renegotiate(self):
+    def ask_to_renegotiate(self):
+        if self._renegotiation_in_progress:
+            return
+
+        self._renegotiation_in_progress = True
         self.send_app_message(
             {"type": SIGNALLING_TYPE, "message": SignallingMessage.RENEGOTIATE.value}
         )