# Version 0.0.21

### New Features and Enhancements
- **Message Lock Renewal:** Implemented a mechanism to automatically renew message locks during processing. This ensures that messages remain active and are not returned to the queue for reprocessing while they are being handled.
- **Concurrent Message Processing:** Enhanced the system to process messages concurrently using a number of worker threads equal to the number of available CPU cores by default. Users can override this default by specifying the `max_concurrent_messages` parameter, for example, `core.get_topic(topic_name=topic_name, max_concurrent_messages=10)`. This optimization leverages system resources for improved performance and throughput.
- **Completion Acknowledgement:** Updated the processing flow to wait until message processing is fully completed before sending the acknowledgement of message completion. This change ensures reliable processing and accurate message handling.
- **Version Tracking:** Introduced a `version.py` file to maintain and track the package version. This addition facilitates version control and package management.
- **Unit Test Updates:** Updated unit test cases to cover the new features and enhancements, ensuring robust testing and quality assurance.
- **Documentation Update:** Updated the README file to reflect the new features and enhancements, providing clearer guidance and information for users.

Author: sujata-m

CHANGELOG.md

@@ -1,10 +1,15 @@
 # Change log
 
-# 0.0.21
-- Introduced renew message lock, which ensures that message is alive while processing and not returned to the queue to be reprocessed.  
-- Added max concurrent messages login which can handle multiple requests at once  
-- Added version.py file to maintain the package version
-- Updated readme
+# Version 0.0.21
+### New Features and Enhancements
+- **Message Lock Renewal:** Implemented a mechanism to automatically renew message locks during processing. This ensures that messages remain active and are not returned to the queue for reprocessing while they are being handled.  
+- **Concurrent Message Processing:** Enhanced the system to process messages concurrently using a number of worker threads equal to the number of available CPU cores by default. Users can override this default by specifying the `max_concurrent_messages` parameter, for example, `core.get_topic(topic_name=topic_name, max_concurrent_messages=10)`. This optimization leverages system resources for improved performance and throughput. 
+- **Completion Acknowledgement:** Updated the processing flow to wait until message processing is fully completed before sending the acknowledgement of message completion. This change ensures reliable processing and accurate message handling. 
+- **Version Tracking:** Introduced a `version.py` file to maintain and track the package version. This addition facilitates version control and package management.
+- **Unit Test Updates:** Updated unit test cases to cover the new features and enhancements, ensuring robust testing and quality assurance.
+- **Documentation Update:** Updated the README file to reflect the new features and enhancements, providing clearer guidance and information for users.
+
+
 
 # 0.0.18
 - Adds extra checks in service bus for retries

README.md

@@ -112,7 +112,10 @@ Topic can be accessed by the core method `get_topic`. This method takes two para
 from python_ms_core import Core
 
 core = Core()
