Introduced renew message lock

This PR contains following changes -

- 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

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,11 @@
 # 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
+
 # 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
 
@@ -293,7 +296,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

@@ -11,9 +11,10 @@
 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')
 
+
+core = Core()
+print(f'Core version: {Core.__version__}')
 topic = 'gtfs-pathways-upload'
 subscription = 'log'
 some_other_sub = 'usdufs'
@@ -42,7 +43,7 @@ def process(message):
         print(e)
 
 
-subscribe(topic, subscription)
+# subscribe(topic, subscription)
 
 # azure_client = core.get_storage_client()
 

src/python_ms_core/__init__.py

@@ -8,6 +8,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 +31,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 +43,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=1):
         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 +118,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

@@ -7,14 +7,26 @@
 from ..resource_errors import ExceptionHandler
 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._stop_event = threading.Event()
+        self._renewal_interval = 10  # seconds
+
+    def _renew_message_lock(self, message, receiver):
+        while not self._stop_event.is_set():
+            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):
@@ -33,41 +45,62 @@ def messages(self, provider, topic, subscription):
             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))
         self._function_to_call(queue_message)
-    
+        receiver.complete_message(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:
+                        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)
+                            )
+                            lock_renewal_thread.start()
                             try:
-                                self.process_message(message=str(message)) # sync call. [By default 1minute ] -> lock renewal for 300 seconds
+                                self.process_message(message, topic_receiver)
                             except Exception as e:
-                                print(f'Error : {e}, Invalid message received : {message}')
+                                logger.error(f'Error processing message: {message}. Error: {e}')
+                                topic_receiver.abandon_message(message)
+                                logger.info(f'Message {message.message_id} abandoned')
                             finally:
-                                topic_receiver.complete_message(message)
+                                stop_event.set()  # Signal the lock renewal thread to stop
+                                lock_renewal_thread.join()
                     except Exception as et:
-                        print(f'Error in service bus connection : {et}')
+                        logger.error(f'Error in service bus connection: {et}')
                         # Change mode from PEEK_LOCK to RECEIVE_AND_DELETE
-                logger.info('Topic receiver invalidated')
+                logger.info('Receiver stopped')
 
 
 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)
+            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)

src/python_ms_core/version.py

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