wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #805 from pipecat-ai/khk/parallel-pipeline-fix

Browse Source 
Check to avoid double-join in ParallelPipeline case
This commit is contained in:
Kwindla Hultman Kramer
2024-12-07 21:49:16 -08:00
committed by GitHub
parent e9f5fe0f37 315b7fcc34
commit c6b401b5d1
1 changed files with 5 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/pipecat/transports/services/daily.py
View File

@@ -200,6 +200,7 @@ class DailyTransportClient(EventHandler):
self._other_participant_has_joined = False self._other_participant_has_joined = False
self._joined = False self._joined = False
self._joining = False
self._leave_counter = 0 self._leave_counter = 0
self._executor = ThreadPoolExecutor(max_workers=5) self._executor = ThreadPoolExecutor(max_workers=5)
@@ -314,11 +315,12 @@ class DailyTransportClient(EventHandler):
async def join(self): async def join(self):
# Transport already joined, ignore. # Transport already joined, ignore.
if self._joined: if self._joined or self._joining:
# Increment leave counter if we already joined. # Increment leave counter if we already joined.
self._leave_counter += 1 self._leave_counter += 1
return return
self._joining = True
logger.info(f"Joining {self._room_url}") logger.info(f"Joining {self._room_url}")
# For performance reasons, never subscribe to video streams (unless a # For performance reasons, never subscribe to video streams (unless a
@@ -351,10 +353,11 @@ class DailyTransportClient(EventHandler):
error_msg = f"Time out joining {self._room_url}" error_msg = f"Time out joining {self._room_url}"
logger.error(error_msg) logger.error(error_msg)
await self._callbacks.on_error(error_msg) await self._callbacks.on_error(error_msg)
self._joining = False
async def _start_transcription(self): async def _start_transcription(self):
if not self._token: if not self._token:
logger.warning(f"Transcription can't be started without a room token") logger.warning("Transcription can't be started without a room token")
return return
logger.info(f"Enabling transcription with settings {self._params.transcription_settings}") logger.info(f"Enabling transcription with settings {self._params.transcription_settings}")