Replace pipe-channel by std::io::Pipe

crates/weechat/Cargo.toml

@@ -17,7 +17,7 @@ rustdoc-args = ["--cfg", "feature=\"docs\""]
 
 [features]
 # Support to run futures on the Weechat main thread.
-async = ["async-task", "pipe-channel", "futures", "async-trait"]
+async = ["async-task", "futures", "async-trait"]
 
 # Declarative configuration macro.
 config_macro = ["paste", "strum"]
@@ -33,7 +33,6 @@ libc = "0.2.132"
 backtrace = "0.3.66"
 async-task = { version = "4.3.0", optional = true }
 async-trait = { version = "0.1.57", optional = true }
-pipe-channel = { version = "1.3.0", optional = true }
 futures = { version = "0.3.24", optional = true }
 paste = { version = "1.0.9", optional = true }
 strum = { version = "0.24.1", optional = true }
@@ -43,7 +42,6 @@ weechat-sys = { version = "0.4.0", path = "../weechat-sys" }
 
 [dev-dependencies]
 async-std = "1.12.0"
-pipe-channel = "1.3.0"
 strum = "0.24.1"
 strum_macros = "0.24.3"
 futures = "0.3.24"

crates/weechat/src/executor.rs

@@ -1,12 +1,12 @@
 use std::{
     collections::VecDeque,
+    io::{PipeReader, PipeWriter, Read, Write},
     panic,
     sync::{Arc, Mutex},
 };
 
 pub use async_task::{Runnable, Task};
 use futures::future::{BoxFuture, Future};
-use pipe_channel::{channel, Receiver, Sender};
 
 use crate::{
     hooks::{FdHook, FdHookCallback, FdHookMode},
@@ -43,17 +43,21 @@ type FutureQueue = Arc<Mutex<VecDeque<ExecutorJob>>>;
 
 #[derive(Clone)]
 pub struct WeechatExecutor {
-    _hook: Arc<Mutex<Option<FdHook<Receiver<()>>>>>,
-    sender: Arc<Mutex<Sender<()>>>,
+    _hook: Arc<Mutex<Option<FdHook<PipeReader>>>>,
+    sender: Arc<Mutex<PipeWriter>>,
     futures: FutureQueue,
     non_local_futures: Arc<Mutex<VecDeque<BoxFuture<'static, ()>>>>,
 }
 
 impl FdHookCallback for WeechatExecutor {
-    type FdObject = Receiver<()>;
+    type FdObject = PipeReader;
 
-    fn callback(&mut self, _weechat: &Weechat, receiver: &mut Receiver<()>) {
-        if receiver.recv().is_err() {
+    fn callback(&mut self, _weechat: &Weechat, receiver: &mut PipeReader) {
+        // Consume exactly one wakeup byte; Weechat only calls us once the
+        // read end is ready, so this does not block.
+        let mut wakeup = [0u8; 1];
+
+        if receiver.read_exact(&mut wakeup).is_err() {
             return;
         }
 
@@ -90,7 +94,8 @@ impl FdHookCallback for WeechatExecutor {
 
 impl WeechatExecutor {
     fn new() -> Self {
-        let (sender, receiver) = channel();
+        let (receiver, sender) =
+            std::io::pipe().expect("Can't create the executor notification pipe");
         let sender = Arc::new(Mutex::new(sender));
         let queue = Arc::new(Mutex::new(VecDeque::new()));
         let non_local = Arc::new(Mutex::new(VecDeque::new()));
@@ -132,7 +137,7 @@ impl WeechatExecutor {
                     q.lock().expect("Lock of the future queue of the Weechat executor is poisoned");
 
                 queue.push_back(ExecutorJob::Job(runnable));
-                weechat_notify.send(()).expect("Can't notify Weechat to run a future");
+                weechat_notify.write_all(&[0]).expect("Can't notify Weechat to run a future");
             }
         };
 
@@ -173,7 +178,7 @@ impl WeechatExecutor {
             .sender
             .lock()
             .unwrap()
-            .send(())
+            .write_all(&[0])
             .expect("Can't notify Weechat to spawn a non-local future");
     }
 
@@ -212,7 +217,7 @@ impl WeechatExecutor {
                     q.lock().expect("Lock of the future queue of the Weechat executor is poisoned");
 
                 queue.push_back(ExecutorJob::BufferJob(BufferJob(runnable, buffer_name.clone())));
-                weechat_notify.send(()).expect("Can't notify Weechat to run a future");
+                weechat_notify.write_all(&[0]).expect("Can't notify Weechat to run a future");
             }
         };
 

crates/weechat/src/hooks/fd.rs

@@ -91,23 +91,25 @@ impl<F> FdHook<F> {
     ///
     /// ```no_run
     /// # use weechat::{Weechat, hooks::{FdHook, FdHookMode, FdHookCallback}};
-    /// # use pipe_channel::{channel, Receiver, Sender};
+    /// # use std::io::{PipeReader, Read};
     ///
     /// struct Data;
     ///
     /// impl FdHookCallback for Data {
-    ///     type FdObject = Receiver<String>;
+    ///     type FdObject = PipeReader;
     ///
-    ///     fn callback(&mut self, _: &Weechat, receiver: &mut Receiver<String>) {
-    ///         if let Ok(data) = receiver.recv() {
+    ///     fn callback(&mut self, _: &Weechat, reader: &mut PipeReader) {
+    ///         let mut data = String::new();
+    ///
+    ///         if reader.read_to_string(&mut data).is_ok() {
     ///             Weechat::print(&data)
     ///         }
     ///     }
     /// }
     ///
-    /// let (sender, receiver): (Sender<String>, Receiver<String>) = channel();
+    /// let (reader, _writer) = std::io::pipe().expect("Can't create a pipe");
     ///
-    /// let hook = FdHook::new(receiver, FdHookMode::Read, Data)
+    /// let hook = FdHook::new(reader, FdHookMode::Read, Data)
     ///     .expect("Can't create executor FD hook");
     /// ```
     pub fn new(