Publish events to RabbitMQ queue if events-rabbitmq feature is enabled.

Author: G8XSU

ldk-server/ldk-server.config

@@ -18,5 +18,12 @@
   "bitcoind_rpc_user": "polaruser",
 
   // Bitcoin Core's RPC password.
-  "bitcoind_rpc_password": "polarpass"
+  "bitcoind_rpc_user": "polarpass",
+
+  // RabbitMQ connection string. (only required if using RabbitMQ based events using `events-rabbitmq` feature)
+  "rabbitmq_connection_string": "",
+
+  // RabbitMQ exchange name. (only required if using RabbitMQ based events using `events-rabbitmq` feature)
+  "rabbitmq_exchange_name": ""
+
 }

ldk-server/src/main.rs

@@ -15,6 +15,8 @@ use hyper_util::rt::TokioIo;
 
 use crate::io::events::event_publisher::{EventPublisher, NoopEventPublisher};
 use crate::io::events::get_event_name;
+#[cfg(feature = "events-rabbitmq")]
+use crate::io::events::rabbitmq::{RabbitMqConfig, RabbitMqEventPublisher};
 use crate::io::persist::paginated_kv_store::PaginatedKVStore;
 use crate::io::persist::sqlite_store::SqliteStore;
 use crate::io::persist::{
@@ -106,6 +108,15 @@ fn main() {
 
 	let event_publisher: Arc<dyn EventPublisher> = Arc::new(NoopEventPublisher);
 
+	#[cfg(feature = "events-rabbitmq")]
+	let event_publisher: Arc<dyn EventPublisher> = {
+		let rabbitmq_config = RabbitMqConfig {
+			connection_string: config_file.rabbitmq_connection_string,
+			exchange_name: config_file.rabbitmq_exchange_name,
+		};
+		Arc::new(RabbitMqEventPublisher::new(rabbitmq_config))
+	};
+
 	println!("Starting up...");
 	match node.start_with_runtime(Arc::clone(&runtime)) {
 		Ok(()) => {},

ldk-server/src/util/config.rs

@@ -16,6 +16,8 @@ pub struct Config {
 	pub bitcoind_rpc_addr: SocketAddr,
 	pub bitcoind_rpc_user: String,
 	pub bitcoind_rpc_password: String,
+	pub rabbitmq_connection_string: String,
+	pub rabbitmq_exchange_name: String,
 }
 
 impl TryFrom<JsonConfig> for Config {
@@ -45,6 +47,16 @@ impl TryFrom<JsonConfig> for Config {
 				)
 			})?;
 
+		#[cfg(feature = "events-rabbitmq")]
+		if json_config.rabbitmq_connection_string.as_deref().map_or(true, |s| s.is_empty())
+			|| json_config.rabbitmq_exchange_name.as_deref().map_or(true, |s| s.is_empty())
+		{
+			return Err(io::Error::new(
+				io::ErrorKind::InvalidInput,
+				"Both `rabbitmq_connection_string` and `rabbitmq_exchange_name` must be configured if enabling `events-rabbitmq` feature.".to_string(),
+			));
+		}
+
 		Ok(Config {
 			listening_addr,
 			network: json_config.network,
@@ -53,6 +65,8 @@ impl TryFrom<JsonConfig> for Config {
 			bitcoind_rpc_addr,
 			bitcoind_rpc_user: json_config.bitcoind_rpc_user,
 			bitcoind_rpc_password: json_config.bitcoind_rpc_password,
+			rabbitmq_connection_string: json_config.rabbitmq_connection_string.unwrap_or_default(),
+			rabbitmq_exchange_name: json_config.rabbitmq_exchange_name.unwrap_or_default(),
 		})
 	}
 }
@@ -67,6 +81,8 @@ pub struct JsonConfig {
 	bitcoind_rpc_address: String,
 	bitcoind_rpc_user: String,
 	bitcoind_rpc_password: String,
+	rabbitmq_connection_string: Option<String>,
+	rabbitmq_exchange_name: Option<String>,
 }
 
 /// Loads the configuration from a JSON file at the given path.
@@ -114,6 +130,8 @@ mod tests {
 			"bitcoind_rpc_address":"127.0.0.1:8332", // comment-1
 			"bitcoind_rpc_user": "bitcoind-testuser",
 			"bitcoind_rpc_password": "bitcoind-testpassword",
+			"rabbitmq_connection_string": "rabbitmq_connection_string",
+			"rabbitmq_exchange_name": "rabbitmq_exchange_name",
 			"unknown_key": "random-value"
 			// comment-2
 			}"#;
@@ -130,6 +148,8 @@ mod tests {
 				bitcoind_rpc_addr: SocketAddr::from_str("127.0.0.1:8332").unwrap(),
 				bitcoind_rpc_user: "bitcoind-testuser".to_string(),
 				bitcoind_rpc_password: "bitcoind-testpassword".to_string(),
+				rabbitmq_connection_string: "rabbitmq_connection_string".to_string(),
+				rabbitmq_exchange_name: "rabbitmq_exchange_name".to_string(),
 			}
 		)
 	}