Added list for events subscription configuration

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,16 @@ 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. Wildcard subscriptions are not
+supported.
+
 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

event-plugin/src/events.rs

@@ -76,49 +76,29 @@ fn encode_envelope(envelope: EventEnvelope) -> Result<Vec<u8>> {
     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<MessageBroker>, 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<MessageBroker>, 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<MessageBroker>, _v: JsonValue) -> Result<()> {
     // TODO: no-op for now, will eff-up integration test teardown otherwise as

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<RwLock<Option<lapin::Channel>>> = 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,34 @@ 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);
+    }
+
+    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";
@@ -146,6 +152,16 @@ async fn main() -> Result<()> {
     Ok(())
 }
 
+fn event_types_from_env() -> Vec<String> {
+    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<u8>,
     version: String,