diff options
Diffstat (limited to 'rmq')
-rw-r--r-- | rmq/__init__.py | 0 | ||||
-rw-r--r-- | rmq/consumer.py | 42 | ||||
-rw-r--r-- | rmq/publisher.py | 35 |
3 files changed, 0 insertions, 77 deletions
diff --git a/rmq/__init__.py b/rmq/__init__.py deleted file mode 100644 index e69de29..0000000 --- a/rmq/__init__.py +++ /dev/null diff --git a/rmq/consumer.py b/rmq/consumer.py deleted file mode 100644 index 2c41e26..0000000 --- a/rmq/consumer.py +++ /dev/null @@ -1,42 +0,0 @@ -import time -import traceback -import pika -from pika.exceptions import ConnectionClosed - - -def consume(username='guest', - password='guest', - host='localhost', - port=5672, - queue='', - callback=None): - - while True: - credentials = pika.PlainCredentials(username, password) - con_params = pika.ConnectionParameters(host, port, '/', credentials) - - try: - connection = pika.BlockingConnection(con_params) - - try: - channel = connection.channel() - channel.queue_declare(queue=queue, durable=False) - channel.basic_qos(prefetch_count=1) - channel.basic_consume(callback, queue=queue) - - print(f"""Connected to host: {host} port: {port} queue: {queue}""", flush=True) - - channel.start_consuming() - - except Exception: - exc = traceback.format_exc() - print(exc, flush=True) - - finally: - connection.close() - - except ConnectionClosed: - print(f"Connection lost, reconnecting to {host}", flush=True) - pass - - time.sleep(2) diff --git a/rmq/publisher.py b/rmq/publisher.py deleted file mode 100644 index 1a9a5cf..0000000 --- a/rmq/publisher.py +++ /dev/null @@ -1,35 +0,0 @@ -import pika - - -def publish(message, - username='guest', - password='guest', - host='localhost', - port=5672, - queue='', - routingkey='', - exchange='', - exchange_type=''): - - credentials = pika.PlainCredentials(username, password) - connection = pika.BlockingConnection(pika.ConnectionParameters(host, port, '/', credentials)) - properties = pika.BasicProperties(content_type='text/plain', delivery_mode=1) - - channel = connection.channel() - channel.queue_declare(queue=queue, durable=False) - channel.exchange_declare(exchange=exchange, exchange_type=exchange_type) - channel.queue_bind(exchange=exchange, queue=queue, routing_key=routingkey) - - result = channel.basic_publish( - exchange=exchange, - routing_key=routingkey, - body=message, - properties=properties - ) - - if result: - print(f"Connecting to host: {host} port: {port} exchange: {exchange} queue: {queue}", flush=True) - else: - print("not delivered") - - connection.close() |