Add RabbitMQ based EventPublisher Impl.

RabbitMQ was selected because it is one of the most robust and
battle-tested open-source messaging systems, allowing for
durable, buffered messages and flexible exchange options.

This ensures that events are not lost, even in the case of
restarts, network failures, or consumer failures, while
keeping consumers decoupled.

* Feature-based: Only enabled if we enable `events-rabbitmq`
  feature, this allows us to support multiple implementations
  and no strict coupling with a single impl while keep our
  dependencies to the minimum.
* Durable Publishing: Uses publisher confirms and persistent
  delivery mode to guarantee events are stored by RabbitMQ
  before returning Ok(()).
* Lazy Initialized connection: Wraps Connection and Channel
  in Option for lazy connection init via ensure_connected.
* Reconnection Handling: Implements a single reconnect
  attempt per publish call. Event publishing is already
  retried, so this keeps it simple for now.

Author: G8XSU

ldk-server/Cargo.toml

@@ -18,3 +18,10 @@ hex = { package = "hex-conservative", version = "0.2.1", default-features = fals
 rusqlite = { version = "0.31.0", features = ["bundled"] }
 rand = { version = "0.8.5", default-features = false }
 async-trait = { version = "0.1.85", default-features = false }
+
+# Required for RabittMQ based EventPublisher. Only enabled for `events-rabbitmq` feature.
+lapin = { version = "2.4.0", features = ["rustls"], default-features = false, optional = true }
+
+[features]
+default = []
+events-rabbitmq = ["dep:lapin"]

ldk-server/src/io/events/mod.rs

@@ -1,5 +1,8 @@
 pub(crate) mod event_publisher;
 
+#[cfg(feature = "events-rabbitmq")]
+pub(crate) mod rabbitmq;
+
 use ldk_server_protos::events::event_envelope;
 
 /// Event variant to event name mapping.

ldk-server/src/io/events/rabbitmq/mod.rs

@@ -0,0 +1,144 @@
+use crate::api::error::LdkServerError;
+use crate::api::error::LdkServerErrorCode::InternalServerError;
+use crate::io::events::event_publisher::EventPublisher;
+use ::prost::Message;
+use async_trait::async_trait;
+use lapin::options::{BasicPublishOptions, ConfirmSelectOptions, ExchangeDeclareOptions};
+use lapin::types::FieldTable;
+use lapin::{
+	BasicProperties, Channel, Connection, ConnectionProperties, ConnectionState, ExchangeKind,
+};
+use ldk_server_protos::events::EventEnvelope;
+use std::sync::Arc;
+use tokio::sync::Mutex;
+
+/// A RabbitMQ-based implementation of the EventPublisher trait.
+pub struct RabbitMqEventPublisher {
+	/// The RabbitMQ connection, used for reconnection logic.
+	connection: Arc<Mutex<Option<Connection>>>,
+	/// The RabbitMQ channel used for publishing events.
+	channel: Arc<Mutex<Option<Channel>>>,
+	/// Configuration details, including connection string and exchange name.
+	config: RabbitMqConfig,
+}
+
+/// Configuration for the RabbitMQ event publisher.
+#[derive(Debug, Clone)]
+pub struct RabbitMqConfig {
+	pub connection_string: String,
+	pub exchange_name: String,
+}
+
+/// Delivery mode for persistent messages (written to disk).
+const DELIVERY_MODE_PERSISTENT: u8 = 2;
+
+impl RabbitMqEventPublisher {
+	/// Creates a new RabbitMqEventPublisher instance.
+	pub fn new(config: RabbitMqConfig) -> Self {
+		Self { connection: Arc::new(Mutex::new(None)), channel: Arc::new(Mutex::new(None)), config }
+	}
+
+	async fn connect(config: &RabbitMqConfig) -> Result<(Connection, Channel), LdkServerError> {
+		let conn = Connection::connect(&config.connection_string, ConnectionProperties::default())
+			.await
+			.map_err(|e| {
+				LdkServerError::new(
+					InternalServerError,
+					format!("Failed to connect to RabbitMQ: {}", e),
+				)
+			})?;
+
+		let channel = conn.create_channel().await.map_err(|e| {
+			LdkServerError::new(InternalServerError, format!("Failed to create channel: {}", e))
+		})?;
+
+		channel.confirm_select(ConfirmSelectOptions::default()).await.map_err(|e| {
+			LdkServerError::new(InternalServerError, format!("Failed to enable confirms: {}", e))
+		})?;
+
+		channel
+			.exchange_declare(
+				&config.exchange_name,
+				ExchangeKind::Fanout,
+				ExchangeDeclareOptions { durable: true, ..Default::default() },
+				FieldTable::default(),
+			)
+			.await
+			.map_err(|e| {
+				LdkServerError::new(
+					InternalServerError,
+					format!("Failed to declare exchange: {}", e),
+				)
+			})?;
+
+		Ok((conn, channel))
+	}
+
+	async fn ensure_connected(&self) -> Result<(), LdkServerError> {
+		{
+			let connection = self.connection.lock().await;
+			if let Some(connection) = &*connection {
+				if connection.status().state() == ConnectionState::Connected {
+					return Ok(());
+				}
+			}
+		}
+
+		// Connection is not alive, attempt reconnecting.
+		let (connection, channel) = Self::connect(&self.config)
+			.await
+			.map_err(|e| LdkServerError::new(InternalServerError, e.to_string()))?;
+		*self.connection.lock().await = Some(connection);
+		*self.channel.lock().await = Some(channel);
+		Ok(())
+	}
+}
+
+#[async_trait]
+impl EventPublisher for RabbitMqEventPublisher {
+	/// Publishes an event to RabbitMQ.
+	///
+	/// The event is published to a fanout exchange with persistent delivery mode,
+	/// and the method waits for confirmation from RabbitMQ to ensure durability.
+	async fn publish(&self, event: EventEnvelope) -> Result<(), LdkServerError> {
+		// Ensure connection is alive before proceeding
+		self.ensure_connected().await?;
+
+		let channel_guard = self.channel.lock().await;
+		let channel = channel_guard.as_ref().ok_or_else(|| {
+			LdkServerError::new(InternalServerError, "Channel not initialized".to_string())
+		})?;
+
+		// Publish the event with persistent delivery mode
+		let confirm = channel
+			.basic_publish(
+				&self.config.exchange_name,
+				"", // Empty routing key should be used for fanout exchange, since it is ignored.
+				BasicPublishOptions::default(),
+				&event.encode_to_vec(),
+				BasicProperties::default().with_delivery_mode(DELIVERY_MODE_PERSISTENT),
+			)
+			.await
+			.map_err(|e| {
+				LdkServerError::new(
+					InternalServerError,
+					format!("Failed to publish event, error: {}", e),
+				)
+			})?;
+
+		let confirmation = confirm.await.map_err(|e| {
+			LdkServerError::new(InternalServerError, format!("Failed to get confirmation: {}", e))
+		})?;
+
+		match confirmation {
+			lapin::publisher_confirm::Confirmation::Ack(_) => Ok(()),
+			lapin::publisher_confirm::Confirmation::Nack(_) => Err(LdkServerError::new(
+				InternalServerError,
+				"Message not acknowledged".to_string(),
+			)),
+			_ => {
+				Err(LdkServerError::new(InternalServerError, "Unexpected confirmation".to_string()))
+			},
+		}
+	}
+}