From 77ae07a71c693f701b478aa471fece8f54ac0fe5 Mon Sep 17 00:00:00 2001 From: zuczkows Date: Thu, 28 May 2026 12:50:51 +0200 Subject: [PATCH 1/2] Improve websocket handling - add lock --- livechat/utils/ws_client.py | 21 +++++++++++++-------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/livechat/utils/ws_client.py b/livechat/utils/ws_client.py index 13d0c90..0b3844b 100644 --- a/livechat/utils/ws_client.py +++ b/livechat/utils/ws_client.py @@ -8,7 +8,7 @@ import ssl import threading from time import sleep -from typing import List, NoReturn, Union +from typing import List, Union from loguru import logger from websocket import WebSocketApp, WebSocketConnectionClosedException @@ -19,7 +19,8 @@ def on_message(ws_client: WebSocketApp, message: str): ''' Custom WebSocketApp handler that inserts new messages in front of `self.messages` list. ''' - ws_client.messages.insert(0, json.loads(message)) + with ws_client._messages_lock: + ws_client.messages.insert(0, json.loads(message)) def on_close(ws_client: WebSocketApp, close_status_code: int, close_msg: str): @@ -39,6 +40,7 @@ class WebsocketClient(WebSocketApp): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.messages: List[dict] = [] + self._messages_lock = threading.Lock() self.on_message = on_message self.on_close = on_close self.on_error = on_error @@ -50,7 +52,7 @@ def open(self, ping_interval: Union[float, int] = 5, ws_conn_timeout: Union[float, int] = 10, keep_alive: bool = True, - response_timeout: Union[float, int] = 3) -> NoReturn: + response_timeout: Union[float, int] = 3) -> None: ''' Opens websocket connection and keep running forever. Args: origin (dict): Specifies origin while creating websocket connection. @@ -77,9 +79,10 @@ def open(self, 'ping_interval': ping_interval, } if keep_alive: - ping_thread = threading.Thread(target=self.run_forever, - kwargs=run_forever_kwargs) - ping_thread.start() + ws_thread = threading.Thread(target=self.run_forever, + kwargs=run_forever_kwargs, + daemon=True) + ws_thread.start() self._wait_till_sock_connected(ws_conn_timeout) return self.run_forever(**run_forever_kwargs) @@ -107,7 +110,9 @@ def send(self, request: dict, opcode=ABNF.OPCODE_TEXT) -> dict: def await_message(stop_event: threading.Event) -> dict: while not stop_event.is_set(): - for item in self.messages: + with self._messages_lock: + messages_snapshot = self.messages.copy() + for item in messages_snapshot: if item.get('request_id') == request_id and item.get( 'type') == 'response': return item @@ -131,7 +136,7 @@ def await_message(stop_event: threading.Event) -> dict: return RtmResponse(response) def _wait_till_sock_connected(self, - timeout: Union[float, int] = 10) -> NoReturn: + timeout: Union[float, int] = 10) -> None: ''' Polls until `self.sock` is connected. Args: timeout (float): timeout value in seconds, default 10. ''' From 5a861ffe117ae4b39db52b66d8e9d25da5a86b55 Mon Sep 17 00:00:00 2001 From: zuczkows Date: Fri, 29 May 2026 13:36:09 +0200 Subject: [PATCH 2/2] fixup! Improve websocket handling - add lock --- livechat/utils/ws_client.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/livechat/utils/ws_client.py b/livechat/utils/ws_client.py index 0b3844b..ead0f4b 100644 --- a/livechat/utils/ws_client.py +++ b/livechat/utils/ws_client.py @@ -79,10 +79,10 @@ def open(self, 'ping_interval': ping_interval, } if keep_alive: - ws_thread = threading.Thread(target=self.run_forever, + ping_thread = threading.Thread(target=self.run_forever, kwargs=run_forever_kwargs, daemon=True) - ws_thread.start() + ping_thread.start() self._wait_till_sock_connected(ws_conn_timeout) return self.run_forever(**run_forever_kwargs)