[backend] Handle rabbitmq close event to catch dropped connections

OpenCTI-Platform · Apr 2, 2024 · d6227cf · d6227cf
1 parent 7b0bcf3
commit d6227cf
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 2 deletions.
diff --git a/opencti-platform/opencti-graphql/src/connector/importCsv/importCsv-connector.ts b/opencti-platform/opencti-graphql/src/connector/importCsv/importCsv-connector.ts
@@ -13,6 +13,7 @@ import { parseCsvMapper } from '../../modules/internal/csvMapper/csvMapper-utils
 import type { ConnectorConfig } from '../connector';
 import { IMPORT_CSV_CONNECTOR } from './importCsv';
 import { internalLoadById } from '../../database/middleware-loader';
+import { UnknownError } from '../../config/errors';
 
 const RETRY_CONNECTION_PERIOD = 10000;
 
@@ -102,7 +103,13 @@ const initImportCsvConnector = () => {
 
   const handleCsvImport = async (context: AuthContext) => {
     consumeQueue(context, connector.id, connectionSetterCallback, consumeQueueCallback).catch(() => {
-      if (rabbitMqConnection) rabbitMqConnection.close();
+      if (rabbitMqConnection) {
+        try {
+          rabbitMqConnection.close();
+        } catch (e) {
+          logApp.error(UnknownError('RabbitMQ connection close fail', { cause: e }));
+        }
+      }
       setTimeout(handleCsvImport, RETRY_CONNECTION_PERIOD);
     });
   };

diff --git a/opencti-platform/opencti-graphql/src/database/rabbitmq.js b/opencti-platform/opencti-graphql/src/database/rabbitmq.js
@@ -245,11 +245,16 @@ export const consumeQueue = async (context, connectorId, connectionSetterCallbac
   } : amqpCred();
   return new Promise((_, reject) => {
     try {
-      amqp.connect(amqpUri(), connOptions, (err, conn) => {
+      amqp.connect(amqpUri(), { ...connOptions }, (err, conn) => {
         if (err) {
           reject(err);
         } else { // Connection success
           logApp.info('Starting connector queue consuming');
+          conn.on('close', (onConnectError) => {
+            if (onConnectError) {
+              reject(onConnectError);
+            }
+          });
           conn.on('error', (onConnectError) => {
             reject(onConnectError);
           });