From d340d044e673847ead192ec463805e215aebe8f4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 14 Jan 2025 12:11:04 -0800 Subject: [PATCH] services(playht): ignore message from a different request id --- src/pipecat/services/playht.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index 7454bb16a..7d9fc30ad 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -130,6 +130,7 @@ class PlayHTTTSService(TTSService): self._websocket = None self._receive_task = None self._request_id = None + self._processing_request_id = None self._settings = { "sample_rate": sample_rate, @@ -243,9 +244,12 @@ class PlayHTTTSService(TTSService): await super()._handle_interruption(frame, direction) await self.stop_all_metrics() self._request_id = None + self._processing_request_id = None async def _receive_messages(self): async for message in self._get_websocket(): + if self._processing_request_id and self._request_id != self._processing_request_id: + continue if isinstance(message, bytes): # Skip the WAV header message if message.startswith(b"RIFF"): @@ -259,12 +263,14 @@ class PlayHTTTSService(TTSService): msg = json.loads(message) if msg.get("type") == "start": # Handle start of stream - logger.debug(f"Started processing request: {msg.get('request_id')}") + self._processing_request_id = msg.get("request_id") + logger.debug(f"Started processing request: {self._processing_request_id}") elif msg.get("type") == "end": # Handle end of stream if "request_id" in msg and msg["request_id"] == self._request_id: await self.push_frame(TTSStoppedFrame()) self._request_id = None + self._processing_request_id = None elif "error" in msg: logger.error(f"{self} error: {msg}") await self.push_error(ErrorFrame(f'{self} error: {msg["error"]}'))