I'm a beginner using Python, and I'm trying to implement a functionality of a monitor program already developed by an other programmer.
This program is working on a Windows machine (made by Inno Setup), which has to listen to the RabbitMQ server and consume messages. Those messages are supposed to start a background After Effect process (via PowerShell) and monitor it's workflow until it finishes. After that, the monitor consumes another message if it exists. This monitor is never supposed to be down and will have to work indefinitely.
So why I'm trying to develop this again? Because the current implementation is buggy at a very particular point : When RabbitMQ closes (reboot, shutdown, or other reason), consumers are disconnected, and no messages are consumed anymore.
My implementation below fixes this behavior (for simplicity I develop this entirely apart from the existing software, it was to reproduce the bug and fix it)
But I'm new in python, so I'm sure this code is bloated or ugly in some way, and I will gladly take all the advice you can give.
My Main.py
from src.RabbitMQ.RabbitMQ import RabbitMQ
class Main:
def __init__(self):
RabbitMQ()
main = Main()
RabbitMQ.py
from src.RabbitMQ.Configuration.Liveness import Liveness
class RabbitMQ:
def __init__(self):
self.start_liveness()
def start_liveness(self):
mainThread = Liveness()
mainThread.start()
ThreadedConnection.py
from src.RabbitMQ.ValueObjects.Connection import Connection
from src.RabbitMQ.Service.Consumer import Consumer
import threading
import time
class ThreadedConnection(threading.Thread):
def __init__(self, queue_name):
threading.Thread.__init__(self)
self.queueName = queue_name
self.should_reopen = False
self.connection = None
self.channel = None
self.connect()
# https://stackoverflow.com/questions/25489292/consuming-rabbitmq-queue-from-inside-python-threads
def connect(self):
try:
result = Connection().connect()
if result:
[self.connection, self.channel] = result
self.channel.queue_declare(queue=self.queueName, durable=True, arguments={"x-queue-type": "quorum"})
self.channel.basic_qos(prefetch_count=1)
self.channel.basic_consume(queue=self.queueName, on_message_callback=Consumer.consume, auto_ack=True)
else:
raise Exception('no connection available')
except:
print('hang on AMQP is closed')
def run(self):
try:
print('try consuming')
self.channel.start_consuming()
except:
print('something wrong happened... waiting for reopenning')
Connection.py
import pika
from src.RabbitMQ.ValueObjects.ConnectionParams import ConnectionParams
class Connection:
def __init__(self):
self.__get_params()
def __get_params(self):
self.connectionParams = ConnectionParams()
def connect(self):
connection = pika.PlainCredentials(self.connectionParams.username, self.connectionParams.password)
parameters = pika.ConnectionParameters(
self.connectionParams.host,
self.connectionParams.port,
self.connectionParams.vhost,
credentials=connection,
heartbeat=3600,
blocked_connection_timeout=3600
)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
return [connection, channel]
Consumer.py
class Consumer:
def __init__(self, channel):
"""
:param pika.adapters.blocking_connection.BlockingChannel channel: The channel
"""
self.channel = channel
def consume(self, method, properties, body):
"""
Cette méthode consomme nos messages
:param method:
:param properties:
:param body:
:return:
"""
print(body)
print('[*] Message received')
ConnectionParams.py
class ConnectionParams:
def __init__(self):
self.port = 5672
self.host = "127.0.0.1"
self.vhost = "/"
self.password = "test"
self.username = "test"
Liveness.py
import threading
import time
from src.RabbitMQ.Configuration.ThreadedConnection import ThreadedConnection
class Liveness(threading.Thread):
def __init__(self):
self.queues = ("preview", "common", "urgent")
threading.Thread.__init__(self)
self.channels = []
def reopen_channels(self):
self.channels = []
for queue in self.queues:
from src.RabbitMQ.RabbitMQ import RabbitMQ
print("The channel " + queue + " is closed reopen...")
self.connect(queue=queue)
def run_queues(self):
for queue in self.queues:
self.connect(queue)
def run(self):
self.run_queues()
self.liveness()
def connect(self, queue):
thread = ThreadedConnection(queue)
thread.start()
if thread.channel:
channel_tag = ''
for key in thread.channel._consumer_infos:
channel_tag = key
conf_dict = {
"channel": thread.channel,
"connection": thread.connection,
"ctag": channel_tag,
"thread": thread,
"queue": queue
}
self.channels.append(conf_dict)
def liveness(self):
while True:
should_reopen = False
if not self.channels:
self.run_queues()
for config in self.channels:
if config['thread'].isAlive():
alive = ". The thread is still alive too."
else:
alive = ". The thread is also closed."
if not config['channel'].is_closed:
channel_alive = " is alive"
else:
should_reopen = True
channel_alive = " is closed and should be reopened"
print("The channel " + config['ctag'] + channel_alive + alive)
if should_reopen:
print('re-openning...')
self.reopen_channels()
print('---')
time.sleep(20)
docker-compose.yml
version: '3'
services:
rabbitmq:
image: 'bitnami/rabbitmq:3.8'
environment:
- RABBITMQ_PASSWORD=test
- RABBITMQ_USERNAME=test
ports:
- '4369:4369'
- '5672:5672'
- '25672:25672'
- '15672:15672'
volumes:
- 'rabbitmq_data:/bitnami'
volumes:
rabbitmq_data:
driver: local
I have some questions. I hope to find an answer :
Is it a good practice to start an infinite loop like this in python? Does a better way exist to "hang on"? (don't end the program's execution)
Is my code well architectured? With python best practices?
Is my test relevant to simulate a rabbitMQ shutdown?
Here's a full example of what happens in the terminal
Before answering: please ask for clarification, if needed, and don't forget to answer my questions during the process.
