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

Merge pull request #882 from pipecat-ai/aleix/stop-transport-parent-first

Browse Source 
transports: call parent stop() before disconnecting
This commit is contained in:
Aleix Conchillo Flaqué
2024-12-18 19:35:39 -08:00
committed by GitHub
parent 0df0194cc1 5bfcac1f5c
commit 17decee788
3 changed files with 19 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/pipecat/transports/network/websocket_server.py
View File

@@ -72,14 +72,14 @@ class WebsocketServerInputTransport(BaseInputTransport):
self._server_task = self.get_event_loop().create_task(self._server_task_handler()) self._server_task = self.get_event_loop().create_task(self._server_task_handler())
async def stop(self, frame: EndFrame): async def stop(self, frame: EndFrame):
await super().stop(frame)
self._stop_server_event.set() self._stop_server_event.set()
await self._server_task await self._server_task
await super().stop(frame)
async def cancel(self, frame: CancelFrame): async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
self._stop_server_event.set() self._stop_server_event.set()
await self._server_task await self._server_task
await super().cancel(frame)
async def _server_task_handler(self): async def _server_task_handler(self):
logger.info(f"Starting websocket server on {self._host}:{self._port}") logger.info(f"Starting websocket server on {self._host}:{self._port}")

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

@@ -694,17 +694,8 @@ class DailyInputTransport(BaseInputTransport):
self._audio_in_task = self.get_event_loop().create_task(self._audio_in_task_handler()) self._audio_in_task = self.get_event_loop().create_task(self._audio_in_task_handler())
async def stop(self, frame: EndFrame): async def stop(self, frame: EndFrame):
# Leave the room.
await self._client.leave()
# Stop audio thread.
if self._audio_in_task and (self._params.audio_in_enabled or self._params.vad_enabled):
self._audio_in_task.cancel()
await self._audio_in_task
self._audio_in_task = None
# Parent stop. # Parent stop.
await super().stop(frame) await super().stop(frame)
async def cancel(self, frame: CancelFrame):
# Leave the room. # Leave the room.
await self._client.leave() await self._client.leave()
# Stop audio thread. # Stop audio thread.
@@ -712,8 +703,17 @@ class DailyInputTransport(BaseInputTransport):
self._audio_in_task.cancel() self._audio_in_task.cancel()
await self._audio_in_task await self._audio_in_task
self._audio_in_task = None self._audio_in_task = None
async def cancel(self, frame: CancelFrame):
# Parent stop. # Parent stop.
await super().cancel(frame) await super().cancel(frame)
# Leave the room.
await self._client.leave()
# Stop audio thread.
if self._audio_in_task and (self._params.audio_in_enabled or self._params.vad_enabled):
self._audio_in_task.cancel()
await self._audio_in_task
self._audio_in_task = None
async def cleanup(self): async def cleanup(self):
await super().cleanup() await super().cleanup()
@@ -817,16 +817,16 @@ class DailyOutputTransport(BaseOutputTransport):
await self._client.join() await self._client.join()
async def stop(self, frame: EndFrame): async def stop(self, frame: EndFrame):
# Leave the room.
await self._client.leave()
# Parent stop. # Parent stop.
await super().stop(frame) await super().stop(frame)
async def cancel(self, frame: CancelFrame):
# Leave the room. # Leave the room.
await self._client.leave() await self._client.leave()
async def cancel(self, frame: CancelFrame):
# Parent stop. # Parent stop.
await super().cancel(frame) await super().cancel(frame)
# Leave the room.
await self._client.leave()
async def cleanup(self): async def cleanup(self):
await super().cleanup() await super().cleanup()

8
src/pipecat/transports/services/livekit.py
View File

@@ -323,19 +323,19 @@ class LiveKitInputTransport(BaseInputTransport):
logger.info("LiveKitInputTransport started") logger.info("LiveKitInputTransport started")
async def stop(self, frame: EndFrame): async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._client.disconnect() await self._client.disconnect()
if self._audio_in_task: if self._audio_in_task:
self._audio_in_task.cancel() self._audio_in_task.cancel()
await self._audio_in_task await self._audio_in_task
await super().stop(frame)
logger.info("LiveKitInputTransport stopped") logger.info("LiveKitInputTransport stopped")
async def cancel(self, frame: CancelFrame): async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._client.disconnect() await self._client.disconnect()
if self._audio_in_task and (self._params.audio_in_enabled or self._params.vad_enabled): if self._audio_in_task and (self._params.audio_in_enabled or self._params.vad_enabled):
self._audio_in_task.cancel() self._audio_in_task.cancel()
await self._audio_in_task await self._audio_in_task
await super().cancel(frame)
def vad_analyzer(self) -> VADAnalyzer | None: def vad_analyzer(self) -> VADAnalyzer | None:
return self._vad_analyzer return self._vad_analyzer
@@ -397,13 +397,13 @@ class LiveKitOutputTransport(BaseOutputTransport):
logger.info("LiveKitOutputTransport started") logger.info("LiveKitOutputTransport started")
async def stop(self, frame: EndFrame): async def stop(self, frame: EndFrame):
await self._client.disconnect()
await super().stop(frame) await super().stop(frame)
await self._client.disconnect()
logger.info("LiveKitOutputTransport stopped") logger.info("LiveKitOutputTransport stopped")
async def cancel(self, frame: CancelFrame): async def cancel(self, frame: CancelFrame):
await self._client.disconnect()
await super().cancel(frame) await super().cancel(frame)
await self._client.disconnect()
async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
if isinstance(frame, (LiveKitTransportMessageFrame, LiveKitTransportMessageUrgentFrame)): if isinstance(frame, (LiveKitTransportMessageFrame, LiveKitTransportMessageUrgentFrame)):