Trace Retrieval: More robust message handling

transaction-hub-in/trace-retrieval-microservice/app/db/MongoRepository.py

@@ -7,11 +7,13 @@ class MongoRepository:
     # TODO extract to docker env var
     _username = 'root'
     _password = 'root'
-    _collection : pymongo.collection.Collection = None
+
+    _collection: pymongo.collection.Collection = None
+    _mongo_client: pymongo.MongoClient = None
 
     def __init__(self, username=_username, password=_password):
-        myclient = pymongo.MongoClient(f"mongodb://{username}:{password}@{MONGO_DB_HOST}:{MONGO_DB_PORT}/")
-        database = myclient['traceRetrievalDB'] # trace retrieval
+        self._mongo_client = pymongo.MongoClient(f"mongodb://{username}:{password}@{MONGO_DB_HOST}:{MONGO_DB_PORT}/")
+        database = self._mongo_client['traceRetrievalDB']
         self._collection = database['traces']
 
     def insert_trace(self, content: dict):
@@ -19,3 +21,8 @@ class MongoRepository:
 
     def get_traces(self, selection: dict = {}):
         return self._collection.find(selection)
+
+    def close_connection(self):
+        self._mongo_client.close()
+        self._collection = None
+        self._mongo_client = None

transaction-hub-in/trace-retrieval-microservice/app/main.py

@@ -21,16 +21,17 @@ message_rec: MessageReceiver = None
 # init message handler
 message_handler = MessageHandler()
 def message_received_callback(channel, method, properties, body):
-    LOGGER.info(f"Received new message: {body}")
     message_handler.handle_generic(body)
 
 def pika_error_callback(error):
     LOGGER.warning(f"RabbitMQ stopped with error: {error}")
     # restart receiver
+    global message_rec
     message_rec.stop()
     init_message_receiver()
 
 def init_message_receiver():
+    global message_rec
     message_rec = MessageReceiver(exchange_name='inhub', exchange_type='direct', queue_name='trace-retrieval', auto_ack=True)
     message_rec.start(message_received_callback, pika_error_callback)
 

transaction-hub-in/trace-retrieval-microservice/app/messaging/MessageHandler.py

@@ -12,9 +12,16 @@ class MessageHandler:
 
     def handle_generic(self, body):
         LOGGER.info(f"Received message: {body}")
-        message = json.loads(body)
+
+        message = None
+        try:
+            message = json.loads(body)
+        except ValueError:
+            LOGGER.warning(f"Message is not in JSON format and is ignored")
+            return
+
         if not 'type' in message:
-            LOGGER.warning(f"Message has no type field -> ignored")
+            LOGGER.warning(f"Message has no type field and is ignored")
             return
         
         if message['type'] == 'blockchain-transaction':

transaction-hub-in/trace-retrieval-microservice/deployment/deployment.yml

@@ -20,7 +20,7 @@ kind: Deployment
 metadata:
   name: trace-retrieval
 spec:
-  replicas: 2
+  replicas: 1
   selector:
     matchLabels:
       app: trace-retrieval
@@ -34,6 +34,7 @@ spec:
         image: 172.16.1.20:5000/trace-retrieval-microservice
         ports:
         - containerPort: 5000
+        imagePullPolicy: Always
 ---
 apiVersion: v1
 kind: Service