minor changes

minor changes

Author: susrisha

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,19 @@
 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
 
     def messages(self, provider, topic, subscription):
         with provider.client:
-            topic_receiver = provider.client.get_subscription_receiver(topic, subscription_name=subscription)
+            topic_receiver = provider.client.get_subscription_receiver(topic, subscription_name=subscription,max_wait_time=200)
+            logger.info(f'Started receiver for {subscription}')
             with topic_receiver:
                 for message in topic_receiver:
                     try:
@@ -23,7 +28,26 @@ 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')
+    
+    def process_message(self, message:str):
+        queue_message = QueueMessage.data_from(message)
+        self._function_to_call(queue_message)
+
+    def start_listening(self, provider, topic, subscription):
+        with provider.client: # service bus client
+            while True:
+                logger.info('Going into while')
+                topic_receiver = provider.client.get_subscription_receiver(topic, subscription_name=subscription) # servicebusclientsubscriptionreceiver
+                with topic_receiver:
+                    for message in topic_receiver:
+                        self.process_message(message=str(message)) # sync call. [By default 1minute ] -> lock renewal for 300 seconds
+                        topic_receiver.complete_message(message) # fails -> peeklock is timedout
+                        # Change mode from PEEK_LOCK to RECEIVE_AND_DELETE
+                logger.info('Completed topic receiver')
 
 
 class Topic(TopicAbstract):
@@ -35,7 +59,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: