From 6163940768e1c92b908d5fc35fc0b7549c41b04c Mon Sep 17 00:00:00 2001 From: EduardMikhrin Date: Thu, 16 Jul 2026 13:09:35 +0300 Subject: [PATCH 1/5] event-plugin: configure subscriptions from environment --- event-plugin/src/events.rs | 64 +++++++++++++------------------------- event-plugin/src/main.rs | 50 +++++++++++++++++------------ 2 files changed, 52 insertions(+), 62 deletions(-) diff --git a/event-plugin/src/events.rs b/event-plugin/src/events.rs index ac3895e..86c9781 100644 --- a/event-plugin/src/events.rs +++ b/event-plugin/src/events.rs @@ -76,49 +76,29 @@ fn encode_envelope(envelope: EventEnvelope) -> Result> { Ok(buf) } -/// Convenience function to handle events. -macro_rules! define_event_handler { - ($fn_name:ident, $event_type:expr) => { - pub async fn $fn_name(p: Plugin, v: JsonValue) -> Result<()> { - debug!( - "handled event, event_type={}, payload={:?}", - $event_type, &v - ); - let state = p.state(); - state - .publish_bytes(encode_envelope( - new_envelope( - $event_type, - state.source_kind(), - state.source_node_id(), - state.producer_version(), - &v, - ) - .await?, - )?) - .await - } - }; -} +pub async fn on_event(p: Plugin, event_type: String, v: JsonValue) -> Result<()> { + if event_type == "warning" { + return on_warning(p, v).await; + } -// Define event handler. -define_event_handler!(on_connect, "connect"); -define_event_handler!(on_disconnect, "disconnect"); -define_event_handler!(on_invoice_creation, "invoice_creation"); -define_event_handler!(on_invoice_payment, "invoice_payment"); -define_event_handler!(on_channel_opened, "channel_opened"); -define_event_handler!(on_channel_open_failed, "channel_open_failed"); -define_event_handler!(on_channel_state_changed, "channel_state_changed"); -define_event_handler!(on_forward_event, "forward_event"); -define_event_handler!(on_block_added, "block_added"); -define_event_handler!(on_custommsg, "custommsg"); -define_event_handler!(on_sendpay_success, "sendpay_success"); -define_event_handler!(on_sendpay_failure, "sendpay_failure"); -define_event_handler!(on_coin_movement, "coin_movement"); -define_event_handler!(on_openchannel_peer_sigs, "openchannel_peer_sigs"); -define_event_handler!(on_onionmessage_forward_fail, "onionmessage_forward_fail"); -define_event_handler!(on_pay_part_start, "pay_part_start"); -define_event_handler!(on_pay_part_end, "pay_part_end"); + debug!( + "handled event, event_type={}, payload={:?}", + &event_type, &v + ); + let state = p.state(); + state + .publish_bytes(encode_envelope( + new_envelope( + &event_type, + state.source_kind(), + state.source_node_id(), + state.producer_version(), + &v, + ) + .await?, + )?) + .await +} pub async fn on_warning(_p: Plugin, _v: JsonValue) -> Result<()> { // TODO: no-op for now, will eff-up integration test teardown otherwise as diff --git a/event-plugin/src/main.rs b/event-plugin/src/main.rs index a4730fd..f60ebeb 100644 --- a/event-plugin/src/main.rs +++ b/event-plugin/src/main.rs @@ -11,6 +11,7 @@ use lapin::options::{ExchangeDeclareOptions, QueueDeclareOptions}; use lapin::types::FieldTable; use lapin::{Connection, ConnectionProperties, ExchangeKind}; use serde::Deserialize; +use std::env; use std::fmt; use std::fmt::{Debug, Display, Formatter}; use std::path::PathBuf; @@ -22,12 +23,14 @@ use tracing::info; const DEFAULT_EXCHANGE_NAME: &str = "cln.events"; const DEFAULT_QUEUE_NAME: &str = "events"; +const EVENT_TYPES_ENV: &str = "EVENT_PLUGIN_EVENTS"; #[tokio::main] async fn main() -> Result<()> { let amqp_channel: Arc>> = Arc::new(RwLock::new(None)); + let event_types = event_types_from_env(); - let configured = Builder::new(tokio::io::stdin(), tokio::io::stdout()) + let mut builder = Builder::new(tokio::io::stdin(), tokio::io::stdout()) .option(ConfigOption::new_str_no_default( "rabbitmq-url", "AMQP URL: user:pass@host:port/vhost", @@ -45,31 +48,28 @@ async fn main() -> Result<()> { .option(ConfigOption::new_str_no_default( "source-kind", "Event source kind: 'gateway' or 'lsp'", - )) - .subscribe("connect", on_connect) - .subscribe("disconnect", on_disconnect) - .subscribe("invoice_creation", on_invoice_creation) - .subscribe("invoice_payment", on_invoice_payment) - .subscribe("channel_opened", on_channel_opened) - .subscribe("channel_open_failed", on_channel_open_failed) - .subscribe("channel_state_changed", on_channel_state_changed) - .subscribe("forward_event", on_forward_event) - .subscribe("block_added", on_block_added) - .subscribe("custommsg", on_custommsg) - .subscribe("warning", on_warning) - .subscribe("sendpay_success", on_sendpay_success) - .subscribe("sendpay_failure", on_sendpay_failure) - .subscribe("coin_movement", on_coin_movement) - .subscribe("openchannel_peer_sigs", on_openchannel_peer_sigs) - .subscribe("onionmessage_forward_fail", on_onionmessage_forward_fail) - .subscribe("pay_part_start", on_pay_part_start) - .subscribe("pay_part_end", on_pay_part_end) + )); + + for event_type in &event_types { + let handler_event_type = event_type.clone(); + builder = builder.subscribe(event_type, move |p, v| { + on_event(p, handler_event_type.clone(), v) + }); + } + + let configured = builder .dynamic() .configure() .await? // Fail early and loud, we don't want to run the node without this plugin. .context("failed to configure event-plugin")?; + if event_types.is_empty() { + let msg = format!("'{EVENT_TYPES_ENV}' must contain at least one event type"); + _ = configured.disable(&msg).await; + bail!(msg); + } + // Fail if RabbitMQ is not configured, we can not operate without. let Some(url) = get_configured_string_option(&configured, "rabbitmq-url") else { let msg = "'rabbitmq-url' option is required but not set"; @@ -146,6 +146,16 @@ async fn main() -> Result<()> { Ok(()) } +fn event_types_from_env() -> Vec { + env::var(EVENT_TYPES_ENV) + .unwrap_or_default() + .split(',') + .map(str::trim) + .filter(|event_type| !event_type.is_empty()) + .map(str::to_string) + .collect() +} + struct NodeInfo { node_id: Vec, version: String, From 6b2dce95c282e56da8c73515fb387b980259d3c8 Mon Sep 17 00:00:00 2001 From: EduardMikhrin Date: Thu, 16 Jul 2026 13:14:40 +0300 Subject: [PATCH 2/5] docs: document configurable event subscriptions --- event-plugin/README.md | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/event-plugin/README.md b/event-plugin/README.md index 07420ec..c268561 100644 --- a/event-plugin/README.md +++ b/event-plugin/README.md @@ -6,7 +6,7 @@ Core Lightning plugin that collects events and sends them to RabbitMQ ## Events handled -There are 8 events (subscriptions): +The plugin subscribes to the CLN notifications listed in `EVENT_PLUGIN_EVENTS`. Common event types are: - `connect` - A peer connected to the node - `disconnect` - A peer disconnected from the node @@ -32,11 +32,15 @@ RabbitMQ, and then return "continue" so that CLN can continue the operation Set RabbitMQ URL with the plugin option: ``` +EVENT_PLUGIN_EVENTS=connect,disconnect,invoice_creation,invoice_payment,channel_opened,channel_state_changed,forward_event,block_added --rabbitmq-url=amqp://guest:guest@localhost:5672/%2f --rabbitmq-exchange=some_exchange --rabbitmq-queue=some_queue ``` +`EVENT_PLUGIN_EVENTS` is required and must be set in the environment before CLN starts the plugin. Its comma-separated +value defines the notifications advertised to CLN during the plugin handshake. + If RabbitMQ is not available at **startup**, the plugin exits with an error and CLN will not load it. If the connection drops **during operation**, the plugin keeps running but events cannot be published. Each failed From 55b47cca961b39fc50c3725ca4b5ad97c652a96d Mon Sep 17 00:00:00 2001 From: EduardMikhrin Date: Thu, 16 Jul 2026 13:16:15 +0300 Subject: [PATCH 3/5] neat: update Cargo.lock --- Cargo.lock | 25 ++++++++++++------------- 1 file changed, 12 insertions(+), 13 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 87388d3..f79f90c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -345,9 +345,9 @@ checksum = "32637268377fc7b10a8c6d51de3e7fba1ce5dd371a96e342b34e6078db558e7f" [[package]] name = "bitcoin" -version = "0.32.101" +version = "0.32.102" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8ed8ccb78a9ff7a6fbb90e2fb9b8588b4a9928d49d8af3cb789108a84ea6b0ce" +checksum = "bb0ce8bd5baaa0d303a19915a6d93afed161f528654e42da2a7a97d05c59499a" dependencies = [ "base58ck", "bech32", @@ -362,21 +362,20 @@ dependencies = [ [[package]] name = "bitcoin-consensus-encoding" -version = "1.0.0" +version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b2d6094e2a1ba3c93b5a596fe5a10d1a10c3c6e06785cde89f693a044c01aa40" +checksum = "207311705279250ba465076a1bac4b1ac982855fff73fc5f67e22158ac58cdc9" dependencies = [ "bitcoin-internals", + "hex-conservative 1.2.0", + "serde", ] [[package]] name = "bitcoin-internals" -version = "0.5.0" +version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a30a22d1f112dde8e16be7b45c63645dc165cef254f835b3e1e9553e485cfa64" -dependencies = [ - "hex-conservative 0.3.2", -] +checksum = "d573f4cf32996a8dce612e4348cece65a241f1882ed594047c9ba348e8869fa5" [[package]] name = "bitcoin-io" @@ -1105,9 +1104,9 @@ dependencies = [ [[package]] name = "hex-conservative" -version = "0.3.2" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "830e599c2904b08f0834ee6337d8fe8f0ed4a63b5d9e7a7f49c0ffa06d08d360" +checksum = "35431185f361ccf3ffc58254628af5f1f5d5f28531da2e02e5d6c82bbc282a10" dependencies = [ "arrayvec", ] @@ -2587,9 +2586,9 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" [[package]] name = "syn" -version = "2.0.118" +version = "2.0.119" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" dependencies = [ "proc-macro2", "quote", From 283d5676ea14eb617eb1b5427e90f1c99cc1f06a Mon Sep 17 00:00:00 2001 From: EduardMikhrin Date: Thu, 16 Jul 2026 13:28:48 +0300 Subject: [PATCH 4/5] event-plugin: reject wildcard subscriptions --- event-plugin/src/main.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/event-plugin/src/main.rs b/event-plugin/src/main.rs index f60ebeb..d9f36d9 100644 --- a/event-plugin/src/main.rs +++ b/event-plugin/src/main.rs @@ -70,6 +70,12 @@ async fn main() -> Result<()> { bail!(msg); } + if event_types.iter().any(|event_type| event_type == "*") { + let msg = format!("'{EVENT_TYPES_ENV}' does not support wildcard subscriptions"); + _ = configured.disable(&msg).await; + bail!(msg); + } + // Fail if RabbitMQ is not configured, we can not operate without. let Some(url) = get_configured_string_option(&configured, "rabbitmq-url") else { let msg = "'rabbitmq-url' option is required but not set"; From 261018f2f0ef0259db4d08fc84ff6b321f92a7a7 Mon Sep 17 00:00:00 2001 From: EduardMikhrin Date: Thu, 16 Jul 2026 13:28:56 +0300 Subject: [PATCH 5/5] docs: clarify wildcard subscription support --- event-plugin/README.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/event-plugin/README.md b/event-plugin/README.md index c268561..808c9be 100644 --- a/event-plugin/README.md +++ b/event-plugin/README.md @@ -39,7 +39,8 @@ EVENT_PLUGIN_EVENTS=connect,disconnect,invoice_creation,invoice_payment,channel_ ``` `EVENT_PLUGIN_EVENTS` is required and must be set in the environment before CLN starts the plugin. Its comma-separated -value defines the notifications advertised to CLN during the plugin handshake. +value defines the notifications advertised to CLN during the plugin handshake. Wildcard subscriptions are not +supported. If RabbitMQ is not available at **startup**, the plugin exits with an error and CLN will not load it.