[client] Ability to reconnect to RabbitMQ in case of errors

Author: SamuelHassine

README.md

@@ -85,4 +85,6 @@ $ pytest ./tests/02-integration/
 
 ## About
 
-OpenCTI is a product powered by the collaboration of the private company [Filigran](https://www.filigran.io), the [French national cybersecurity agency (ANSSI)](https://ssi.gouv.fr), the [CERT-EU](https://cert.europa.eu) and the [Luatix](https://www.luatix.org) non-profit organization.
+OpenCTI is a product designed and developed by the company [Filigran](https://www.filigran.io).
+
+<a href="https://www.filigran.io" alt="Filigran"><img src="https://www.filigran.io/wp-content/uploads/2022/08/filigran_text_horizontal_dense_margin.png" width="230" /></a>

pycti/connector/opencti_connector_helper.py

@@ -36,7 +36,7 @@ def killProgramHook(etype, value, tb):
     os.kill(os.getpid(), signal.SIGTERM)
 
 
-def run_loop(loop):
+def start_loop(loop):
     asyncio.set_event_loop(loop)
     loop.run_forever()
 
@@ -128,7 +128,6 @@ def __init__(self, helper, config: Dict, callback) -> None:
         self.user = config["connection"]["user"]
         self.password = config["connection"]["pass"]
         self.queue_name = config["listen"]
-        self.exit_event = threading.Event()
         self.connector_thread = None
         self.connector_event_loop = None
         self.queue_event_loop = asyncio.new_event_loop()
@@ -207,7 +206,7 @@ def _data_handler(self, json_data) -> None:
                 LOGGER.error("Failing reporting the processing")
 
     def run(self) -> None:
-        while not self.exit_event.is_set():
+        while True:
             try:
                 # Connect the broker
                 self.pika_credentials = pika.PlainCredentials(self.user, self.password)
@@ -223,25 +222,39 @@ def run(self) -> None:
                 if asyncio.iscoroutinefunction(self.callback):
                     self.connector_event_loop = asyncio.new_event_loop()
                     self.connector_thread = threading.Thread(
-                        target=lambda: run_loop(self.connector_event_loop)
-                    ).start()
+                        target=lambda: start_loop(self.connector_event_loop)
+                    )
+                    self.connector_thread.start()
                 self.pika_connection = AsyncioConnection(
                     self.pika_parameters,
                     on_open_callback=self.on_connection_open,
+                    on_open_error_callback=self.on_connection_open_error,
+                    on_close_callback=self.on_connection_closed,
                     custom_ioloop=self.queue_event_loop,
                 )
                 self.pika_connection.ioloop.run_forever()
+                # If the connection fails, sleep between reconnect attempts
+                time.sleep(10)
             except (KeyboardInterrupt, SystemExit):
                 LOGGER.info("Connector stop")
                 sys.exit(0)
-            except Exception as e:  # pylint: disable=broad-except
-                LOGGER.error("%s", e)
-                time.sleep(10)
+            except Exception as err:  # pylint: disable=broad-except
+                LOGGER.error("%s", err)
 
     # noinspection PyUnusedLocal
     def on_connection_open(self, _unused_connection):
         self.pika_connection.channel(on_open_callback=self.on_channel_open)
 
+    # noinspection PyUnusedLocal
+    def on_connection_open_error(self, _unused_connection, err):
+        LOGGER.info("Unable to connect to the queue. %s", err)
+        self.pika_connection.ioloop.stop()
+
+    # noinspection PyUnusedLocal
+    def on_connection_closed(self, _unused_connection, reason):
+        LOGGER.info("The connection to the queue closed: %s", reason)
+        self.pika_connection.ioloop.stop()
+
     def on_channel_open(self, channel):
         self.channel = channel
         assert self.channel is not None
@@ -252,13 +265,6 @@ def on_channel_open(self, channel):
             ),
         )
 
-    def stop(self):
-        self.queue_event_loop.stop()
-        self.exit_event.set()
-        if self.connector_thread:
-            self.connector_event_loop.stop()
-            self.connector_thread.join()
-
 
 class PingAlive(threading.Thread):
     def __init__(self, connector_id, api, get_state, set_state) -> None: