Merge pull request #15 from TaskarCenterAtUW/bugfix-connection-close

Bugfix connection close

Author: sujata-m

CHANGELOG.md

@@ -1,5 +1,19 @@
 # Change log
 
+### 0.0.15
+This fixes the inconsistent listening behavior for the subscription of a topic from core.
+- Due to the usage of `for message in receiver` logic in the topic, it is unknown when the loop will end and the core will stop listening to the topic.
+- This is tried with various combinations and figured out that at probably 4 hours from launch, this happens. The root cause for this is the socket / connection timeout of the underlying amqp client which throws an exception when trying to iterate next message
+
+Fix made:
+- The above for loop is kept in an infinite while loop which triggers the creation of the receiver and subsequent listening to the messages. 
+- This was tested overnight with messages varying in the time differences (1h to 3h)
+- Specific method to look for the fix `start_listening`
+
+Reference task:
+[302](https://dev.azure.com/TDEI-UW/TDEI/_workitems/edit/302)
+
+
 ### 0.0.14
 - Fixed https://dev.azure.com/TDEI-UW/TDEI/_workitems/edit/302
 

freeze_version.py

@@ -11,7 +11,7 @@
 
 build_date = date.today().strftime('%Y-%m-%d')
 
-version = '0.0.14'
+version = '0.0.15'
 
 with open(version_file_path, 'w+') as version_file:
     version_file.write("version = '{}'\n".format(version))

src/example.py

@@ -11,11 +11,11 @@
 from python_ms_core.core.queue.models.queue_message import QueueMessage
 from python_ms_core.core.auth.models.permission_request import PermissionRequest
 
-core = Core(config='Local')
+core = Core()
 print('Hello')
 
-topic = 'gtfspathways'
-subscription = 'upload-validation-processor-test'
+topic = 'gtfs-pathways-upload'
+subscription = 'log'
 some_other_sub = 'usdufs'
 
 
@@ -32,6 +32,8 @@ def publish_messages(topic_name):
 def subscribe(topic_name, subscription_name):
     def process(message):
         print(f'Message Received: {message}')
+        # Spawn and thread process it -> 1 hr no issues
+        # return
 
     topic_object = core.get_topic(topic_name=topic_name)
     try:
@@ -42,52 +44,52 @@ def process(message):
 
 subscribe(topic, subscription)
 
-azure_client = core.get_storage_client()
-
-container = azure_client.get_container(container_name='gtfspathways')
-
-list_of_files = container.list_files()
-if len(list_of_files) > 0:
-    for single in list_of_files:
-        print(single.path)
-    firstFile = list_of_files[0]
-    # print(firstFile.name+'<><>')
-    file_content = firstFile.get_body_text()
-
-# Creating a text stream
-txt = 'foo\nbar\nbaz'
-file_like_io = StringIO(txt)
-basename = 'sample-file'
-suffix = datetime.datetime.now().strftime("%y%m%d_%H%M%S")
-filename = '_'.join([basename, suffix])
-try:
-    test_file = container.create_file(f'{filename}.txt')
-except Exception as e:
-    print(e)
-print('Start uploading...')
-test_file.upload(file_like_io.read())
-print(test_file.get_remote_url())
-print('Uploaded Successfully')
-
-logger = core.get_logger()
-logger.record_metric(name='test', value='test')
-
-publish_messages(topic)
-time.sleep(2)
-
-permission_params = PermissionRequest(
-    user_id='7961d767-a352-464f-95b6-cd1c5189a93c',
-    org_id='5e339544-3b12-40a5-8acd-78c66d1fa981',
-    should_satisfy_all=False,
-    permissions=['poc']
-)
-
-try:
-    auth = core.get_authorizer()
-    resp = auth.has_permission(request_params=permission_params)
-    print(resp)
-except Exception as e:
-    print(f'Request failed with Code: {e.status_code}, Message: {e.message}')
-    print()
-
-os._exit(os.EX_OK)
+# azure_client = core.get_storage_client()
+
+# container = azure_client.get_container(container_name='gtfspathways')
+
+# list_of_files = container.list_files()
+# if len(list_of_files) > 0:
+#     for single in list_of_files:
+#         print(single.path)
+#     firstFile = list_of_files[0]
+#     # print(firstFile.name+'<><>')
+#     file_content = firstFile.get_body_text()
+
+# # Creating a text stream
+# txt = 'foo\nbar\nbaz'
+# file_like_io = StringIO(txt)
+# basename = 'sample-file'
+# suffix = datetime.datetime.now().strftime("%y%m%d_%H%M%S")
+# filename = '_'.join([basename, suffix])
+# try:
+#     test_file = container.create_file(f'{filename}.txt')
+# except Exception as e:
+#     print(e)
+# print('Start uploading...')
+# test_file.upload(file_like_io.read())
+# print(test_file.get_remote_url())
+# print('Uploaded Successfully')
+
+# logger = core.get_logger()
+# logger.record_metric(name='test', value='test')
+
+# publish_messages(topic)
+# time.sleep(2)
+
+# permission_params = PermissionRequest(
+#     user_id='7961d767-a352-464f-95b6-cd1c5189a93c',
+#     org_id='5e339544-3b12-40a5-8acd-78c66d1fa981',
+#     should_satisfy_all=False,
+#     permissions=['poc']
+# )
+
+# try:
+#     auth = core.get_authorizer()
+#     resp = auth.has_permission(request_params=permission_params)
+#     print(resp)
+# except Exception as e:
+#     print(f'Request failed with Code: {e.status_code}, Message: {e.message}')
+#     print()
+
+# os._exit(os.EX_OK)

src/python_ms_core/core/topic/config/topic_config.py

@@ -1,11 +1,24 @@
 from azure.servicebus import ServiceBusClient, ServiceBusMessage
-
+import logging
 
 class Config:
     def __init__(self, config=None, topic_name=None):
         self.topic = topic_name
         self.provider = config.provider
         self.connection_string = config.connection_string
         if self.provider.lower() == 'azure':
+            # The logging levels below may need to be changed based on the logging that you want to suppress.
+            uamqp_logger = logging.getLogger('uamqp')
+            uamqp_logger.setLevel(logging.ERROR)
+
+            # or even further fine-grained control, suppressing the warnings in uamqp.connection module
+            uamqp_connection_logger = logging.getLogger('uamqp.connection')
+            uamqp_connection_logger.setLevel(logging.ERROR)
             self.client = ServiceBusClient.from_connection_string(conn_str=self.connection_string, retry_total=10, retry_backoff_factor=1, retry_backoff_max=30)
             self.sender = ServiceBusMessage
+            # receiver = self.client.get_subscription_receiver('','')
+            # receiver.receive_messages(1,50)
+            # with receiver:
+            #     for message in receiver:
+            #         print('')
+            # self.client.get_subscription_receiver()

src/python_ms_core/core/topic/topic.py

@@ -7,14 +7,20 @@
 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')
+logger = logging.getLogger('Topic')
+logger.setLevel(logging.INFO)
+
 
 class Callback:
     def __init__(self, fn=None):
         self._function_to_call = fn
 
+    # 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:
@@ -23,7 +29,33 @@ def messages(self, provider, topic, subscription):
                     except Exception as e:
                         print(f'Error: {e}, Invalid message received: {message}')
                     finally:
+                        print(f'Completing message')
                         topic_receiver.complete_message(message)
+                logger.info(f'Completed gathering messages')
+            logger.info('Completed topic receiver')
+    
+    # Sends data to the callback function
+    def process_message(self, message:str):
+        queue_message = QueueMessage.data_from(message)
+        self._function_to_call(queue_message)
+    
+    # Starts listening to the messages
+    def start_listening(self, provider, topic, subscription):
+        with provider.client: # service bus client
+            while True:
+                logger.info('Initiatig receiver')
+                topic_receiver = provider.client.get_subscription_receiver(topic, subscription_name=subscription) # servicebusclientsubscriptionreceiver
+                with topic_receiver:
+                    for message in topic_receiver:
+                        try:
+                            self.process_message(message=str(message)) # sync call. [By default 1minute ] -> lock renewal for 300 seconds
+                            topic_receiver.complete_message(message)
+                        except Exception as e:
+                            print(f'Error : {e}, Invalid message received : {message}')
+                        finally:
+                            topic_receiver.complete_message(message)
+                        # Change mode from PEEK_LOCK to RECEIVE_AND_DELETE
+                logger.info('Topic receiver invalidated')
 
 
 class Topic(TopicAbstract):
@@ -35,7 +67,7 @@ def __init__(self, config=None, topic_name=None):
     def subscribe(self, subscription=None, callback=None):
         if subscription is not None:
             cb = Callback(callback)
-            thread = threading.Thread(target=cb.messages, args=(self.provider, self.topic, subscription))
+            thread = threading.Thread(target=cb.start_listening, args=(self.provider, self.topic, subscription))
             thread.start()
             time.sleep(5)
         else: