Fix #558 #619 by revising RTMClient to accept simultaneous incoming messages

integration_tests/rtm/test_issue_558.py

@@ -27,7 +27,7 @@ def tearDown(self):
         # Reset the decorators by @RTMClient.run_on
         RTMClient._callbacks = collections.defaultdict(list)
 
-    @pytest.mark.skipif(condition=is_not_specified(), reason="still unfixed")
+    @pytest.mark.skipif(condition=is_not_specified(), reason="To avoid rate limited errors")
     @async_test
     async def test_issue_558(self):
         channel_id = os.environ[SLACK_SDK_TEST_RTM_TEST_CHANNEL_ID]
@@ -38,7 +38,7 @@ async def test_issue_558(self):
         async def process_messages(**payload):
             self.logger.debug(payload)
             self.message_count += 1
-            await asyncio.sleep(10)  # this blocks all handlers
+            await asyncio.sleep(10)  # this used to block all other handlers
 
         async def process_reactions(**payload):
             self.logger.debug(payload)
@@ -65,15 +65,15 @@ async def process_reactions(**payload):
 
             message = await web_client.chat_postMessage(channel=channel_id, text=text)
             self.assertFalse("error" in message)
-            # start blocking here
+            # used to start blocking here
 
             # This reaction_add event won't be handled due to a bug
             second_reaction = await web_client.reactions_add(channel=channel_id, timestamp=ts, name="tada")
             self.assertFalse("error" in second_reaction)
             await asyncio.sleep(2)
 
             self.assertEqual(self.message_count, 1)
-            self.assertEqual(self.reaction_count, 2)  # fails
+            self.assertEqual(self.reaction_count, 2)  # used to fail
         finally:
             if not rtm._stopped:
                 rtm.stop()

slack/rtm/client.py

@@ -8,7 +8,8 @@
 import concurrent
 import inspect
 import signal
-from typing import Optional, Callable, DefaultDict
+from asyncio import Future
+from typing import Optional, Callable, DefaultDict, List
 from ssl import SSLContext
 from threading import current_thread, main_thread
 
@@ -364,7 +365,12 @@ async def _connect_and_read(self):
 
     async def _read_messages(self):
         """Process messages received on the WebSocket connection."""
+        text_message_callback_executions: List[Future] = []
         while not self._stopped and self._websocket is not None:
+            for future in text_message_callback_executions:
+                if future.done():
+                    text_message_callback_executions.remove(future)
+
             try:
                 # Wait for a message to be received, but timeout after a second so that
                 # we can check if the socket has been closed, or if self._stopped is
@@ -384,11 +390,20 @@ async def _read_messages(self):
                 )
                 self._websocket = None
                 await self._dispatch_event(event="close")
+                num_of_running_callbacks = len(text_message_callback_executions)
+                if num_of_running_callbacks > 0:
+                    self._logger.info(
+                        "WebSocket connection has been closed "
+                        f"though {num_of_running_callbacks} callback executions were still in progress"
+                    )
                 return
+
             if message.type == aiohttp.WSMsgType.TEXT:
                 payload = message.json()
                 event = payload.pop("type", "Unknown")
-                await self._dispatch_event(event, data=payload)
+                # Asynchronously run callbacks to handle simultaneous incoming messages from Slack
+                f = asyncio.ensure_future(self._dispatch_event(event, data=payload))
+                text_message_callback_executions.append(f)
             elif message.type == aiohttp.WSMsgType.ERROR:
                 self._logger.error("Received an error on the websocket: %r", message)
                 await self._dispatch_event(event="error", data=message)