4
\$\begingroup\$

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

sample

Before answering: please ask for clarification, if needed, and don't forget to answer my questions during the process.

\$\endgroup\$
  • 2
    \$\begingroup\$ Pika has some example code for re-connection as well as using multiple hosts if you're using a RabbitMQ cluster: github.com/pika/pika/blob/master/examples/… \$\endgroup\$ – Luke Bakken Nov 13 at 13:42
  • \$\begingroup\$ thanks for this @LukeBakken the problem I have with the snippet you linked is the connection.close method, I never want to close the connection, maybe I missunderstood this ? not sure. \$\endgroup\$ – Hooli Nov 13 at 14:50

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.