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

transports: on_client_disconnected only if remote client disconnects

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-09-15 11:35:40 -07:00
parent 22cbba002a
commit 8569b61598
3 changed files with 17 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
CHANGELOG.md
View File

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed ### Fixed
- Fixed a `FastAPIWebsocketTransport` and `SmallWebRTCTransport` issue where
`on_client_disconnected` would be triggered when the bot ends the
conversation. That is, `on_client_disconnected` should only be triggered when
the remote client actually disconnects.
- Fixed an issue in `HeyGenVideoService` where the `BotStartedSpeakingFrame` - Fixed an issue in `HeyGenVideoService` where the `BotStartedSpeakingFrame`
was blocked from moving through the Pipeline. was blocked from moving through the Pipeline.

6
src/pipecat/transports/smallwebrtc/transport.py
View File

@@ -478,7 +478,11 @@ class SmallWebRTCClient:
self._screen_video_track = None self._screen_video_track = None
self._audio_output_track = None self._audio_output_track = None
self._video_output_track = None self._video_output_track = None
await self._callbacks.on_client_disconnected(self._webrtc_connection)
# Trigger `on_client_disconnected` if the client actually disconnects,
# that is, we are not the ones disconnecting.
if not self._closing:
await self._callbacks.on_client_disconnected(self._webrtc_connection)
async def _handle_app_message(self, message: Any): async def _handle_app_message(self, message: Any):
"""Handle incoming application messages.""" """Handle incoming application messages."""

11
src/pipecat/transports/websocket/fastapi.py
View File

@@ -138,7 +138,6 @@ class FastAPIWebsocketClient:
): ):
logger.warning("Closing already disconnected websocket!") logger.warning("Closing already disconnected websocket!")
self._closing = True self._closing = True
await self.trigger_client_disconnected()
async def disconnect(self): async def disconnect(self):
"""Disconnect the WebSocket client.""" """Disconnect the WebSocket client."""
@@ -152,8 +151,6 @@ class FastAPIWebsocketClient:
await self._websocket.close() await self._websocket.close()
except Exception as e: except Exception as e:
logger.error(f"{self} exception while closing the websocket: {e}") logger.error(f"{self} exception while closing the websocket: {e}")
finally:
await self.trigger_client_disconnected()
async def trigger_client_disconnected(self): async def trigger_client_disconnected(self):
"""Trigger the client disconnected callback.""" """Trigger the client disconnected callback."""
@@ -298,7 +295,10 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
except Exception as e: except Exception as e:
logger.error(f"{self} exception receiving data: {e.__class__.__name__} ({e})") logger.error(f"{self} exception receiving data: {e.__class__.__name__} ({e})")
await self._client.trigger_client_disconnected() # Trigger `on_client_disconnected` if the client actually disconnects,
# that is, we are not the ones disconnecting.
if not self._client.is_closing:
await self._client.trigger_client_disconnected()
async def _monitor_websocket(self): async def _monitor_websocket(self):
"""Wait for self._params.session_timeout seconds, if the websocket is still open, trigger timeout event.""" """Wait for self._params.session_timeout seconds, if the websocket is still open, trigger timeout event."""
@@ -446,6 +446,9 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
async def _write_frame(self, frame: Frame): async def _write_frame(self, frame: Frame):
"""Serialize and send a frame through the WebSocket.""" """Serialize and send a frame through the WebSocket."""
if self._client.is_closing or not self._client.is_connected:
return
if not self._params.serializer: if not self._params.serializer:
return return