Merge pull request #35 from TaskarCenterAtUW/feature-1102

Implemented automatic message lock renewal and enhanced concurrent message processing

Author: sujata-m

.gitignore

@@ -93,7 +93,6 @@ ipython_config.py
 
 # PEP 582; used by e.g. github.com/David-OConnor/pyflow
 __pypackages__/
-version.py
 .build/
 .env
 service_details.json

CHANGELOG.md

@@ -1,5 +1,16 @@
 # Change log
 
+# 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
 - Additional logging done if there is no network and  service bus crashes

README.md

@@ -25,6 +25,9 @@ Eg.
 ```python
 from python_ms_core import Core
 core = Core() or Core(config='Local')
+
+# To check the package version
+Core.__version__ # 0.0.21
 ```
 The method analyzes the `.env` variables and does a health check on what components are available
 
@@ -109,7 +112,8 @@ 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  
 
 ```
 
@@ -293,7 +297,7 @@ The project is configured with `python` to figure out the coverage of the unit t
 - The terminal will show the output of coverage like this
 ```shell
 
->  coverage run --source=src/python_ms_core -m unittest discover -v tests/unit_tests
+>  python -m coverage run --source=src/python_ms_core -m unittest discover -v tests/unit_tests
 test_has_permission (test_auth.abstract.test_authorizer_abstraction.TestAuthorizerAbstract) ... ok
 test_get_search_params (test_auth.models.test_permission_request.TestPermissionRequest) ... ok
 test_has_permission_with_invalid_permissions (test_auth.provider.test_hosted_authorizer.TestHostedAuthorizer) ... ok

freeze_version.py

@@ -2,6 +2,7 @@
 import os
 import git
 from datetime import date
+from src.python_ms_core import Core
 
 project_path = os.path.dirname(os.path.abspath(__file__))
 version_file_path = '{}/version.py'.format(project_path)
@@ -11,7 +12,7 @@
 
 build_date = date.today().strftime('%Y-%m-%d')
 
-version = '0.0.20'
+version = Core.__version__
 
 with open(version_file_path, 'w+') as version_file:
     version_file.write("version = '{}'\n".format(version))

src/example.py

@@ -1,39 +1,45 @@
 # 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('Hello')
-
-topic = 'gtfs-pathways-upload'
-subscription = 'log'
-some_other_sub = 'usdufs'
+print(f'Core version: {Core.__version__}')
+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:
@@ -43,6 +49,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
@@ -8,6 +9,7 @@
 from .core.auth.provider.hosted.hosted_authorizer import HostedAuthorizer
 from .core.auth.provider.simulated.simulated_authorizer import SimulatedAuthorizer
 from .core.config.config import CoreConfig, LocalConfig, AuthConfig, UnknownConfig
+from .version import __version__
 
 LOCAL_ENV = 'LOCAL'
 AZURE_ENV = 'AZURE'
@@ -30,6 +32,9 @@ def __init__(self, config=None):
             self.config = CoreConfig()
             self.__check_health()
 
+    def __version__(self):
+        return
+
     def get_logger(self):
         logger_config = self.config.logger()
         if logger_config.provider.upper() == LOCAL_ENV:
@@ -39,12 +44,12 @@ 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):
+    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)
         elif topic_config.provider.upper() == AZURE_ENV:
-            return Topic(config=topic_config, topic_name=topic_name)
+            return Topic(config=topic_config, topic_name=topic_name, max_concurrent_messages=max_concurrent_messages)
         else:
             logging.error(f'Failed to initialize core.get_topic for provider: {topic_config.provider}')
 
@@ -114,3 +119,6 @@ def __check_health(self):
             print('\x1b[32m Logger configured \x1b[0m')
         print('\x1b[32m ------------------------- \x1b[0m')
         return True
+
+
+Core.__version__ = __version__

src/python_ms_core/core/topic/topic.py

@@ -1,76 +1,96 @@
 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')
+logging.basicConfig(format='%(asctime)s %(levelname)-8s %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
 logger = logging.getLogger('Topic')
 logger.setLevel(logging.INFO)
 
 
 class Callback:
-    def __init__(self, fn=None):
+    def __init__(self, fn=None, max_concurrent_messages=1):
         self._function_to_call = fn
+        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:
+            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:str):
-        queue_message = QueueMessage.data_from(message)
+    def process_message(self, message, receiver):
+        queue_message = QueueMessage.data_from(str(message))
+        secondary_executor = ThreadPoolExecutor(max_workers=1)
+        if not message._lock_expired:
+            secondary_executor.submit(self._renew_message_lock, message, receiver)
         self._function_to_call(queue_message)
-    
+
     # Starts listening to the messages
     def start_listening(self, provider, topic, subscription):
-        with provider.client: # service bus client
+        with provider.client:  # service bus client
             logger.info('Initiating receiver')
-            topic_receiver = provider.client.get_subscription_receiver(topic, subscription_name=subscription) # servicebusclientsubscriptionreceiver
-            logger.info('Done')
+            topic_receiver = provider.client.get_subscription_receiver(
+                topic_name=topic,
+                subscription_name=subscription
+            )
+            logger.info('Receiver started')
             with topic_receiver:
                 while True:
-                    try:
-                        for message in topic_receiver:
-                            try:
-                                self.process_message(message=str(message)) # sync call. [By default 1minute ] -> lock renewal for 300 seconds
-                            except Exception as e:
-                                print(f'Error : {e}, Invalid message received : {message}')
-                            finally:
-                                topic_receiver.complete_message(message)
-                    except Exception as et:
-                        print(f'Error in service bus connection : {et}')
-                        # Change mode from PEEK_LOCK to RECEIVE_AND_DELETE
-                logger.info('Topic receiver invalidated')
+                    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
+                            )
+                            if not messages:
+                                continue
+                            for message in messages:
+                                self.message_processing += 1
+                                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:
+                        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):
+    def __init__(self, config=None, topic_name=None, max_concurrent_messages=1):
         self.topic = topic_name
         self.provider = Config(config=config, topic_name=topic_name)
+        self.max_concurrent_messages = max_concurrent_messages
 
     @ExceptionHandler.decorated
     def subscribe(self, subscription=None, callback=None):
         if subscription is not None:
-            cb = Callback(callback)
-            thread = threading.Thread(target=cb.start_listening, args=(self.provider, self.topic, subscription))
-            thread.start()
-            time.sleep(5)
+            cb = Callback(callback, max_concurrent_messages=self.max_concurrent_messages)
+            cb.start_listening(self.provider, self.topic, subscription)
         else:
             logging.error(
                 f'Unimplemented initialize for core {self.provider.provider}, Subscription name is required!')

src/python_ms_core/version.py

@@ -0,0 +1 @@
+__version__ = '0.0.21'