Small websocket improvements
This commit is contained in:
1 parent
a79755c5b0
commit
89c5e55936
1 file changed
+4
-5
+4
-5
@@ -119,8 +119,8 @@ class Websocket:
|
|||||||
for delay in backoff:
|
for delay in backoff:
|
||||||
try:
|
try:
|
||||||
async with session.ws_connect(ws_url, ssl=True, proxy=proxy) as websocket:
|
async with session.ws_connect(ws_url, ssl=True, proxy=proxy) as websocket:
|
||||||
backoff.reset()
|
|
||||||
yield websocket
|
yield websocket
|
||||||
|
backoff.reset()
|
||||||
except aiohttp.ClientConnectionError:
|
except aiohttp.ClientConnectionError:
|
||||||
ws_logger.info(
|
ws_logger.info(
|
||||||
f"Websocket[{self._idx}] connection error (sleep: {delay:.3}s)", exc_info=True
|
f"Websocket[{self._idx}] connection error (sleep: {delay:.3}s)", exc_info=True
|
||||||
@@ -225,7 +225,7 @@ class Websocket:
|
|||||||
)
|
)
|
||||||
self._submitted.update(added)
|
self._submitted.update(added)
|
||||||
|
|
||||||
async def _gather_recv(self, messages: list[JsonType]):
|
async def _gather_recv(self, messages: list[JsonType], timeout: float = 0.5):
|
||||||
"""
|
"""
|
||||||
Gather incoming messages over the timeout specified.
|
Gather incoming messages over the timeout specified.
|
||||||
Note that there's no return value - this modifies `messages` in-place.
|
Note that there's no return value - this modifies `messages` in-place.
|
||||||
@@ -233,12 +233,11 @@ class Websocket:
|
|||||||
ws = self._ws.get_with_default(None)
|
ws = self._ws.get_with_default(None)
|
||||||
assert ws is not None
|
assert ws is not None
|
||||||
while True:
|
while True:
|
||||||
raw_message: aiohttp.WSMessage = await ws.receive()
|
raw_message: aiohttp.WSMessage = await ws.receive(timeout=timeout)
|
||||||
ws_logger.debug(f"Websocket[{self._idx}] received: {raw_message}")
|
ws_logger.debug(f"Websocket[{self._idx}] received: {raw_message}")
|
||||||
if raw_message.type is WSMsgType.TEXT:
|
if raw_message.type is WSMsgType.TEXT:
|
||||||
message: JsonType = json.loads(raw_message.data)
|
message: JsonType = json.loads(raw_message.data)
|
||||||
messages.append(message)
|
messages.append(message)
|
||||||
continue # shortcut to avoid checking all other elifs if not necessary
|
|
||||||
elif raw_message.type is WSMsgType.CLOSE:
|
elif raw_message.type is WSMsgType.CLOSE:
|
||||||
raise WebsocketClosed(received=True)
|
raise WebsocketClosed(received=True)
|
||||||
elif raw_message.type is WSMsgType.CLOSED:
|
elif raw_message.type is WSMsgType.CLOSED:
|
||||||
@@ -265,7 +264,7 @@ class Websocket:
|
|||||||
# listen over 0.5s for incoming messages
|
# listen over 0.5s for incoming messages
|
||||||
messages: list[JsonType] = []
|
messages: list[JsonType] = []
|
||||||
with suppress(asyncio.TimeoutError):
|
with suppress(asyncio.TimeoutError):
|
||||||
await asyncio.wait_for(self._gather_recv(messages), timeout=0.5)
|
await self._gather_recv(messages, timeout=0.5)
|
||||||
# process them
|
# process them
|
||||||
for message in messages:
|
for message in messages:
|
||||||
msg_type = message["type"]
|
msg_type = message["type"]
|
||||||
|
|||||||
Reference in new issue
Block a user