fixup! feat(plugin): add async io for the plugin

plugin/src/io.rs

@@ -31,11 +31,13 @@ impl AsyncIO {
         Ok(())
     }
 
-    pub fn into_loop<F: FnMut(String) -> String>(&mut self, mut async_callback: F) {
+    pub fn into_loop<F: FnMut(String) -> Option<String>>(
+        &mut self,
+        mut async_callback: F,
+    ) -> io::Result<()> {
+        let mut events = mio::Events::with_capacity(1024);
         loop {
-            let mut events = mio::Events::with_capacity(1024);
-            self.poll.poll(&mut events, None).unwrap();
-
+            self.poll.poll(&mut events, None)?;
             for event in events.iter() {
                 #[cfg(feature = "log")]
                 log::info!("getting the event: {:?}", event);
@@ -56,9 +58,11 @@ impl AsyncIO {
                                     break; // Exit the loop
                                 }
                             }
-                            let resp = async_callback(buffer.clone());
-                            io::stdout().write_all(resp.as_bytes()).unwrap();
-                            io::stdout().flush().unwrap();
+                            let Some(resp) = async_callback(buffer.clone()) else {
+                                continue;
+                            };
+                            io::stdout().write_all(resp.as_bytes())?;
+                            io::stdout().flush()?;
                         }
                     }
                     _ => unreachable!(),

plugin/src/plugin.rs

@@ -7,9 +7,10 @@ use std::io::Write;
 use std::string::String;
 use std::sync::Arc;
 
+use serde_json::Value;
+
 use clightningrpc_common::json_utils::{add_str, init_payload, init_success_response};
 use clightningrpc_common::types::Request;
-use serde_json::Value;
 
 use crate::commands::builtin::{InitRPC, ManifestRPC};
 use crate::commands::types::{CLNConf, RPCHookInfo, RPCMethodInfo};
@@ -256,7 +257,7 @@ impl<'a, T: 'a + Clone> Plugin<T> {
         }
     }
 
-    pub fn start(mut self) {
+    pub fn start(mut self) -> io::Result<()> {
         #[cfg(feature = "log")]
         {
             use std::str::FromStr;
@@ -273,26 +274,33 @@ impl<'a, T: 'a + Clone> Plugin<T> {
                 on_init: self.on_init.clone(),
             }),
         );
-        let mut asyncio = AsyncIO::new().unwrap();
-        asyncio.register().unwrap();
+        let mut asyncio = AsyncIO::new()?;
+        asyncio.register()?;
         asyncio.into_loop(|buffer| {
-            self.log(
-                LogLevel::Info,
-                &format!("looping around the string: {buffer}"),
-            );
+            #[cfg(feature = "log")]
+            log::info!("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);
-                return serde_json::to_string(&rpc_response).unwrap();
+                #[cfg(feature = "log")]
+                log::info!(
+                    "rpc or hook: {} with reponse {:?}",
+                    request.method,
+                    rpc_response
+                );
+                return Some(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();
+                #[cfg(feature = "log")]
+                log::info!("notification: {}", request.method);
+                return None;
             }
-        });
+        })?;
+        Ok(())
     }
 }