diff --git a/examples/aiortc/video-transform/client/typescript/src/smallWebRTCTransport.ts b/examples/aiortc/video-transform/client/typescript/src/smallWebRTCTransport.ts index cad68f5bc..05de049c0 100644 --- a/examples/aiortc/video-transform/client/typescript/src/smallWebRTCTransport.ts +++ b/examples/aiortc/video-transform/client/typescript/src/smallWebRTCTransport.ts @@ -157,7 +157,7 @@ export class SmallWebRTCTransport { await this.pc.setLocalDescription(offer); // Wait for ICE gathering to complete - await new Promise((resolve) => { + /*await new Promise((resolve) => { if (this.pc!.iceGatheringState === 'complete') { resolve(); } else { @@ -169,7 +169,7 @@ export class SmallWebRTCTransport { }; this.pc!.addEventListener('icegatheringstatechange', checkState); } - }); + });*/ let offerSdp = this.pc!.localDescription!; let codec: string; diff --git a/examples/aiortc/voice-agent/index.html b/examples/aiortc/voice-agent/index.html index 879a124a3..dea1bbccf 100644 --- a/examples/aiortc/voice-agent/index.html +++ b/examples/aiortc/voice-agent/index.html @@ -24,7 +24,7 @@ let connected = false let peerConnection = null - const waitForIceGatheringComplete = async (pc) => { + /*const waitForIceGatheringComplete = async (pc) => { if (pc.iceGatheringState === 'complete') return; return new Promise((resolve) => { const checkState = () => { @@ -35,14 +35,14 @@ }; pc.addEventListener('icegatheringstatechange', checkState); }); - } + }*/ const createSmallWebRTCConnection = async (audioTrack) => { const pc = new RTCPeerConnection() pc.ontrack = e => audioEl.srcObject = e.streams[0] pc.addTransceiver(audioTrack, { direction: 'sendrecv' }) await pc.setLocalDescription(await pc.createOffer()) - await waitForIceGatheringComplete(pc) + //await waitForIceGatheringComplete(pc) const offer = pc.localDescription const response = await fetch('/api/offer', { body: JSON.stringify({ sdp: offer.sdp, type: offer.type}), diff --git a/src/pipecat/transports/network/small_webrtc.py b/src/pipecat/transports/network/small_webrtc.py index a23baddb5..bb3044a75 100644 --- a/src/pipecat/transports/network/small_webrtc.py +++ b/src/pipecat/transports/network/small_webrtc.py @@ -285,7 +285,7 @@ class SmallWebRTCClient: self._params = _params async def connect(self): - if self._webrtcConnection.is_connected() or self._webrtcConnection.is_connecting(): + if self._webrtcConnection.is_connected(): # already initialized return @@ -304,6 +304,10 @@ class SmallWebRTCClient: self._webrtcConnection.send_app_message(frame.message) async def _handle_client_connected(self): + # There is nothing to do here yet, the pipeline is still not ready + if not self._params: + return + self._audio_input_track = self._webrtcConnection.audio_input_track() self._video_input_track = self._webrtcConnection.video_input_track() if self._params.audio_out_enabled: diff --git a/src/pipecat/transports/network/webrtc_connection.py b/src/pipecat/transports/network/webrtc_connection.py index 84a869cb1..6b4bf4d75 100644 --- a/src/pipecat/transports/network/webrtc_connection.py +++ b/src/pipecat/transports/network/webrtc_connection.py @@ -24,7 +24,7 @@ class SmallWebRTCConnection(EventEmitter): self.ice_servers = [RTCIceServer(urls=server) for server in ice_servers] else: self.ice_servers = [] - self._is_connecting = False + self._connect_invoked = False self._initialize() def _initialize(self): @@ -72,6 +72,12 @@ class SmallWebRTCConnection(EventEmitter): f"Ice connection state is {self.pc.iceConnectionState}, connection is {self.pc.connectionState}" ) + @self.pc.on("icegatheringstatechange") + async def on_icegatheringstatechange(): + logger.info( + f"Ice gathering state is {self.pc.iceGatheringState}" + ) + @self.pc.on("track") async def on_track(track): logger.info(f"Track {track.kind} received") @@ -92,14 +98,24 @@ class SmallWebRTCConnection(EventEmitter): # so we are basically forcing it to act this way self.force_transceivers_to_send_recv() - self.answer = await self.pc.createAnswer() + # this answer does not contain the ice candidates, which will be gathered later, after the setLocalDescription + logger.info(f"Creating answer") + local_answer = await self.pc.createAnswer() + await self.pc.setLocalDescription(local_answer) + logger.info(f"Setting the answer after the local description is created") + self.answer = self.pc.localDescription async def initialize(self, sdp: str, type: str): await self._create_answer(sdp, type) async def connect(self): - self._is_connecting = True - await self.pc.setLocalDescription(self.answer) + self._connect_invoked = True + # If we already connected, trigger again the connected event + if self.is_connected(): + await self.emit("connected", self) + # We are renegotiating here, because likely we have loose the first video frames + # and aiortc does not handle that pretty well. + self.ask_to_renegotiate() async def renegotiate(self, sdp: str, type: str, restart_pc: bool = False): logger.info(f"Renegotiating {self.pc_id}") @@ -114,7 +130,6 @@ class SmallWebRTCConnection(EventEmitter): self._initialize() await self._create_answer(sdp, type) - await self.pc.setLocalDescription(self.answer) # Maybe we should refactor to receive a message from the client side when the renegotiation is completed. # or look at the peer connection listeners @@ -158,7 +173,6 @@ class SmallWebRTCConnection(EventEmitter): async def close(self): if self.pc: await self.pc.close() - self._is_connecting = False def get_answer(self): if not self.answer: @@ -182,6 +196,11 @@ class SmallWebRTCConnection(EventEmitter): # So, there is no advantage in looking at self.pc.connectionState # That is why we are trying to keep our own state def is_connected(self): + # If the small webrtc transport has never invoked to connect + # we are acting like if we are not connected + if not self._connect_invoked: + return False + if self._last_received_time is None: # if we have never received a message, it is probably because the client has not created a data channel # so we are going to trust aiortc in this case @@ -189,9 +208,6 @@ class SmallWebRTCConnection(EventEmitter): # Checks if the last received ping was within the last 3 seconds. return (time.time() - self._last_received_time) < 3 - def is_connecting(self): - return self._is_connecting - def audio_input_track(self): # Transceivers always appear in creation-order for both peers # For now we are only considering that we are going to have 02 transceivers,