Implement registering and sending custom notifications to lightningd

cln-rpc/Makefile

@@ -8,13 +8,13 @@ DEFAULT_TARGETS += $(CLN_RPC_EXAMPLES) $(CLN_RPC_GENALL)
 
 MSGGEN_GENALL += $(CLN_RPC_GENALL)
 
-target/${RUST_PROFILE}/examples/cln-rpc-getinfo: $(shell find cln-rpc -name *.rs)
+target/${RUST_PROFILE}/examples/cln-rpc-getinfo: ${CLN_RPC_SOURCES} cln-rpc/examples/getinfo.rs
 	cargo build ${CARGO_OPTS} --example cln-rpc-getinfo
 
-target/${RUST_PROFILE}/examples/cln-plugin-startup: $(shell find cln-rpc -name *.rs)
+target/${RUST_PROFILE}/examples/cln-plugin-startup: ${CLN_RPC_SOURCES} plugins/examples/cln-plugin-startup.rs
 	cargo build ${CARGO_OPTS} --example cln-plugin-startup
 
-target/${RUST_PROFILE}/examples/cln-plugin-reentrant: $(shell find plugins/examples -name *.rs)
+target/${RUST_PROFILE}/examples/cln-plugin-reentrant: ${CLN_RPC_SOURCES} plugins/examples/cln-plugin-reentrant.rs
 	cargo build ${CARGO_OPTS} --example cln-plugin-reentrant
 
 

plugins/examples/cln-plugin-startup.rs

@@ -2,8 +2,11 @@
 //! plugins using the Rust API against Core Lightning.
 #[macro_use]
 extern crate serde_json;
-use cln_plugin::{options, Builder, Error, Plugin};
+use cln_plugin::{messages, options, Builder, Error, Plugin};
 use tokio;
+
+const TEST_NOTIF_TAG: &str = "test_custom_notification";
+
 #[tokio::main]
 async fn main() -> Result<(), anyhow::Error> {
     let state = ();
@@ -25,8 +28,15 @@ async fn main() -> Result<(), anyhow::Error> {
             "Retrieve options from this plugin",
             testoptions,
         )
+        .rpcmethod(
+            "test-custom-notification",
+            "send a test_custom_notification event",
+            test_send_custom_notification,
+        )
         .subscribe("connect", connect_handler)
+        .subscribe("test_custom_notification", test_receive_custom_notification)
         .hook("peer_connected", peer_connected_handler)
+        .notification(messages::NotificationTopic::new(TEST_NOTIF_TAG))
         .start(state)
         .await?
     {
@@ -46,6 +56,26 @@ async fn testmethod(_p: Plugin<()>, _v: serde_json::Value) -> Result<serde_json:
     Ok(json!("Hello"))
 }
 
+async fn test_send_custom_notification(
+    p: Plugin<()>,
+    _v: serde_json::Value,
+) -> Result<serde_json::Value, Error> {
+    let custom_notification = json!({
+        "test": "test",
+    });
+    p.send_custom_notification(TEST_NOTIF_TAG.to_string(), custom_notification)
+        .await?;
+    Ok(json!("Notification sent"))
+}
+
+async fn test_receive_custom_notification(
+    _p: Plugin<()>,
+    v: serde_json::Value,
+) -> Result<(), Error> {
+    log::info!("Received a test_custom_notification: {}", v);
+    Ok(())
+}
+
 async fn connect_handler(_p: Plugin<()>, v: serde_json::Value) -> Result<(), Error> {
     log::info!("Got a connect notification: {}", v);
     Ok(())

plugins/src/lib.rs

@@ -5,7 +5,7 @@ use futures::sink::SinkExt;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 extern crate log;
 use log::trace;
-use messages::Configuration;
+use messages::{Configuration, NotificationTopic};
 use options::ConfigOption;
 use std::collections::HashMap;
 use std::future::Future;
@@ -46,6 +46,7 @@ where
     options: Vec<ConfigOption>,
     rpcmethods: HashMap<String, RpcMethod<S>>,
     subscriptions: HashMap<String, Subscription<S>>,
+    notifications: Vec<NotificationTopic>,
     dynamic: bool,
 }
 
@@ -65,6 +66,8 @@ where
     rpcmethods: HashMap<String, AsyncCallback<S>>,
     hooks: HashMap<String, AsyncCallback<S>>,
     subscriptions: HashMap<String, AsyncNotificationCallback<S>>,
+    #[allow(dead_code)] // unsure why rust thinks this field isn't used
+    notifications: Vec<NotificationTopic>,
 }
 
 /// The [PluginDriver] is used to run the IO loop, reading messages
