Added some basic connection problems handling

This commit is contained in:
DevilXD committed 2022-01-22 21:10:12 +01:00
1 parent 0227362537
commit 59cceaf474
2 files changed
+35 -16

No files matched your search

+11 -2
View File
@@ -9,6 +9,8 @@ from functools import cached_property
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import Any, Optional, SupportsInt, TYPE_CHECKING from typing import Any, Optional, SupportsInt, TYPE_CHECKING
import aiohttp
from exceptions import MinerException from exceptions import MinerException
from utils import Game, invalidate_cache from utils import Game, invalidate_cache
from constants import JsonType, BASE_URL, GQL_OPERATIONS, ONLINE_DELAY, DROPS_ENABLED_TAG from constants import JsonType, BASE_URL, GQL_OPERATIONS, ONLINE_DELAY, DROPS_ENABLED_TAG
@@ -304,5 +306,12 @@ class Channel:
if self._spade_url is None: if self._spade_url is None:
self._spade_url = await self.get_spade_url() self._spade_url = await self.get_spade_url()
logger.debug(f"Sending minute-watched to {self.name}") logger.debug(f"Sending minute-watched to {self.name}")
async with self._twitch._session.post(self._spade_url, data=self._payload) as response: for attempt in range(5):
return response.status == 204 try:
async with self._twitch._session.post(
self._spade_url, data=self._payload
) as response:
return response.status == 204
except (aiohttp.ClientConnectionError, aiohttp.ServerTimeoutError):
continue
return False
+24 -14
View File
@@ -7,6 +7,7 @@ from yarl import URL
from time import time from time import time
from itertools import chain from itertools import chain
from functools import partial from functools import partial
from contextlib import suppress
from typing import ( from typing import (
Optional, Union, List, Dict, Set, OrderedDict, Callable, Iterable, cast, TYPE_CHECKING Optional, Union, List, Dict, Set, OrderedDict, Callable, Iterable, cast, TYPE_CHECKING
) )
@@ -309,13 +310,25 @@ class Twitch:
break break
await self._state_change.wait() await self._state_change.wait()
async def _watch_sleep(self, delay: float) -> None:
# we use wait_for here to allow an asyncio.sleep that can be ended prematurely,
# without cancelling the containing task
self._watching_restart.clear()
with suppress(asyncio.TimeoutError):
await asyncio.wait_for(self._watching_restart.wait(), timeout=delay)
@task_wrapper @task_wrapper
async def _watch_loop(self) -> None: async def _watch_loop(self) -> None:
interval = WATCH_INTERVAL.total_seconds() interval = WATCH_INTERVAL.total_seconds()
i = 1 i = 1
while True: while True:
channel = await self.watching_channel.get() channel = await self.watching_channel.get()
await channel.send_watch() succeeded = await channel.send_watch()
if not succeeded:
# this usually means there are connection problems
self.gui.print()
await self._watch_sleep(60)
continue
last_watch = time() last_watch = time()
self._drop_update = asyncio.Future() self._drop_update = asyncio.Future()
use_active = False use_active = False
@@ -359,15 +372,7 @@ class Twitch:
# cleanup channels every hour # cleanup channels every hour
self.change_state(State.CHANNELS_CLEANUP) self.change_state(State.CHANNELS_CLEANUP)
i = (i + 1) % 3600 i = (i + 1) % 3600
# we use wait_for here to allow an asyncio.sleep that can be ended prematurely, await self._watch_sleep(last_watch + interval - time())
# without cancelling the containing task
self._watching_restart.clear()
try:
await asyncio.wait_for(
self._watching_restart.wait(), timeout=last_watch + interval - time()
)
except asyncio.TimeoutError:
pass
def can_watch(self, channel: Channel) -> bool: def can_watch(self, channel: Channel) -> bool:
if self.game is None: if self.game is None:
@@ -712,10 +717,15 @@ class Twitch:
"Client-Id": CLIENT_ID, "Client-Id": CLIENT_ID,
} }
gql_logger.debug(f"GQL Request: {op}") gql_logger.debug(f"GQL Request: {op}")
async with self._session.post(GQL_URL, json=op, headers=headers) as response: for attempt in range(5):
response_json = await response.json() try:
gql_logger.debug(f"GQL Response: {response_json}") async with self._session.post(GQL_URL, json=op, headers=headers) as response:
return response_json response_json = await response.json()
gql_logger.debug(f"GQL Response: {response_json}")
return response_json
except (aiohttp.ClientConnectionError, aiohttp.ServerTimeoutError):
continue
raise RuntimeError(f"Ran out of attempts while handling a GQL request: {op}")
async def fetch_campaign(self, campaign_id: str, claimed_benefits: Set[str]) -> DropsCampaign: async def fetch_campaign(self, campaign_id: str, claimed_benefits: Set[str]) -> DropsCampaign:
response = await self.gql_request( response = await self.gql_request(