feat(plugin): add async io for the plugin

Adding the support of the async io for reading to the
std io.

Link: #98
Signed-off-by: Vincenzo Palazzo <[email protected]>

Author: vincenzopalazzo

plugin/Cargo.toml

@@ -11,10 +11,11 @@ keywords = [ "plugin", "cln", "rpc", "lightning", "bitcoin" ]
 readme = "README.md"
 
 [dependencies]
+clightningrpc-common = { version = "0.3.0-beta.4" }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-clightningrpc-common = { version = "0.3.0-beta.4" }
 log = { version = "0.4.17", optional = true }
+mio = "0.8.10"
 
 [features]
 log = ["dep:log"]

plugin/src/io.rs

@@ -0,0 +1,70 @@
+//! async io module of the plugin io.
+//!
+//! Vincenzo Palazzo <[email protected]>
+use std::io;
+use std::io::{Read, Write};
+use std::os::fd::AsRawFd;
+
+const SERVER: mio::Token = mio::Token(0);
+
+pub(crate) struct AsyncIO {
+    poll: mio::Poll,
+}
+
+impl AsyncIO {
+    /// Create a new instance of an AsyncIO
+    pub fn new() -> io::Result<Self> {
+        Ok(Self {
+            poll: mio::Poll::new()?,
+        })
+    }
+
+    pub fn register(&mut self) -> io::Result<()> {
+        let stdin = std::io::stdin().as_raw_fd();
+        let mut stdin = mio::unix::SourceFd(&stdin);
+        let stdout = std::io::stdout().as_raw_fd();
+        let mut stdout = mio::unix::SourceFd(&stdout);
+
+        self.poll
+            .registry()
+            .register(&mut stdin, SERVER, mio::Interest::READABLE)?;
+        self.poll
+            .registry()
+            .register(&mut stdout, SERVER, mio::Interest::WRITABLE)?;
+        Ok(())
+    }
+
+    pub fn into_loop<F: FnMut(String) -> String>(&mut self, mut async_callback: F) {
+        let mut events = mio::Events::with_capacity(1024);
+        loop {
+            self.poll.poll(&mut events, None).unwrap();
+
+            for event in events.iter() {
+                match event.token() {
+                    SERVER => {
+                        if event.is_readable() {
+                            let mut reader = io::stdin().lock();
+                            let mut buffer = String::new();
+                            loop {
+                                let mut byte = [0; 1];
+                                reader.read_exact(&mut byte).unwrap();
+
+                                // Append the byte to the buffer
+                                buffer.push(byte[0] as char);
+
+                                // Check if the buffer ends with double newline
+                                if buffer.ends_with("\n\n") {
+                                    break; // Exit the loop
+                                }
+                            }
+                            let resp = async_callback(buffer.clone());
+                            io::stdout().write_all(resp.as_bytes()).unwrap();
+                            io::stdout().flush().unwrap();
+                        }
+                    }
+                    _ => unreachable!(),
+                }
+            }
+        }
+    }
+}

plugin/src/lib.rs

@@ -9,6 +9,7 @@
 #![crate_name = "clightningrpc_plugin"]
 pub mod commands;
 pub mod errors;
+mod io;
 pub mod macros;
 pub mod plugin;
 pub mod types;

plugin/src/plugin.rs

@@ -15,6 +15,7 @@ use crate::commands::builtin::{InitRPC, ManifestRPC};
 use crate::commands::types::{CLNConf, RPCHookInfo, RPCMethodInfo};
 use crate::commands::RPCCommand;
 use crate::errors::PluginError;
+use crate::io::AsyncIO;
 use crate::types::{LogLevel, RpcOption};
 
 #[cfg(feature = "log")]
@@ -257,9 +258,6 @@ impl<'a, T: 'a + Clone> Plugin<T> {
     }
 
     pub fn start(mut self) {
-        let reader = io::stdin();
-        let mut writer = io::stdout();
-        let mut buffer = String::new();
         #[cfg(feature = "log")]
         {
             use std::str::FromStr;
@@ -276,29 +274,26 @@ impl<'a, T: 'a + Clone> Plugin<T> {
                 on_init: self.on_init.clone(),
             }),
         );
-        // FIXME: core lightning end with the double endline, so this can cause
-        // problem for some input reader.
-        // we need to parse the writer, and avoid this while loop
-        while let Ok(_) = reader.read_line(&mut buffer) {
-            let req_str = buffer.to_string();
-            buffer.clear();
-            let Ok(request) = serde_json::from_str::<Request<serde_json::Value>>(&req_str) else {
-                continue;
-            };
+        let mut asyncio = AsyncIO::new().unwrap();
+        asyncio.register().unwrap();
+        asyncio.into_loop(|buffer| {
+            self.log(
+                LogLevel::Info,
+                &format!("looping around the string: {buffer}"),
+            );
+            let request: Request<serde_json::Value> = serde_json::from_str(&buffer).unwrap();
             if let Some(id) = request.id {
                 // when the id is specified this is a RPC or Hook, so we need to return a response
                 let response = self.call_rpc_method(&request.method, request.params);
                 let mut rpc_response = init_success_response(id);
                 self.write_respose(&response, &mut rpc_response);
-                writer
-                    .write_all(serde_json::to_string(&rpc_response).unwrap().as_bytes())
-                    .unwrap();
-                writer.flush().unwrap();
+                return serde_json::to_string(&rpc_response).unwrap();
             } else {
                 // in case of the id is None, we are receiving the notification, so the server is not
                 // interested in the answer.
                 self.handle_notification(&request.method, request.params);
+                return String::new();
             }
-        }
+        });
     }
 }