Fixed multi-threading and Updated new version

Author: sujata-m

CHANGELOG.md

@@ -1,8 +1,11 @@
 # Change log
 
+### 0.0.4
+- Fixed threading
+
 ### 0.0.3
 - Removed blocking call from topic subscription
-- Added TODOs abf FIXMEs
+- Added TODOs and FIXMEs
 
 ### 0.0.2
 - Added filtered exception handling

freeze_version.py

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

src/python_ms_core/core/queue/queue.py

@@ -28,7 +28,7 @@ def send(self, data=None):
         if data:
             message = QueueMessage.to_dict(data)
             with self.provider.client:
-                sender = self.provider.client.get_queue_sender(queue_name=self.azure.queue_name)
+                sender = self.provider.client.get_queue_sender(queue_name=self.provider.queue_name)
                 with sender:
                     sender.send_messages(self.provider.sender(json.dumps(message)))
         self.queue = list()

src/python_ms_core/core/queue/topic.py

@@ -11,16 +11,18 @@ class Callback:
     def __init__(self, fn=None):
         self._function_to_call = fn
 
-    def messages(self, client):
-        for message in client:
-            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:
-                client.complete_message(message)
-
+    def messages(self, provider, topic, subscription):
+        with provider.client:
+            topic_receiver = provider.client.get_subscription_receiver(topic, subscription_name=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)
 
 class Topic:
     def __init__(self, config=None, topic_name=None):
@@ -37,14 +39,9 @@ def __init__(self, config=None, topic_name=None):
     def subscribe(self, subscription=None, callback=None):
         if subscription is not None:
             cb = Callback(callback)
-            with self.provider.client:
-                topic_receiver = self.provider.client.get_subscription_receiver(self.provider.topic,
-                                                                                subscription_name=subscription)
-                with topic_receiver:
-                    if callback and callable(callback):
-                        thread = threading.Thread(target=cb.messages, args=(topic_receiver,))
-                        thread.start()
-                        time.sleep(5)
+            thread = threading.Thread(target=cb.messages, args=(self.provider,self.topic, subscription))
+            thread.start()
+            time.sleep(5)
         else:
             logging.error(
                 f'Unimplemented initialization for core {self.provider.provider}, Subscription name is required!')