Add SubscribeEvents server-streaming RPC

Implement the SubscribeEvents RPC end-to-end: a Stream variant in
GrpcBody that delivers multiple gRPC-framed messages from an mpsc
channel, a handler in service.rs that bridges the broadcast channel
to the streaming response, a client EventStream type that reads
gRPC frames incrementally from the HTTP/2 body, and e2e test
assertions that verify payment events arrive via the stream.

Also regenerates proto code to include SubscribeEventsRequest.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: benthecarman

Cargo.lock

@@ -11,7 +11,4 @@ ldk-server-client = { path = "../ldk-server-client" }
 ldk-server-grpc = { path = "../ldk-server-grpc", features = ["serde"] }
 serde_json = "1.0"
 hex-conservative = { version = "0.2", features = ["std"] }
-lapin = { version = "2.4.0", features = ["rustls"], default-features = false }
-prost = { version = "0.11.6", default-features = false, features = ["std"] }
-futures-util = "0.3"
 ldk-node = { git = "https://github.com/lightningdevkit/ldk-node", rev = "3aef2b39265ae60b29f4d60de8291895f12eb880" }

e2e-tests/Cargo.lock

@@ -21,7 +21,7 @@ fn main() {
 			"-p",
 			"ldk-server",
 			"--features",
-			"events-rabbitmq,experimental-lsps2-support",
+			"experimental-lsps2-support",
 			"-p",
 			"ldk-server-cli",
 		])

e2e-tests/Cargo.toml

