Add Integration test for RabbitMqEventPublisher.

Author: G8XSU

.github/workflows/integration-tests-events-rabbitmq.yml

@@ -0,0 +1,34 @@
+name: RabbitMQ Integration Tests
+
+on: [push, pull_request]
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  integration-tests:
+    runs-on: ubuntu-latest
+
+    services:
+      rabbitmq:
+        image: rabbitmq:3
+        env:
+          RABBITMQ_DEFAULT_USER: guest
+          RABBITMQ_DEFAULT_PASS: guest
+        ports:
+          - 5672:5672
+        options: >-
+          --health-cmd "rabbitmqctl node_health_check"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Run RabbitMQ integration tests
+        run: cargo test --features integration-tests-events-rabbitmq -- --nocapture --verbose --color=always
+        env:
+          RUST_BACKTRACE: 1

ldk-server/Cargo.toml

@@ -25,3 +25,9 @@ lapin = { version = "2.5.1", optional = true }
 [features]
 default = []
 events-rabbitmq = ["dep:lapin"]
+
+# Feature-flags related to integration tests.
+integration-tests-events-rabbitmq = ["events-rabbitmq"]
+
+[dev-dependencies]
+futures-util = "0.3.31"

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

@@ -143,6 +143,91 @@ impl EventPublisher for RabbitMqEventPublisher {
 	}
 }
 
-// Ensure the publisher is Send and Sync for use in async contexts
-unsafe impl Send for RabbitMqEventPublisher {}
-unsafe impl Sync for RabbitMqEventPublisher {}
+#[cfg(test)]
+#[cfg(feature = "integration-tests-events-rabbitmq")]
+mod integration_tests_events_rabbitmq {
+	use super::*;
+	use lapin::{
+		options::{BasicAckOptions, BasicConsumeOptions, QueueBindOptions, QueueDeclareOptions},
+		types::FieldTable,
+		Channel, Connection,
+	};
+	use ldk_server_protos::events::event_envelope::Event;
+	use ldk_server_protos::events::PaymentForwarded;
+	use std::io;
+	use std::time::Duration;
+	use tokio;
+
+	use futures_util::stream::StreamExt;
+	#[tokio::test]
+	async fn test_publish_and_consume_event() {
+		let config = RabbitMqConfig {
+			connection_string: "amqp://guest:guest@localhost:5672/%2f".to_string(),
+			exchange_name: "test_exchange".to_string(),
+		};
+
+		let publisher = RabbitMqEventPublisher::new(config.clone());
+
+		let conn = Connection::connect(&config.connection_string, ConnectionProperties::default())
+			.await
+			.expect("Failed make rabbitmq connection");
+		let channel = conn.create_channel().await.expect("Failed to create rabbitmq channel");
+
+		let queue_name = "test_queue";
+		setup_queue(&queue_name, &channel, &config).await;
+
+		let event = EventEnvelope {
+			event: Some(Event::PaymentForwarded(PaymentForwarded::default())),
+		};
+		publisher.publish(event.clone()).await.expect("Failed to publish event");
+
+		consume_event(&queue_name, &channel, &event).await.expect("Failed to consume event");
+	}
+
+	async fn setup_queue(queue_name: &str, channel: &Channel, config: &RabbitMqConfig) {
+		channel
+			.queue_declare(queue_name, QueueDeclareOptions::default(), FieldTable::default())
+			.await
+			.unwrap();
+		channel
+			.exchange_declare(
+				&config.exchange_name,
+				ExchangeKind::Fanout,
+				ExchangeDeclareOptions { durable: true, ..Default::default() },
+				FieldTable::default(),
+			)
+			.await
+			.unwrap();
+
+		channel
+			.queue_bind(
+				queue_name,
+				&config.exchange_name,
+				"",
+				QueueBindOptions::default(),
+				FieldTable::default(),
+			)
+			.await
+			.unwrap();
+	}
+
+	async fn consume_event(
+		queue_name: &str, channel: &Channel, expected_event: &EventEnvelope,
+	) -> io::Result<()> {
+		let mut consumer = channel
+			.basic_consume(
+				queue_name,
+				"test_consumer",
+				BasicConsumeOptions::default(),
+				FieldTable::default(),
+			)
+			.await
+			.unwrap();
+		let delivery =
+			tokio::time::timeout(Duration::from_secs(10), consumer.next()).await?.unwrap().unwrap();
+		let received_event = EventEnvelope::decode(&*delivery.data)?;
+		assert_eq!(received_event, *expected_event, "Event mismatch");
+		channel.basic_ack(delivery.delivery_tag, BasicAckOptions::default()).await.unwrap();
+		Ok(())
+	}
+}