Watchchannel: Moved message consuming task cancelling exception

Moved exception logging when cog is being unloaded and messages is still
not consumed from `cog_unload` to `consume_messages` itself in
try-except block to avoid case when requesting result too early
(before cancel finished).

1 files changed, 26 insertions, 27 deletions

diff --git a/bot/cogs/watchchannels/watchchannel.py b/bot/cogs/watchchannels/watchchannel.py
index 436778c46..d78d45f26 100644
--- a/bot/cogs/watchchannels/watchchannel.py
+++ b/bot/cogs/watchchannels/watchchannel.py
@@ -169,32 +169,38 @@ class WatchChannel(metaclass=CogABCMeta):
 
     async def consume_messages(self, delay_consumption: bool = True) -> None:
         """Consumes the message queues to log watched users' messages."""
-        if delay_consumption:
-            self.log.trace(f"Sleeping {BigBrotherConfig.log_delay} seconds before consuming message queue")
-            await asyncio.sleep(BigBrotherConfig.log_delay)
+        try:
+            if delay_consumption:
+                self.log.trace(f"Sleeping {BigBrotherConfig.log_delay} seconds before consuming message queue")
+                await asyncio.sleep(BigBrotherConfig.log_delay)
 
-        self.log.trace("Started consuming the message queue")
+            self.log.trace("Started consuming the message queue")
 
-        # If the previous consumption Task failed, first consume the existing comsumption_queue
-        if not self.consumption_queue:
-            self.consumption_queue = self.message_queue.copy()
-            self.message_queue.clear()
+            # If the previous consumption Task failed, first consume the existing comsumption_queue
+            if not self.consumption_queue:
+                self.consumption_queue = self.message_queue.copy()
+                self.message_queue.clear()
 
-        for user_channel_queues in self.consumption_queue.values():
-            for channel_queue in user_channel_queues.values():
-                while channel_queue:
-                    msg = channel_queue.popleft()
+            for user_channel_queues in self.consumption_queue.values():
+                for channel_queue in user_channel_queues.values():
+                    while channel_queue:
+                        msg = channel_queue.popleft()
 
-                    self.log.trace(f"Consuming message {msg.id} ({len(msg.attachments)} attachments)")
-                    await self.relay_message(msg)
+                        self.log.trace(f"Consuming message {msg.id} ({len(msg.attachments)} attachments)")
+                        await self.relay_message(msg)
 
-        self.consumption_queue.clear()
+            self.consumption_queue.clear()
 
-        if self.message_queue:
-            self.log.trace("Channel queue not empty: Continuing consuming queues")
-            self._consume_task = self.bot.loop.create_task(self.consume_messages(delay_consumption=False))
-        else:
-            self.log.trace("Done consuming messages.")
+            if self.message_queue:
+                self.log.trace("Channel queue not empty: Continuing consuming queues")
+                self._consume_task = self.bot.loop.create_task(self.consume_messages(delay_consumption=False))
+            else:
+                self.log.trace("Done consuming messages.")
+        except asyncio.CancelledError as e:
+            self.log.exception(
+                "The consume task was canceled. Messages may be lost.",
+                exc_info=e
+            )
 
     async def webhook_send(
         self,
@@ -330,10 +336,3 @@ class WatchChannel(metaclass=CogABCMeta):
         self.log.trace("Unloading the cog")
         if self._consume_task and not self._consume_task.done():
             self._consume_task.cancel()
-            try:
-                self._consume_task.result()
-            except asyncio.CancelledError as e:
-                self.log.exception(
-                    "The consume task was canceled. Messages may be lost.",
-                    exc_info=e
-                )