github-repos/blrec
1
0
Fork 0
mirror of https://github.com/acgnhiki/blrec.git synced 2025-04-02 17:10:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: refactor danmaku client

Browse Source
This commit is contained in:
acgnhik 2022-12-04 14:29:07 +08:00
parent 46811d4677
commit 08130d5e61
1 changed files with 31 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
src/blrec/bili/danmaku_client.py
View File

@ -52,7 +52,7 @@ class DanmakuClient(EventEmitter[DanmakuListener], AsyncStoppableMixin):
webapi: WebApi, webapi: WebApi,
room_id: int, room_id: int,
*, *,
max_retries: int = 10, max_retries: int = 60,
headers: Optional[Dict[str, str]] = None, headers: Optional[Dict[str, str]] = None,
) -> None: ) -> None:
super().__init__() super().__init__()
@ -237,7 +237,11 @@ class DanmakuClient(EventEmitter[DanmakuListener], AsyncStoppableMixin):
try: try:
await self._ws.send_bytes(data) await self._ws.send_bytes(data)
except Exception as exc: except Exception as exc:
logger.debug(f'Failed to send heartbeat due to: {repr(exc)}') logger.warning(f'Failed to send heartbeat: {repr(exc)}')
await self._emit('error_occurred', exc)
task = asyncio.create_task(self.restart())
task.add_done_callback(exception_callback)
break
await asyncio.sleep(self._HEARTBEAT_INTERVAL) await asyncio.sleep(self._HEARTBEAT_INTERVAL)
async def _create_message_loop(self) -> None: async def _create_message_loop(self) -> None:
@ -261,48 +265,53 @@ class DanmakuClient(EventEmitter[DanmakuListener], AsyncStoppableMixin):
await self._emit('danmaku_received', msg) await self._emit('danmaku_received', msg)
async def _receive(self) -> List[Dict[str, Any]]: async def _receive(self) -> List[Dict[str, Any]]:
self._retry_count = 0 self._reset_retry()
self._retry_delay = 0
while True: while True:
try: try:
wsmsg = await self._ws.receive(timeout=self._HEARTBEAT_INTERVAL) wsmsg = await self._ws.receive(timeout=self._HEARTBEAT_INTERVAL * 2)
except asyncio.TimeoutError as e:
logger.debug(f'Failed to receive message due to: {repr(e)}')
continue
except Exception as e: except Exception as e:
await self._handle_error(e) await self._handle_receive_error(e)
else: else:
if wsmsg.type == aiohttp.WSMsgType.BINARY: if wsmsg.type == aiohttp.WSMsgType.BINARY:
if result := await self._handle_data(wsmsg.data): if result := await self._handle_data(wsmsg.data):
return result return result
elif wsmsg.type == aiohttp.WSMsgType.ERROR: elif wsmsg.type == aiohttp.WSMsgType.ERROR:
await self._handle_error(cast(Exception, wsmsg.data)) await self._handle_receive_error(cast(Exception, wsmsg.data))
elif wsmsg.type == aiohttp.WSMsgType.CLOSED: elif wsmsg.type == aiohttp.WSMsgType.CLOSED:
msg = 'WebSocket Closed' msg = 'WebSocket Closed'
exc = aiohttp.WebSocketError(self._ws.close_code or 1006, msg) exc = aiohttp.WebSocketError(self._ws.close_code or 1006, msg)
await self._handle_error(exc) await self._handle_receive_error(exc)
else: else:
await self._handle_error(ValueError(wsmsg)) await self._handle_receive_error(ValueError(wsmsg))
@staticmethod @staticmethod
async def _handle_data(data: bytes) -> Optional[List[Dict[str, Any]]]: async def _handle_data(data: bytes) -> Optional[List[Dict[str, Any]]]:
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
op, msg = await loop.run_in_executor(None, Frame.decode, data)
if op == WS.OP_MESSAGE: try:
msg = cast(List[str], msg) op, msg = await loop.run_in_executor(None, Frame.decode, data)
return [json.loads(m) for m in msg] if op == WS.OP_MESSAGE:
elif op == WS.OP_HEARTBEAT_REPLY: msg = cast(List[str], msg)
return None return [json.loads(m) for m in msg]
else: elif op == WS.OP_HEARTBEAT_REPLY:
return None pass
except Exception as e:
logger.warning(f'Failed to handle data: {repr(e)}, data: {repr(data)}')
async def _handle_error(self, exc: Exception) -> None: return None
logger.debug(f'Failed to receive message due to: {repr(exc)}')
async def _handle_receive_error(self, exc: Exception) -> None:
logger.warning(f'Failed to receive message: {repr(exc)}')
await self._emit('error_occurred', exc) await self._emit('error_occurred', exc)
if isinstance(exc, asyncio.TimeoutError):
return
await self._retry() await self._retry()
def _reset_retry(self) -> None:
self._retry_count = 0
self._retry_delay = 0
async def _retry(self) -> None: async def _retry(self) -> None:
if self._retry_count < self._MAX_RETRIES: if self._retry_count < self._MAX_RETRIES:
if self._retry_delay > 0: if self._retry_delay > 0: