transports: call parent stop() before disconnecting
This rollbacks a previous change https://github.com/pipecat-ai/pipecat/pull/855 which was trying to fix an issue in the wrong way. The reasoning behind this fix is that the parent class might be sending audio or messages (through the subclass) and if we disconnect before all the data is sent we will run into incomplete audio or even errors. Therefore, we first make sure the parent tasks stop and then it will be safe to disconnect.
This commit is contained in:
@@ -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}")
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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)):
|
||||||
|
|||||||
Reference in New Issue
Block a user