@@ -87,26 +87,20 @@ impl TestBitcoind {
 /// Handle to a running ldk-server child process.
 pub struct LdkServerHandle {
 	child: Option<Child>,
-	pub rest_port: u16,
+	pub grpc_port: u16,
 	pub p2p_port: u16,
 	pub storage_dir: PathBuf,
 	pub api_key: String,
 	pub tls_cert_path: PathBuf,
 	pub node_id: String,
-	pub exchange_name: String,
 	client: LdkServerClient,
 }
 
+#[derive(Default)]
 pub struct LdkServerConfig {
 	pub metrics_auth: Option<(String, String)>,
 }
 
-impl Default for LdkServerConfig {
-	fn default() -> Self {
-		Self { metrics_auth: None }
-	}
-}
-
 impl LdkServerHandle {
 	/// Starts a new ldk-server instance against the given bitcoind.
 	/// Waits until the server is ready to accept requests.
@@ -117,14 +111,12 @@ impl LdkServerHandle {
 	pub async fn start_with_config(bitcoind: &TestBitcoind, config: LdkServerConfig) -> Self {
 		#[allow(deprecated)]
 		let storage_dir = tempfile::tempdir().unwrap().into_path();
-		let rest_port = find_available_port();
+		let grpc_port = find_available_port();
 		let p2p_port = find_available_port();
 
 		let (rpc_host, rpc_port_num, rpc_user, rpc_password) = bitcoind.rpc_details();
 		let rpc_address = format!("{rpc_host}:{rpc_port_num}");
 
-		let exchange_name = format!("e2e_test_exchange_{rest_port}");
-
 		let metrics_auth_config = if let Some((user, pass)) = config.metrics_auth {
 			format!("username = \"{}\"\npassword = \"{}\"", user, pass)
 		} else {
@@ -135,7 +127,7 @@ impl LdkServerHandle {
 			r#"[node]
 network = "regtest"
 listening_addresses = ["127.0.0.1:{p2p_port}"]
-rest_service_address = "127.0.0.1:{rest_port}"
+grpc_service_address = "127.0.0.1:{grpc_port}"
 alias = "e2e-test-node"
 
 [storage.disk]
@@ -146,10 +138,6 @@ rpc_address = "{rpc_address}"
 rpc_user = "{rpc_user}"
 rpc_password = "{rpc_password}"
 
-[rabbitmq]
-connection_string = "amqp://guest:guest@localhost:5672/%2f"
-exchange_name = "{exchange_name}"
-
 [liquidity.lsps2_service]
 advertise_service = false
 channel_opening_fee_ppm = 10000
@@ -216,18 +204,17 @@ poll_metrics_interval = 1
 		// Read TLS cert
 		let tls_cert_pem = std::fs::read(&tls_cert_path).unwrap();
 
-		let base_url = format!("127.0.0.1:{rest_port}");
+		let base_url = format!("127.0.0.1:{grpc_port}");
 		let client = LdkServerClient::new(base_url, api_key.clone(), &tls_cert_pem).unwrap();
 
 		let mut handle = Self {
 			child: Some(child),
-			rest_port,
+			grpc_port,
 			p2p_port,
 			storage_dir,
 			api_key,
 			tls_cert_path,
 			node_id: String::new(),
-			exchange_name,
 			client,
 		};
 
@@ -247,7 +234,7 @@ poll_metrics_interval = 1
 	}
 
 	pub fn base_url(&self) -> String {
-		format!("127.0.0.1:{}", self.rest_port)
+		format!("127.0.0.1:{}", self.grpc_port)
 	}
 }
 
@@ -452,101 +439,3 @@ pub async fn setup_funded_channel(
 
 	open_resp.user_channel_id
 }
-
-/// RabbitMQ event consumer for verifying events published by ldk-server.
-pub struct RabbitMqEventConsumer {
-	_connection: lapin::Connection,
-	channel: lapin::Channel,
-	consumer: lapin::Consumer,
-}
-
-impl RabbitMqEventConsumer {
-	/// Connect to RabbitMQ and create an exclusive queue bound to the given exchange.
-	pub async fn new(exchange_name: &str) -> Self {
-		use lapin::options::{
-			BasicConsumeOptions, ExchangeDeclareOptions, QueueBindOptions, QueueDeclareOptions,
-		};
-		use lapin::types::FieldTable;
-		use lapin::{ConnectionProperties, ExchangeKind};
-
-		let connection = lapin::Connection::connect(
-			"amqp://guest:guest@localhost:5672/%2f",
-			ConnectionProperties::default(),
-		)
-		.await
-		.expect("Failed to connect to RabbitMQ");
-
-		let channel = connection.create_channel().await.expect("Failed to create channel");
-
-		// Declare exchange (idempotent — may already exist from the server)
-		channel
-			.exchange_declare(
-				exchange_name,
-				ExchangeKind::Fanout,
-				ExchangeDeclareOptions { durable: true, ..Default::default() },
-				FieldTable::default(),
-			)
-			.await
-			.expect("Failed to declare exchange");
-
-		// Create exclusive auto-delete queue with server-generated name
-		let queue = channel
-			.queue_declare(
-				"",
-				QueueDeclareOptions { exclusive: true, auto_delete: true, ..Default::default() },
-				FieldTable::default(),
-			)
-			.await
-			.expect("Failed to declare queue");
-		let queue_name = queue.name().to_string();
-
-		channel
-			.queue_bind(
-				&queue_name,
-				exchange_name,
-				"",
-				QueueBindOptions::default(),
-				FieldTable::default(),
-			)
-			.await
-			.expect("Failed to bind queue");
-
-		let consumer = channel
-			.basic_consume(
-				&queue_name,
-				&format!("consumer_{}", queue_name),
-				BasicConsumeOptions::default(),
-				FieldTable::default(),
-			)
-			.await
-			.expect("Failed to start consumer");
-
-		Self { _connection: connection, channel, consumer }
-	}
-
-	/// Consume up to `count` events, waiting up to `timeout` for each.
-	pub async fn consume_events(
-		&mut self, count: usize, timeout: Duration,
-	) -> Vec<ldk_server_protos::events::EventEnvelope> {
-		use futures_util::StreamExt;
-		use lapin::options::BasicAckOptions;
-		use prost::Message;
-
-		let mut events = Vec::new();
-		for _ in 0..count {
-			match tokio::time::timeout(timeout, self.consumer.next()).await {
-				Ok(Some(Ok(delivery))) => {
-					let event = ldk_server_protos::events::EventEnvelope::decode(&*delivery.data)
-						.expect("Failed to decode event");
-					self.channel
-						.basic_ack(delivery.delivery_tag, BasicAckOptions::default())
-						.await
-						.expect("Failed to ack");
-					events.push(event);
-				},
-				_ => break,
-			}
-		}
-		events
-	}
-}