Change join_channels method

join_channels now chunks into groups of 20 rather than a single list relying on tick rates.
Adjust timeouts based on list length.

Author: chillymosh

docs/changelog.rst

@@ -9,9 +9,10 @@ Master
         - Added repr for :class:`~twitchio.CustomReward`
     - Bug fixes
         - Added ``self.registered_callbacks = {}`` to :func:`~twitchio.Client.from_client_credentials`
-        - Allow empty or missing initial_channels to trigger :func:`~twitchio.Client.event_ready`.
+        - Allow empty or missing initial_channels to trigger :func:`~twitchio.Client.event_ready`
         - Corrected :func:`twitchio.CustomRewardRedemption.fulfill` endpoint typo and creation
         - Corrected :func:`twitchio.CustomRewardRedemption.refund` endpoint typo and creation
+        - Changed :func:`~twitchio.Client.join_channels` logic to handle bigger channel lists better
 
 - ext.commands
     - Bug fixes

twitchio/websocket.py

@@ -249,6 +249,16 @@ async def part_channels(self, *channels: str):
             if self._retain_cache:
                 self._cache.pop(channel, None)
 
+    def _assign_timeout(self, channel_count: int):
+        if channel_count <= 40:
+            return 30
+        elif channel_count <= 60:
+            return 40
+        elif channel_count <= 80:
+            return 50
+        else:
+            return 60
+
     async def join_channels(self, *channels: str):
         """|coro|
 
@@ -259,26 +269,29 @@ async def join_channels(self, *channels: str):
         *channels : str
             An argument list of channels to attempt joining.
         """
-        async with self._join_lock:  # acquire a lock, allowing only one join_channels at once...
-            for channel in channels:
-                if self._join_handle < time.time():  # Handle is less than the current time
-                    self._join_tick = 20  # So lets start a new rate limit bucket..
-                    self._join_handle = time.time() + 11  # Set the handle timeout time
-                if self._join_tick == 0:  # We have exhausted the bucket, wait so we can make a new one...
-                    await asyncio.sleep(self._join_handle - time.time())
-                asyncio.create_task(self._join_channel(channel))
-                self._join_tick -= 1
-
-    async def _join_channel(self, entry):
+        async with self._join_lock:
+            channel_count = len(channels)
+            if channel_count > 20:
+                timeout = self._assign_timeout(channel_count)
+                chunks = [channels[i : i + 20] for i in range(0, len(channels), 20)]
+                for chunk in chunks:
+                    for channel in chunk:
+                        asyncio.create_task(self._join_channel(channel, timeout))
+                    await asyncio.sleep(11)
+            else:
+                for channel in channels:
+                    asyncio.create_task(self._join_channel(channel, 11))
+
+    async def _join_channel(self, entry: str, timeout: int):
         channel = re.sub("[#]", "", entry).lower()
         await self.send(f"JOIN #{channel}\r\n")
 
         self._join_pending[channel] = fut = self._loop.create_future()
-        asyncio.create_task(self._join_future_handle(fut, channel))
+        asyncio.create_task(self._join_future_handle(fut, channel, timeout))
 
-    async def _join_future_handle(self, fut: asyncio.Future, channel: str):
+    async def _join_future_handle(self, fut: asyncio.Future, channel: str, timeout: int):
         try:
-            await asyncio.wait_for(fut, timeout=11)
+            await asyncio.wait_for(fut, timeout=timeout)
         except asyncio.TimeoutError:
             log.error(f'The channel "{channel}" was unable to be joined. Check the channel is valid.')
             self._join_pending.pop(channel)