EaZyCode/twitch-miner-docker
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add chunking to the websocket topics management

Browse Source
This commit is contained in:
DevilXD committed 2025-11-19 21:24:08 +01:00
1 parent 5f54c26f4a
commit 5d7a697c2b
1 file changed
+19 -16
Download Patch File Download Diff File Expand all files Collapse all files
websocket.py
+19 -16
View File
@@ -14,6 +14,7 @@ from exceptions import MinerException, WebsocketClosed
from constants import PING_INTERVAL, PING_TIMEOUT, MAX_WEBSOCKETS, WS_TOPICS_LIMIT from constants import PING_INTERVAL, PING_TIMEOUT, MAX_WEBSOCKETS, WS_TOPICS_LIMIT
from utils import ( from utils import (
CHARS_ASCII, CHARS_ASCII,
chunk,
task_wrapper, task_wrapper,
create_nonce, create_nonce,
json_minify, json_minify,
@@ -210,30 +211,32 @@ class Websocket:
if removed: if removed:
topics_list = list(map(str, removed)) topics_list = list(map(str, removed))
ws_logger.debug(f"Websocket[{self._idx}]: Removing topics: {', '.join(topics_list)}") ws_logger.debug(f"Websocket[{self._idx}]: Removing topics: {', '.join(topics_list)}")
await self.send( for topics in chunk(topics_list, 20):
{ await self.send(
"type": "UNLISTEN", {
"data": { "type": "UNLISTEN",
"topics": topics_list, "data": {
"auth_token": auth_state.access_token, "topics": topics,
"auth_token": auth_state.access_token,
}
} }
} )
)
self._submitted.difference_update(removed) self._submitted.difference_update(removed)
# handle added topics # handle added topics
added = current.difference(self._submitted) added = current.difference(self._submitted)
if added: if added:
topics_list = list(map(str, added)) topics_list = list(map(str, added))
ws_logger.debug(f"Websocket[{self._idx}]: Adding topics: {', '.join(topics_list)}") ws_logger.debug(f"Websocket[{self._idx}]: Adding topics: {', '.join(topics_list)}")
await self.send( for topics in chunk(topics_list, 20):
{ await self.send(
"type": "LISTEN", {
"data": { "type": "LISTEN",
"topics": topics_list, "data": {
"auth_token": auth_state.access_token, "topics": topics,
"auth_token": auth_state.access_token,
}
} }
} )
)
self._submitted.update(added) self._submitted.update(added)
async def _gather_recv(self, messages: list[JsonType], timeout: float = 0.5): async def _gather_recv(self, messages: list[JsonType], timeout: float = 0.5):