for the docker

rmq_helper/__init__.py

@@ -15,18 +15,23 @@ class RabbitMQ:
 
     def init_app(self, app: Flask):
         self.host = app.config.get('RABBITMQ_HOST', 'localhost')
-        self.connection = pika.BlockingConnection(
-            pika.ConnectionParameters(host=self.host))
-        self.channel = self.connection.channel()
+
+    def _ensure_connection(self):
+        if self.channel is None:
+            self.connection = pika.BlockingConnection(
+                pika.ConnectionParameters(host=self.host))
+            self.channel = self.connection.channel()
 
     def ensure_queue_exists(self, queue_name: str):
         """Ensures the queue exists on the default exchange
 
         :param queue_name: the name of the queue
         """
+        self._ensure_connection()
         self.channel.queue_declare(queue=queue_name, durable=True)
 
     def queue_publish(self, queue_name: str, message):
+        self._ensure_connection()
         self.ensure_queue_exists(queue_name)
 
         body = json.dumps(message)
@@ -38,6 +43,7 @@ class RabbitMQ:
 
         :param name: the name of the exchange
         """
+        self._ensure_connection()
         self.channel.exchange_declare(exchange=name, exchange_type='topic', durable=True)
 
     def exchange_publish(self, exchange_name: str, topic: str, message):
@@ -47,6 +53,7 @@ class RabbitMQ:
         :param topic: the topic to publish to
         :param message: the message to publish (will be turned into a json)
         """
+        self._ensure_connection()
         self.ensure_exchange_exists(exchange_name)
 
         body = json.dumps(message)

rmq_helper/exchanges.py

@@ -1,31 +0,0 @@
-import json
-
-from pika.adapters.blocking_connection import BlockingChannel
-
-from .extension import with_pika_channel
-
-
-@with_pika_channel
-def ensure_exchange_exists(name: str, channel: BlockingChannel = None) -> None:
-    """Ensures exchange exists
-
-    :param name: the name of the exchange
-    :param channel: a pika channel, if none specified a new one will be created
-    """
-    channel.exchange_declare(exchange=name, exchange_type='topic')
-
-
-@with_pika_channel
-def exchange_publish(exchange_name: str, topic: str, message, channel: BlockingChannel = None):
-    """Publishes a message to a
-
-    :param exchange_name: the name of the exchange
-    :param topic: the topic to publish to
-    :param message: the message to publish (will be turned into a json)
-    :param channel: a pika channel, if none specified a new one will be created
-    """
-    ensure_exchange_exists(exchange_name, channel=channel)
-
-    body = json.dumps(message)
-    body_bytes = body.encode('utf-8')
-    channel.basic_publish(exchange=exchange_name, routing_key=topic, body=body_bytes)

rmq_helper/extension.py

@@ -1,25 +0,0 @@
-from functools import wraps
-
-import pika
-
-from .settings import RABBITMQ_HOST
-
-
-def with_pika_channel(f):
-    @wraps(f)
-    def wrapper(*args, **kwargs):
-        # if we are already passing channel, just reuse it
-        if 'channel' in kwargs:
-            return f(*args, **kwargs)
-
-        # otherwise set up a new connection and channel to wrap our code with
-        connection = pika.BlockingConnection(
-            pika.ConnectionParameters(host=RABBITMQ_HOST))
-        channel = connection.channel()
-
-        ret = f(*args, **kwargs, channel=channel)
-
-        channel.close()
-        return ret
-
-    return wrapper

rmq_helper/queues.py

@@ -1,19 +0,0 @@
-import json
-
-from pika.adapters.blocking_connection import BlockingChannel
-
-from .extension import with_pika_channel
-
-
-@with_pika_channel
-def ensure_queue_exists(name: str, channel: BlockingChannel = None):
-    channel.queue_declare(queue=name)
-
-
-@with_pika_channel
-def queue_publish(queue_name: str, message, channel: BlockingChannel = None):
-    ensure_queue_exists(queue_name, channel=channel)
-
-    body = json.dumps(message)
-    body_bytes = body.encode('utf-8')
-    channel.basic_publish(exchange='', routing_key=queue_name, body=body_bytes)

rmq_helper/settings.py

@@ -1,3 +0,0 @@
-import os
-
-RABBITMQ_HOST = os.environ.get("RABBITMQ_HOST", "localhost")