editoast: use pooling for amqp

Signed-off-by: ElysaSrc <[email protected]>

Author: ElysaSrc

editoast/Cargo.lock

@@ -96,6 +96,7 @@ chrono.workspace = true
 clap = { version = "4.5.19", features = ["derive", "env"] }
 colored = "2.1.0"
 dashmap = "6.1.0"
+deadpool-lapin = "0.12.1"
 derivative.workspace = true
 diesel.workspace = true
 diesel-async = { workspace = true }

editoast/Cargo.toml

@@ -4067,7 +4067,9 @@ components:
       - $ref: '#/components/schemas/EditoastEditionErrorSplitTrackSectionBadOffset'
       - $ref: '#/components/schemas/EditoastEditoastUrlErrorInvalidUrl'
       - $ref: '#/components/schemas/EditoastElectricalProfilesErrorNotFound'
+      - $ref: '#/components/schemas/EditoastErrorCeatePoolLapin'
       - $ref: '#/components/schemas/EditoastErrorConnectionDoesNotExist'
+      - $ref: '#/components/schemas/EditoastErrorDeadpoolLapin'
       - $ref: '#/components/schemas/EditoastErrorLapin'
       - $ref: '#/components/schemas/EditoastErrorResponseTimeout'
       - $ref: '#/components/schemas/EditoastErrorSerialization'
@@ -4129,6 +4131,25 @@ components:
       description: Generated error type for Editoast
       discriminator:
         propertyName: type
+    EditoastErrorCeatePoolLapin:
+      type: object
+      required:
+      - type
+      - status
+      - message
+      properties:
+        context:
+          type: object
+        message:
+          type: string
+        status:
+          type: integer
+          enum:
+          - 500
+        type:
+          type: string
+          enum:
+          - editoast:coreclient:CeatePoolLapin
     EditoastErrorConnectionDoesNotExist:
       type: object
       required:
@@ -4148,6 +4169,25 @@ components:
           type: string
           enum:
           - editoast:coreclient:ConnectionDoesNotExist
+    EditoastErrorDeadpoolLapin:
+      type: object
+      required:
+      - type
+      - status
+      - message
+      properties:
+        context:
+          type: object
+        message:
+          type: string
+        status:
+          type: integer
+          enum:
+          - 500
+        type:
+          type: string
+          enum:
+          - editoast:coreclient:DeadpoolLapin
     EditoastErrorLapin:
       type: object
       required:

editoast/openapi.yaml

@@ -1,23 +1,21 @@
+use deadpool_lapin::{Config, CreatePoolError, Pool, PoolError, Runtime};
 use editoast_derive::EditoastError;
 use futures_util::StreamExt;
 use itertools::Itertools;
 use lapin::{
     options::{BasicConsumeOptions, BasicPublishOptions},
     types::{ByteArray, FieldTable, ShortString},
-    BasicProperties, Connection, ConnectionProperties,
+    BasicProperties,
 };
 use serde::Serialize;
 use serde_json::to_vec;
-use std::{fmt::Debug, sync::Arc};
+use std::fmt::Debug;
 use thiserror::Error;
