From 2ff8ed76d02529a8b8ed3bcca31a9850b3dc1a47 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dirk=20M=C3=BCller?= Date: Mon, 10 Jun 2024 17:04:25 +0200 Subject: [PATCH] Reconnect to the AMQP bus when the connection breaks down --- opensuse-monitor.py | 55 ++++++++++++++++++++++++++++----------------- 1 file changed, 35 insertions(+), 20 deletions(-) diff --git a/opensuse-monitor.py b/opensuse-monitor.py index 3a1e569..8479d10 100755 --- a/opensuse-monitor.py +++ b/opensuse-monitor.py @@ -2,36 +2,51 @@ import json from pathlib import Path import pika +import random import sys +import time MY_TASKS_DIR = Path(__file__).parent / "tasks" -connection = pika.BlockingConnection(pika.URLParameters("amqps://opensuse:opensuse@rabbit.opensuse.org")) -channel = connection.channel() +def listen_events(): + connection = pika.BlockingConnection(pika.URLParameters("amqps://opensuse:opensuse@rabbit.opensuse.org")) + channel = connection.channel() -channel.exchange_declare(exchange='pubsub', exchange_type='topic', passive=True, durable=True) + channel.exchange_declare(exchange='pubsub', exchange_type='topic', passive=True, durable=False) -result = channel.queue_declare("", exclusive=True) -queue_name = result.method.queue + result = channel.queue_declare("", exclusive=True) + queue_name = result.method.queue -channel.queue_bind(exchange='pubsub', - queue=queue_name,routing_key='#') + channel.queue_bind(exchange='pubsub', + queue=queue_name,routing_key='opensuse.obs.package.commit') -print(' [*] Waiting for logs. To exit press CTRL+C') + print(' [*] Waiting for logs. To exit press CTRL+C') -def callback(ch, method, properties, body): - if method.routing_key not in ("opensuse.obs.package.commit",): - return - body = json.loads(body) - if 'project' in body and 'package' in body and body['project'] == 'openSUSE:Factory': - if '/' in body['package']: + def callback(ch, method, properties, body): + if method.routing_key not in ("opensuse.obs.package.commit",): return + body = json.loads(body) + if 'project' in body and 'package' in body and body['project'] == 'openSUSE:Factory': + if '/' in body['package']: + return - (MY_TASKS_DIR / body['package']).touch() - print(" [x] %r:%r" % (method.routing_key, body['package'])) + (MY_TASKS_DIR / body['package']).touch() + print(" [x] %r:%r" % (method.routing_key, body['package'])) -channel.basic_consume(queue_name, - callback, - auto_ack=True) + channel.basic_consume(queue_name, + callback, + auto_ack=True) -channel.start_consuming() + channel.start_consuming() + + +def main(): + while True: + try: + listen_events(); + except pika.exceptions.ConnectionClosed: + time.sleep(random.randint(10,100)) + + +if __name__ == '__main__': + main()