io: manage would block event

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

plugin/src/io.rs

@@ -7,6 +7,40 @@ use std::os::fd::AsRawFd;
 
 const IN: mio::Token = mio::Token(0);
 
+#[macro_export]
+macro_rules! pool_check {
+    ($expr:expr) => {{
+        match $expr {
+            Ok(val) => Ok::<_, std::io::Error>(val),
+            Err(ref err) if err.kind() == std::io::ErrorKind::WouldBlock => {
+                // Handle WouldBlock error
+                // For example, wait for readiness event and retry
+                // You may need to use mio's event loop to wait for readiness
+                // and then retry the operation
+                // For simplicity, we'll just continue the loop here
+                continue;
+            }
+            Err(err) => Err(err.into()),
+        }
+    }};
+}
+
+#[macro_export]
+macro_rules! poll_loop {
+    ($code:block) => {{
+        while let Err(ref err) = $code {
+            if err.kind() == std::io::ErrorKind::WouldBlock {
+                // Handle WouldBlock error
+                // For example, wait for readiness event and retry
+                // You may need to use mio's event loop to wait for readiness
+                // and then retry the operation
+                // For simplicity, we'll just continue the loop here
+                continue;
+            }
+        }
+    }};
+}
+
 pub(crate) struct AsyncIO {
     poll: mio::Poll,
 }
@@ -48,23 +82,22 @@ impl AsyncIO {
                             let mut buffer = String::new();
                             loop {
                                 let mut byte = [0; 1];
-                                reader.read_exact(&mut byte).unwrap();
+                                pool_check!(reader.read_exact(&mut byte))?;
 
                                 // 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") {
-                                    drop(reader);
                                     break; // Exit the loop
                                 }
                             }
                             let Some(resp) = async_callback(buffer.clone()) else {
                                 continue;
                             };
                             let mut writer = io::stdout().lock();
-                            writer.write_all(resp.as_bytes())?;
-                            writer.flush()?;
+                            pool_check!(writer.write_all(resp.as_bytes()))?;
+                            pool_check!(writer.flush())?;
                         }
                     }
                     _ => unreachable!(),

plugin/src/plugin.rs

@@ -79,8 +79,11 @@ impl log::Log for Log {
                 method: "log".to_owned(),
                 params: payload,
             };
-            let _ = writer.write_all(serde_json::to_string(&request).unwrap().as_bytes());
-            let _ = writer.flush();
+
+            crate::poll_loop!({
+                writer.write_all(serde_json::to_string(&request).unwrap().as_bytes())
+            });
+            crate::poll_loop!({ writer.flush() });
         }
     }
 
@@ -122,9 +125,10 @@ impl<'a, T: 'a + Clone> Plugin<T> {
             method: "log".to_owned(),
             params: payload,
         };
-        // We do not like unwrap there
-        let _ = writer.write_all(serde_json::to_string(&request).unwrap().as_bytes());
-        let _ = writer.flush();
+        crate::poll_loop!({
+            writer.write_all(serde_json::to_string(&request).unwrap().as_bytes())
+        });
+        crate::poll_loop!({ writer.flush() });
     }
 
     /// register the plugin option.