-use tokio::{
-    sync::RwLock,
-    time::{timeout, Duration},
-};
+use tokio::time::{timeout, Duration};
 
 #[derive(Debug, Clone)]
 pub struct RabbitMQClient {
-    connection: Arc<RwLock<Option<Connection>>>,
+    pub pool: Pool,
     exchange: String,
     timeout: u64,
     hostname: String,
@@ -51,6 +49,12 @@ pub enum Error {
     #[error("Connection does not exist")]
     #[editoast_error(status = "500")]
     ConnectionDoesNotExist,
+    #[error("Cannot create the pool")]
+    #[editoast_error(status = "500")]
+    CeatePoolLapin(CreatePoolError),
+    #[error("Cannot acquire connection from pool")]
+    #[editoast_error(status = "500")]
+    DeadpoolLapin(PoolError),
 }
 
 pub struct MQResponse {
@@ -64,61 +68,20 @@ impl RabbitMQClient {
             .map(|name| name.to_string_lossy().into_owned())
             .unwrap_or_else(|_| "unknown".to_string());
 
-        let conn = Arc::new(RwLock::new(None));
-
-        tokio::spawn(Self::connection_loop(options.uri, conn.clone()));
+        let mut cfg = Config::default();
+        cfg.url = Some(options.uri);
+        let pool = cfg
+            .create_pool(Some(Runtime::Tokio1))
+            .map_err(Error::CeatePoolLapin)?;
 
         Ok(RabbitMQClient {
-            connection: conn,
+            pool,
             exchange: format!("{}-req-xchg", options.worker_pool_identifier),
             timeout: options.timeout,
             hostname,
         })
     }
 
-    async fn connection_ok(connection: &Arc<RwLock<Option<Connection>>>) -> bool {
-        let guard = connection.as_ref().read().await;
-        let conn = guard.as_ref();
-        let status = match conn {
-            None => return false,
-            Some(conn) => conn.status().state(),
-        };
-        match status {
-            lapin::ConnectionState::Initial => true,
-            lapin::ConnectionState::Connecting => true,
-            lapin::ConnectionState::Connected => true,
-            lapin::ConnectionState::Closing => true,
-            lapin::ConnectionState::Closed => false,
-            lapin::ConnectionState::Error => false,
-        }
-    }
-
-    async fn connection_loop(uri: String, connection: Arc<RwLock<Option<Connection>>>) {
-        loop {
-            if Self::connection_ok(&connection).await {
-                tokio::time::sleep(Duration::from_secs(2)).await;
-                continue;
-            }
-
-            tracing::info!("Reconnecting to RabbitMQ");
-
-            // Connection should be re-established
-            let new_connection = Connection::connect(&uri, ConnectionProperties::default()).await;
-
-            match new_connection {
-                Ok(new_connection) => {
-                    *connection.write().await = Some(new_connection);
-                    tracing::info!("Reconnected to RabbitMQ");
-                }
-                Err(e) => {
-                    tracing::error!("Error while reconnecting to RabbitMQ: {:?}", e);
-                }
-            }
-
-            tokio::time::sleep(Duration::from_secs(2)).await;
-        }
-    }
-
     #[allow(dead_code)]
     pub async fn call<T>(
         &self,
@@ -131,14 +94,8 @@ impl RabbitMQClient {
     where
         T: Serialize,
     {
-        // Get current connection
-        let connection = self.connection.read().await;
-        if connection.is_none() {
-            return Err(Error::ConnectionDoesNotExist);
-        }
-        let connection = connection.as_ref().unwrap();
-
         // Create a channel
+        let connection = self.pool.get().await.map_err(Error::DeadpoolLapin)?;
         let channel = connection.create_channel().await.map_err(Error::Lapin)?;
 
         let serialized_payload_vec = to_vec(published_payload).map_err(Error::Serialization)?;
@@ -172,6 +129,12 @@ impl RabbitMQClient {
             .await
             .map_err(Error::Lapin)?;
 
+        // Explicitly close the channel
+        channel
+            .close(200, "Normal shutdown")
+            .await
+            .map_err(Error::Lapin)?;
+
         Ok(())
     }
 
@@ -186,14 +149,8 @@ impl RabbitMQClient {
     where
         T: Serialize,
     {
-        // Get current connection
-        let connection = self.connection.read().await;
-        if connection.is_none() {
-            return Err(Error::ConnectionDoesNotExist);
-        }
-        let connection = connection.as_ref().unwrap();
-
         // Create a channel
+        let connection = self.pool.get().await.map_err(Error::DeadpoolLapin)?;
         let channel = connection.create_channel().await.map_err(Error::Lapin)?;
 
         let serialized_payload_vec = to_vec(published_payload).map_err(Error::Serialization)?;
@@ -247,7 +204,7 @@ impl RabbitMQClient {
         .await
         .map_err(|_| Error::ResponseTimeout)?;
 
-        match response_delivery {
+        let result = match response_delivery {
             Some(Ok(delivery)) => {
                 let status = delivery
                     .properties
@@ -265,7 +222,15 @@ impl RabbitMQClient {
             }
             Some(Err(e)) => Err(e.into()),
             None => panic!("Rabbitmq consumer was cancelled unexpectedly"),
-        }
+        };
+
+        // Explicitly close the channel
+        channel
+            .close(200, "Normal shutdown")
+            .await
+            .map_err(Error::Lapin)?;
+
+        result
     }
 }
 

editoast/src/core/mq_client.rs

@@ -47,7 +47,9 @@
       "Serialization": "Core: cannot serialize request",
       "StatusParsing": "Core: cannot parse status",
       "UnparsableErrorOutput": "Core returned an error in an unknown format",
-      "ConnectionDoesNotExist": "Core: message queue: connection not established"
+      "ConnectionDoesNotExist": "Core: message queue: connection not established",
+      "CeatePoolLapin": "Core: message queue: cannot create pool",
+      "DeadpoolLapin": "Core: message queue: pool error"
     },
     "DatabaseAccessError": "Database access fatal error",
     "document": {

front/public/locales/en/errors.json

@@ -47,7 +47,9 @@
       "Serialization": "Core: impossible de sérialiser la requête",
       "StatusParsing": "Core: impossible d'obtenir le status",
       "UnparsableErrorOutput": "Core: a renvoyé une erreur dans un format inconnu",
-      "ConnectionDoesNotExist": "Core: file d'attente de messages: connexion non établie"
+      "ConnectionDoesNotExist": "Core: file d'attente de messages: connexion non établie",
+      "CeatePoolLapin": "Core: file d'attente de messages: erreur de création de pool",
+      "DeadpoolLapin": "Core: file d'attente de messages: erreur de pool"
     },
     "document": {
       "NotFound": "Document '{{document_key}}' non trouvé"