Running flag fix and WS restart improvements

This commit is contained in:
DevilXD committed 2021-12-16 19:08:48 +01:00
1 parent 39363a0aee
commit 38365ab3f0
1 file changed
+10 -13
+10 -13
View File
@@ -86,10 +86,14 @@ class Websocket:
return return
if self._handle_task is None: if self._handle_task is None:
self._handle_task = asyncio.create_task(self._handle()) self._handle_task = asyncio.create_task(self._handle())
else:
self.request_reconnect()
await self.wait_until_connected() await self.wait_until_connected()
def start_nowait(self):
if self.connected:
return
if self._handle_task is None:
self._handle_task = asyncio.create_task(self._handle())
async def stop(self): async def stop(self):
if self._ws is not None: if self._ws is not None:
await self._ws.close() await self._ws.close()
@@ -97,17 +101,10 @@ class Websocket:
await self._handle_task await self._handle_task
self._handle_task = None self._handle_task = None
def start_nowait(self):
if self.connected:
return
if self._handle_task is None:
self._handle_task = asyncio.create_task(self._handle())
else:
self.request_reconnect()
def stop_nowait(self): def stop_nowait(self):
if self._ws is not None: if self._ws is not None:
asyncio.create_task(self._ws.close()) asyncio.create_task(self._ws.close())
# note: this detaches the handle task, so we have to assume it closes properly
self._handle_task = None self._handle_task = None
@task_wrapper @task_wrapper
@@ -280,7 +277,6 @@ class WebsocketPool:
return self._running.wait() return self._running.wait()
async def start(self): async def start(self):
self._running.set()
await self._twitch.wait_until_login() await self._twitch.wait_until_login()
# Add default topics # Add default topics
assert self._twitch._user_id is not None assert self._twitch._user_id is not None
@@ -289,6 +285,7 @@ class WebsocketPool:
WebsocketTopic("User", "Drops", user_id, self.process_drops), WebsocketTopic("User", "Drops", user_id, self.process_drops),
WebsocketTopic("User", "CommunityPoints", user_id, self.process_points), WebsocketTopic("User", "CommunityPoints", user_id, self.process_points),
]) ])
self._running.set()
await asyncio.gather(*(ws.start() for ws in self.websockets)) await asyncio.gather(*(ws.start() for ws in self.websockets))
async def stop(self): async def stop(self):
@@ -311,8 +308,8 @@ class WebsocketPool:
ws = self.websockets[ws_idx] ws = self.websockets[ws_idx]
else: else:
# create new # create new
ws = Websocket(self, len(self.websockets)) ws = Websocket(self, ws_idx)
if self._running: if self.running:
ws.start_nowait() ws.start_nowait()
self.websockets.append(ws) self.websockets.append(ws)
# ask websocket to take any topics it can - this modifies the set in-place # ask websocket to take any topics it can - this modifies the set in-place