Introduces ThreadPoolExecutor and updated unit test cases

Author: sujata-m

src/python_ms_core/core/topic/topic.py

@@ -1,10 +1,10 @@
 import json
 import logging
-import threading
 import time
 from .config.topic_config import Config
-from .abstract.topic_abstract import TopicAbstract
 from ..resource_errors import ExceptionHandler
+from concurrent.futures import ThreadPoolExecutor
+from .abstract.topic_abstract import TopicAbstract
 from ..queue.models.queue_message import QueueMessage
 
 logging.basicConfig(format='%(asctime)s %(levelname)-8s %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
@@ -18,6 +18,7 @@ def __init__(self, fn=None, max_concurrent_messages=1):
         self._max_concurrent_messages = max_concurrent_messages
         self._renewal_interval = 30  # seconds
         self.message_processing = 0
+        self.executor = ThreadPoolExecutor(max_workers=max_concurrent_messages)
 
     def _renew_message_lock(self, message, receiver):
         while True:
@@ -31,12 +32,9 @@ def _renew_message_lock(self, message, receiver):
     # Sends data to the callback function
     def process_message(self, message, receiver):
         queue_message = QueueMessage.data_from(str(message))
+        secondary_executor = ThreadPoolExecutor(max_workers=1)
         if not message._lock_expired:
-            lock_renewal_thread = threading.Thread(
-                target=self._renew_message_lock,
-                args=(message, receiver)
-            )
-            lock_renewal_thread.start()
+            secondary_executor.submit(self._renew_message_lock, message, receiver)
         self._function_to_call(queue_message)
 
     # Starts listening to the messages
@@ -61,9 +59,7 @@ def start_listening(self, provider, topic, subscription):
                                 continue
                             for message in messages:
                                 self.message_processing += 1
-                                threading.Thread(
-                                    target=self._process_message_in_thread,
-                                    args=(message, topic_receiver,)).start()
+                                self.executor.submit(self._process_message_in_thread, message, topic_receiver)
                         except Exception as et:
                             logger.error(f'Error in service bus connection: {et}')
                     else:
@@ -94,9 +90,7 @@ def __init__(self, config=None, topic_name=None, max_concurrent_messages=1):
     def subscribe(self, subscription=None, callback=None):
         if subscription is not None:
             cb = Callback(callback, max_concurrent_messages=self.max_concurrent_messages)
-            thread = threading.Thread(target=cb.start_listening, args=(self.provider, self.topic, subscription))
-            thread.start()
-            time.sleep(5)
+            cb.start_listening(self.provider, self.topic, subscription)
         else:
             logging.error(
                 f'Unimplemented initialize for core {self.provider.provider}, Subscription name is required!')

tests/unit_tests/test_topic/test_topic.py

@@ -19,22 +19,33 @@ def test_subscribe_with_subscription(self):
         mock_callback = MagicMock()
         mock_provider = MagicMock()
         mock_client = MagicMock()
-        mock_thread = MagicMock()
-
-        with patch('threading.Thread', return_value=mock_thread) as mock_thread_class:
-            topic = Topic(config=self.mock_config, topic_name='mock_topic')
-            topic.provider = mock_provider
-            mock_provider.client = mock_client
-
-            with patch.object(Callback, 'start_listening', return_value=None) as mock_start_listening:
-                topic.subscribe(subscription='mock_subscription', callback=mock_callback)
-
-                # Assertions on the objects
-                mock_thread.start.assert_called_once()
-                mock_thread_class.assert_called_once_with(
-                    target=mock_start_listening,
-                    args=(mock_provider, 'mock_topic', 'mock_subscription')
-                )
+
+        topic = Topic(config=self.mock_config, topic_name='mock_topic')
+        topic.provider = mock_provider
+        mock_provider.client = mock_client
+
+        with patch.object(Callback, 'start_listening', return_value=None) as mock_start_listening:
+            topic.subscribe(subscription='mock_subscription', callback=mock_callback)
+
+            # Assertions on the objects
+            mock_start_listening.assert_called_once_with(mock_provider, 'mock_topic', 'mock_subscription')
+
+    def test_subscribe_without_subscription(self):
+        mock_callback = MagicMock()
+        mock_provider = MagicMock()
+        mock_client = MagicMock()
+
+        topic = Topic(config=self.mock_config, topic_name='mock_topic')
+        topic.provider = mock_provider
+        mock_provider.client = mock_client
+
+        with patch('logging.error') as mock_logging_error:
+            topic.subscribe(subscription=None, callback=mock_callback)
+
+            # Assertions
+            mock_logging_error.assert_called_once_with(
+                f'Unimplemented initialize for core {mock_provider.provider}, Subscription name is required!'
+            )
 
     def test_publish(self):
         data = {'test': random.randint(0, 1000)}