TMSS-307: while working on TMSS commissioning with TMSS-573 it was discovered...

TMSS-307: while working on TMSS commissioning with TMSS-573 it was discovered that cobalt sends its feedback to a queue with a single consumer. Now with two consumers (the extra tmss_feedback_service) they were fighting over the messages. Hence, the queue has been replaced with a topic exchange which duplicates the message in two seperate queues for otdb and for tmss.

SAS/Feedback_Service/src/FeedbackService.conf

@@ -3,7 +3,7 @@
 #
 # Parameters for FeedbackService to connect to Qpid.
 #
-FeedbackQueuenames = [ "otdb.task.feedback.dataproducts" , "otdb.task.feedback.processing" ]
+FeedbackQueuenames = [ "otdb.task.feedback.dataproducts.for_otdb" , "otdb.task.feedback.processing.for_otdb" ]
 
 FeedbackService.OTDBdatabase = LOFAR_4
 FeedbackService.OTDBhostname = sasdb

SAS/QPIDInfrastructure/amqp-infra-setup.sh

@@ -43,8 +43,18 @@ fi
 
 # MessageBus
 qpid-config -b $CCU add queue ${PREFIX}mac.task.feedback.state --durable
-qpid-config -b $MCU add queue ${PREFIX}otdb.task.feedback.dataproducts --durable
-qpid-config -b $MCU add queue ${PREFIX}otdb.task.feedback.processing --durable
+qpid-config -b $MCU del queue topic ${PREFIX}otdb.task.feedback.dataproducts --durable
+qpid-config -b $MCU add exchange topic ${PREFIX}otdb.task.feedback.dataproducts --durable
+qpid-config -b $MCU add queue ${PREFIX}otdb.task.feedback.dataproducts.for_otdb --durable
+qpid-config -b $MCU bind ${PREFIX}otdb.task.feedback.dataproducts ${PREFIX}otdb.task.feedback.dataproducts.for_otdb "#" --durable
+qpid-config -b $MCU add queue ${PREFIX}otdb.task.feedback.dataproducts.for_tmss --durable
+qpid-config -b $MCU bind ${PREFIX}otdb.task.feedback.dataproducts ${PREFIX}otdb.task.feedback.dataproducts.for_tmss "#" --durable
+qpid-config -b $MCU del queue topic ${PREFIX}otdb.task.feedback.processing --durable
+qpid-config -b $MCU add exchange topic ${PREFIX}otdb.task.feedback.processing --durable
+qpid-config -b $MCU add queue ${PREFIX}otdb.task.feedback.processing.for_otdb --durable
+qpid-config -b $MCU bind ${PREFIX}otdb.task.feedback.processing ${PREFIX}otdb.task.feedback.processing.for_otdb "#" --durable
+qpid-config -b $MCU add queue ${PREFIX}otdb.task.feedback.processing.for_tmss --durable
+qpid-config -b $MCU bind ${PREFIX}otdb.task.feedback.processing ${PREFIX}otdb.task.feedback.processing.for_tmss "#" --durable
 qpid-config -b $MCU add queue ${PREFIX}lofar.task.specification.system --durable
 qpid-config -b $CCU add queue ${PREFIX}lofar.task.specification.system --durable
 qpid-config -b $CCU add queue ${PREFIX}mom.task.specification.system --durable

SAS/TMSS/backend/services/feedback_handling/lib/feedback_handling.py

@@ -42,7 +42,7 @@ class TMSSFeedbackListener:
     # Note we can't use fancy bus listeners since we have to use the really old messagebus package for feedback
     # todo: get rid of old-style messaging or improve this service stub
     threads = []
-    exchanges = ["otdb.task.feedback.processing", "otdb.task.feedback.dataproducts"]
+    exchanges = ["otdb.task.feedback.processing.for_tmss", "otdb.task.feedback.dataproducts.for_tmss"]
 
     def __init__(self, rest_client_creds_id: str="TMSSClient", qpid_broker: str=broker_feedback) -> None:
         super().__init__()
@@ -75,6 +75,7 @@ class TMSSFeedbackListener:
                     if msg is not None:
                         content = msg.content()
                         logger.info("received feedback from bus='%s' %s", exchange_name, content)
+                        fbus.ack(msg)
 
                         # note: cobalt/rtcp creates feedback and assumes that the observationID has its origin in OTDB.
                         # hence, it stores this id in the 'sasid' property of the message.