@@ -115,6 +118,7 @@ where
             subscriptions: HashMap::new(),
             options: vec![],
             rpcmethods: HashMap::new(),
+            notifications: vec![],
             dynamic: false,
         }
     }
@@ -124,6 +128,11 @@ where
         self
     }
 
+    pub fn notification(mut self, notif: messages::NotificationTopic) -> Builder<S, I, O> {
+        self.notifications.push(notif);
+        self
+    }
+
     /// Subscribe to notifications for the given `topic`. The handler
     /// is an async function that takes a `Plugin<S>` and the
     /// notification as a `serde_json::Value` as inputs. Since
@@ -274,6 +283,7 @@ where
             input,
             output,
             rpcmethods,
+            notifications: self.notifications,
             subscriptions,
             options: self.options,
             configuration,
@@ -318,6 +328,7 @@ where
             subscriptions: self.subscriptions.keys().map(|s| s.clone()).collect(),
             hooks: self.hooks.keys().map(|s| s.clone()).collect(),
             rpcmethods,
+            notifications: self.notifications.clone(),
             dynamic: self.dynamic,
             nonnumericids: true,
         }
@@ -647,6 +658,22 @@ impl<S> Plugin<S>
 where
     S: Send + Clone,
 {
+    pub async fn send_custom_notification(
+        &self,
+        method: String,
+        v: serde_json::Value,
+    ) -> Result<(), Error> {
+        self.sender
+            .send(json!({
+                "jsonrpc": "2.0",
+                "method": method,
+                "params": v,
+            }))
+            .await
+            .context("sending custom notification")?;
+        Ok(())
+    }
+
     /// Wait for plugin shutdown
     pub async fn join(&self) -> Result<(), Error> {
         self.wait_handle

plugins/src/messages.rs

@@ -150,11 +150,31 @@ pub(crate) struct RpcMethod {
     pub(crate) usage: String,
 }
 
+#[derive(Serialize, Default, Debug, Clone)]
+pub struct NotificationTopic {
+    pub method: String,
+}
+
+impl NotificationTopic {
+    pub fn method(&self) -> &str {
+        &self.method
+    }
+}
+
+impl NotificationTopic {
+    pub fn new(method: &str) -> Self {
+        Self {
+            method: method.to_string(),
+        }
+    }
+}
+
 #[derive(Serialize, Default, Debug)]
 pub(crate) struct GetManifestResponse {
     pub(crate) options: Vec<ConfigOption>,
     pub(crate) rpcmethods: Vec<RpcMethod>,
     pub(crate) subscriptions: Vec<String>,
+    pub(crate) notifications: Vec<NotificationTopic>,
     pub(crate) hooks: Vec<String>,
     pub(crate) dynamic: bool,
     pub(crate) nonnumericids: bool,

tests/test_cln_rs.py

@@ -44,7 +44,7 @@ def test_plugin_start(node_factory):
     assert str(bin_path) in l1.rpc.listconfigs()['configs']['plugin']['values_str']
 
     # Now check that the `testmethod was registered ok
-    l1.rpc.help("testmethod") == {
+    assert l1.rpc.help("testmethod") == {
         'help': [
             {
                 'command': 'testmethod ',
@@ -57,6 +57,8 @@ def test_plugin_start(node_factory):
     }
 
     assert l1.rpc.testmethod() == "Hello"
+    assert l1.rpc.test_custom_notification() == "Notification sent"
+    l1.daemon.wait_for_log(r'Received a test_custom_notification')
 
     l1.connect(l2)
     l1.daemon.wait_for_log(r'Got a connect hook call')