Implemented reconnect after a closed rabbitmq connection

django_backend/backend/models.py

@@ -2,8 +2,10 @@
 # pylint: disable=unused-argument
 
 import json
+import time
 import logging
 import pika
+from pika.exceptions import ConnectionClosed
 import requests
 from requests.auth import HTTPBasicAuth
 from django.conf import settings
@@ -17,6 +19,7 @@ from rest_framework.authtoken.models import Token
 from .auth.v1.models import OIDCConfig
 
 LOGGER = logging.getLogger(__name__)
+RECONNECT_TIMEOUT = 3
 
 
 # singleton for simple configs
@@ -99,7 +102,7 @@ class RabbitMQInstance(SingletonModel):
 
     @property
     def connection(self):
-        LOGGER.debug(self._msg('opened connection'))
+        LOGGER.debug(self._msg('connection opened'))
         return pika.BlockingConnection(
             self._connection_parameters,
         )
@@ -114,7 +117,7 @@ class RabbitMQInstance(SingletonModel):
             exchange_type='topic',
         )
         rabbitmq_channel.confirm_delivery()
-        LOGGER.debug(self._msg('opened channel'))
+        LOGGER.debug(self._msg('channel opened'))
 
         return rabbitmq_channel
 
@@ -245,15 +248,19 @@ class RabbitMQInstance(SingletonModel):
                 if self.is_client_connected(site)]
 
     def publish_by_service(self, service, msg):
-        # FIXME publish can fail -> catch error
-        return self.channel.basic_publish(
-            exchange=self.exchange,
-            routing_key=service.routing_key,
-            body=msg,
-            properties=pika.BasicProperties(
-                delivery_mode=1,
-            ),
-        )
+        while True:
+            try:
+                return self.channel.basic_publish(
+                    exchange=self.exchange,
+                    routing_key=service.routing_key,
+                    body=msg,
+                    properties=pika.BasicProperties(
+                        delivery_mode=1,
+                    ),
+                )
+            except ConnectionClosed as exception:
+                LOGGER.info(self._msg('ConnectionClosed: {}'.format(exception)))
+                time.sleep(RECONNECT_TIMEOUT)
 
 
 def user_info_default():