From 5f37f666c06df28757f1e973efb0cac5bfc1d3b7 Mon Sep 17 00:00:00 2001 From: Kazuhiro Sera Date: Fri, 1 May 2020 15:59:40 +0900 Subject: [PATCH] Fix #558 by revising RTMClient to accept simultaneous incoming messages --- integration_tests/rtm/test_issue_558.py | 8 ++++---- slack/rtm/client.py | 19 +++++++++++++++++-- 2 files changed, 21 insertions(+), 6 deletions(-) diff --git a/integration_tests/rtm/test_issue_558.py b/integration_tests/rtm/test_issue_558.py index a4c9992d9..e20b43d45 100644 --- a/integration_tests/rtm/test_issue_558.py +++ b/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,7 +65,7 @@ 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") @@ -73,7 +73,7 @@ async def process_reactions(**payload): 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() diff --git a/slack/rtm/client.py b/slack/rtm/client.py index 8fef13671..5fc88ad72 100644 --- a/slack/rtm/client.py +++ b/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)