From ee9a0796932f0cc1d9c758a8530037825415fb43 Mon Sep 17 00:00:00 2001
From: Jorrit Schaap <schaap@astron.nl>
Date: Thu, 6 Jun 2019 13:17:12 +0000
Subject: [PATCH] SW-699: wrap event_listener in a janitor to auto cleanup the
designed listening queue upon exit. this tool is only meant for live
listening, not for historic logs
---
.../LTAIngestClient/lib/ingestbuslistener.py | 12 ++++++++----
1 file changed, 8 insertions(+), 4 deletions(-)
diff --git a/LTA/LTAIngest/LTAIngestClient/lib/ingestbuslistener.py b/LTA/LTAIngest/LTAIngestClient/lib/ingestbuslistener.py
index 6d6a4f129cb..b17274eb89a 100644
--- a/LTA/LTAIngest/LTAIngestClient/lib/ingestbuslistener.py
+++ b/LTA/LTAIngest/LTAIngestClient/lib/ingestbuslistener.py
@@ -20,7 +20,7 @@
from lofar.common.util import humanreadablesize
-from lofar.messaging.messagebus import BusListener, AbstractMessageHandler
+from lofar.messaging.messagebus import BusListener, AbstractMessageHandler, BusListenerJanitor
from lofar.lta.ingest.common.config import INGEST_NOTIFICATION_PREFIX
from lofar.messaging.messagebus import DEFAULT_BROKER, DEFAULT_BUSNAME
from lofar.messaging.messages import EventMessage
@@ -191,15 +191,19 @@ def main():
# Check the invocation arguments
parser = OptionParser('%prog [options]',
description='run the ingest job monitor')
- parser.add_option('-q', '--broker', dest='broker', type='string', default=DEFAULT_BROKER, help='Address of the message broker, default: %default')
- parser.add_option('--exchange', dest='exchange', type='string', default=DEFAULT_BUSNAME, help='Name of the exchange on which the ingest notifications are published, default: %default')
+ parser.add_option('-b', '--broker', dest='broker', type='string', default=DEFAULT_BROKER, help='Address of the message broker, default: %default')
+ parser.add_option('-e', '--exchange', dest='exchange', type='string', default=DEFAULT_BUSNAME, help='Name of the exchange on which the ingest notifications are published, default: %default')
(options, args) = parser.parse_args()
logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s',
level=logging.INFO,
stream=sys.stdout)
- with IngestEventMesssageBusListener(exchange=options.exchange, broker=options.broker):
+ event_listener = IngestEventMesssageBusListener(exchange=options.exchange, broker=options.broker)
+
+ # wrap event_listener in a janitor to auto cleanup the designed listening queue upon exit.
+ # this tool is only meant for live listening, not for historic logs.
+ with BusListenerJanitor(event_listener):
waitForInterrupt()
if __name__ == '__main__':
--
GitLab