-topic = core.get_topic(topic_name='topicName')
+topic = core.get_topic(topic_name='topicName') # By default, process messages concurrently which are available CPU cores  
+topic = core.get_topic(topic_name='topicName', max_concurrent_messages=10) # Process 10 messages concurrently  
+
+
 
 ```
 

src/example.py

@@ -1,40 +1,41 @@
 # Testing code
-import sys
 import os
 import time
 import uuid
 import random
-import datetime
-from io import BytesIO, StringIO
+import threading
 
 from python_ms_core import Core
 from python_ms_core.core.queue.models.queue_message import QueueMessage
-from python_ms_core.core.auth.models.permission_request import PermissionRequest
-
-
 
 core = Core()
 print(f'Core version: {Core.__version__}')
-topic = 'gtfs-pathways-upload'
-subscription = 'log'
-some_other_sub = 'usdufs'
+topic = 'temp-request'
+subscription = 'temp'
+some_other_sub = 'temp'
 
 
 def publish_messages(topic_name):
     topic_object = core.get_topic(topic_name=topic_name)
     queue_message = QueueMessage.data_from({
         'message': str(uuid.uuid4().hex),
-        'data': {'a': random.randint(0, 1000)}
+        'data': {'a': random.randint(60, 120)}
     })
     topic_object.publish(data=queue_message)
     print('Message Published')
 
+def long_running_task(sleep_time):
+    # Simulate a long-running task
+    time.sleep(sleep_time)
+
 
 def subscribe(topic_name, subscription_name):
     def process(message):
-        print(f'Message Received: {message}')
-        # Spawn and thread process it -> 1 hr no issues
-        # return
+        print(f'Message Received: {message.data}')
+        long_running_thread = threading.Thread(target=long_running_task, args=(message.data['a'],))
+        long_running_thread.start()
+        long_running_thread.join()
+        print(f' > Message Completed: {message.data}')
 
     topic_object = core.get_topic(topic_name=topic_name)
     try:
@@ -44,6 +45,8 @@ def process(message):
 
 
 # subscribe(topic, subscription)
+# for x in range(10):
+#     publish_messages(topic_name=topic)
 
 # azure_client = core.get_storage_client()
 

src/python_ms_core/__init__.py

@@ -1,3 +1,4 @@
+import os
 import logging
 from .core.logger.logger import Logger
 from .core.logger.local_logger import LocalLogger
@@ -43,7 +44,7 @@ def get_logger(self):
         else:
             logging.error(f'Failed to initialize core.get_logger for provider: {logger_config.provider}')
 
-    def get_topic(self, topic_name: str, max_concurrent_messages=1):
+    def get_topic(self, topic_name: str, max_concurrent_messages=os.cpu_count()):
         topic_config = self.config.topic()
         if topic_config.provider.upper() == LOCAL_ENV:
             return LocalTopic(config=topic_config, topic_name=topic_name)

src/python_ms_core/core/topic/topic.py

@@ -16,39 +16,28 @@ class Callback:
     def __init__(self, fn=None, max_concurrent_messages=1):
         self._function_to_call = fn
         self._max_concurrent_messages = max_concurrent_messages
-        self._stop_event = threading.Event()
-        self._renewal_interval = 10  # seconds
+        self._renewal_interval = 30  # seconds
+        self.message_processing = 0
 
     def _renew_message_lock(self, message, receiver):
-        while not self._stop_event.is_set():
+        while True:
             try:
                 time.sleep(self._renewal_interval)
                 if not message._lock_expired:
                     receiver.renew_message_lock(message)
             except Exception as e:
                 break
 
-    # old method to fetch messages. Not used anymore
-    def messages(self, provider, topic, subscription):
-        with provider.client:
-            topic_receiver = provider.client.get_subscription_receiver(topic, subscription_name=subscription)
-            logger.info(f'Started receiver for {subscription}')
-            with topic_receiver:
-                for message in topic_receiver:
-                    try:
-                        queue_message = QueueMessage.data_from(str(message))
-                        self._function_to_call(queue_message)
-                    except Exception as e:
-                        print(f'Error: {e}, Invalid message received: {message}')
-                    finally:
-                        topic_receiver.complete_message(message)
-            logger.info('Completed topic receiver')
-    
     # Sends data to the callback function
     def process_message(self, message, receiver):
         queue_message = QueueMessage.data_from(str(message))
+        if not message._lock_expired:
+            lock_renewal_thread = threading.Thread(
+                target=self._renew_message_lock,
+                args=(message, receiver)
+            )
+            lock_renewal_thread.start()
         self._function_to_call(queue_message)
-        receiver.complete_message(message)
 
     # Starts listening to the messages
     def start_listening(self, provider, topic, subscription):
@@ -61,35 +50,39 @@ def start_listening(self, provider, topic, subscription):
             logger.info('Receiver started')
             with topic_receiver:
                 while True:
-                    try:
-                        messages = topic_receiver.receive_messages(
-                            max_message_count=self._max_concurrent_messages,
-                            max_wait_time=5
-                        )
-                        if not messages:
-                            continue
-
-                        for message in messages:
-                            stop_event = threading.Event()
-                            lock_renewal_thread = threading.Thread(
-                                target=self._renew_message_lock,
-                                args=(message, topic_receiver)
+                    available_slots = self._max_concurrent_messages - self.message_processing
+                    if available_slots > 0:
+                        try:
+                            messages = topic_receiver.receive_messages(
+                                max_message_count=available_slots,
+                                max_wait_time=5
                             )
-                            lock_renewal_thread.start()
-                            try:
-                                self.process_message(message, topic_receiver)
-                            except Exception as e:
-                                logger.error(f'Error processing message: {message}. Error: {e}')
-                                topic_receiver.abandon_message(message)
-                                logger.info(f'Message {message.message_id} abandoned')
-                            finally:
-                                stop_event.set()  # Signal the lock renewal thread to stop
-                                lock_renewal_thread.join()
-                    except Exception as et:
-                        logger.error(f'Error in service bus connection: {et}')
-                        # Change mode from PEEK_LOCK to RECEIVE_AND_DELETE
+                            if not messages:
+                                continue
+                            for message in messages:
+                                self.message_processing += 1
+                                threading.Thread(
+                                    target=self._process_message_in_thread,
+                                    args=(message, topic_receiver,)).start()
+                        except Exception as et:
+                            logger.error(f'Error in service bus connection: {et}')
+                    else:
+                        time.sleep(10)  # Short sleep to prevent tight loop if no slots available
+
                 logger.info('Receiver stopped')
 
+    def _process_message_in_thread(self, message, topic_receiver):
+        try:
+            self.process_message(message=message, receiver=topic_receiver)
+        except Exception as e:
+            logger.error(f'Error: {e}, Invalid message received: {message}')
+        finally:
+            try:
+                topic_receiver.complete_message(message)  # Mark the message as complete
+            except Exception as err:
+                logger.error(f'Error completing the message: {err}')
+            self.message_processing -= 1
+
 
 class Topic(TopicAbstract):
     def __init__(self, config=None, topic_name=None, max_concurrent_messages=1):

tests/unit_tests/test_topic/test_topic.py

@@ -61,5 +61,78 @@ def test_publish(self):
         mock_sender.send_messages.assert_called_once()
 
 
+class TestCallback(unittest.TestCase):
+
+    def setUp(self):
+        self.mock_function = MagicMock()
+        self.callback = Callback(fn=self.mock_function, max_concurrent_messages=2)
+        self.mock_message = MagicMock()
+        self.mock_receiver = MagicMock()
+        self.mock_provider = MagicMock()
+        self.mock_topic_receiver = MagicMock()
+        self.mock_provider.client.get_subscription_receiver.return_value = self.mock_topic_receiver
+        self.mock_topic_receiver.receive_messages.return_value = [self.mock_message]
+
+    def test_renew_message_lock(self):
+        self.mock_message._lock_expired = False
+        self.callback._renewal_interval = 0  # For fast testing
+        with patch.object(self.mock_receiver, 'renew_message_lock') as mock_renew_message_lock:
+            with patch('time.sleep', side_effect=[None, Exception()]):
+                self.callback._renew_message_lock(self.mock_message, self.mock_receiver)
+                mock_renew_message_lock.assert_called_with(self.mock_message)
+
+    @patch('threading.Thread')
+    def test_process_message_lock_expired(self, mock_thread_class):
+        self.mock_message._lock_expired = True
+        with patch.object(QueueMessage, 'data_from', return_value=QueueMessage()):
+            self.callback.process_message(self.mock_message, self.mock_receiver)
+            self.mock_function.assert_called_once()
+            mock_thread_class.assert_not_called()
+
+    @patch('threading.Thread')
+    def test_process_message(self, mock_thread_class):
+        self.mock_message._lock_expired = False
+        with patch.object(QueueMessage, 'data_from', return_value=QueueMessage()):
+            self.callback.process_message(self.mock_message, self.mock_receiver)
+            self.mock_function.assert_called_once()
+            mock_thread_class.assert_called_once()
+
+    def test_process_message_in_thread_success(self):
+        with patch.object(self.callback, 'process_message') as mock_process_message:
+            self.callback.message_processing = 1
+            self.callback._process_message_in_thread(self.mock_message, self.mock_receiver)
+            mock_process_message.assert_called_once_with(message=self.mock_message, receiver=self.mock_receiver)
+            self.mock_receiver.complete_message.assert_called_once_with(self.mock_message)
+            self.assertEqual(self.callback.message_processing, 0)
+
+    def test_process_message_in_thread_process_message_error(self):
+        with patch.object(self.callback, 'process_message', side_effect=Exception("Mocked Exception")):
+            self.callback.message_processing = 1
+            self.callback._process_message_in_thread(self.mock_message, self.mock_receiver)
+            self.mock_receiver.complete_message.assert_called_once_with(self.mock_message)
+            self.assertEqual(self.callback.message_processing, 0)
+
+    def test_process_message_in_thread_complete_message_error(self):
+        with patch.object(self.callback, 'process_message'):
+            self.mock_receiver.complete_message.side_effect = Exception("Mocked Exception")
+            self.callback.message_processing = 1
+            self.callback._process_message_in_thread(self.mock_message, self.mock_receiver)
+            self.mock_receiver.complete_message.assert_called_once_with(self.mock_message)
+            self.assertEqual(self.callback.message_processing, 0)
+
+    @patch('src.python_ms_core.core.topic.topic.logger')
+    def test_process_message_in_thread_both_errors(self, mock_logger):
+        with patch.object(self.callback, 'process_message', side_effect=Exception("Mocked Process Exception")):
+            self.mock_receiver.complete_message.side_effect = Exception("Mocked Complete Exception")
+            self.callback.message_processing = 1
+            self.callback._process_message_in_thread(self.mock_message, self.mock_receiver)
+            self.mock_receiver.complete_message.assert_called_once_with(self.mock_message)
+            self.assertEqual(self.callback.message_processing, 0)
+            self.assertEqual(mock_logger.error.call_count, 2)
+            mock_logger.error.assert_any_call(
+                f'Error: Mocked Process Exception, Invalid message received: {self.mock_message}')
+            mock_logger.error.assert_any_call(f'Error completing the message: Mocked Complete Exception')
+
+
 if __name__ == '__main__':
     unittest.main()