Remove reader thread in AsyncReader (#309)

CHANGELOG.md

@@ -1,3 +1,7 @@
+# Unreleased
+
+- Remove thread from AsyncReader on Windows ([PR #309](https://github.com/crossterm-rs/crossterm/pull/309))
+
 # Version 0.13.2
 
 - New `input::stop_reading_thread()` function

Cargo.toml

@@ -26,7 +26,7 @@ serde = { version = "1.0.0", features = ["derive"], optional = true }
 
 [target.'cfg(windows)'.dependencies]
 winapi = "0.3.8"
-crossterm_winapi = "0.3.0"
+crossterm_winapi = { version = "0.4.0" }
 
 [target.'cfg(unix)'.dependencies]
 libc = "0.2.51"

src/cursor/sys/windows.rs

@@ -7,7 +7,6 @@ use crossterm_winapi::{is_true, Coord, Handle, HandleType, ScreenBuffer};
 use winapi::{
     shared::minwindef::{FALSE, TRUE},
     um::wincon::{SetConsoleCursorInfo, SetConsoleCursorPosition, CONSOLE_CURSOR_INFO, COORD},
-    um::winnt::HANDLE,
 };
 
 use lazy_static::lazy_static;
@@ -22,7 +21,7 @@ lazy_static! {
 ///
 /// The top left cell is represented `0,0`.
 pub fn position() -> Result<(u16, u16)> {
-    let cursor = ScreenBufferCursor::new()?;
+    let cursor = ScreenBufferCursor::output()?;
     Ok(cursor.position()?.into())
 }
 
@@ -31,7 +30,7 @@ pub(crate) fn show_cursor(show_cursor: bool) -> Result<()> {
 }
 
 pub(crate) fn move_to(column: u16, row: u16) -> Result<()> {
-    let cursor = ScreenBufferCursor::new()?;
+    let cursor = ScreenBufferCursor::output()?;
     cursor.move_to(column as i16, row as i16)?;
     Ok(())
 }
@@ -61,12 +60,12 @@ pub(crate) fn move_left(count: u16) -> Result<()> {
 }
 
 pub(crate) fn save_position() -> Result<()> {
-    ScreenBufferCursor::new()?.save_position()?;
+    ScreenBufferCursor::output()?.save_position()?;
     Ok(())
 }
 
 pub(crate) fn restore_position() -> Result<()> {
-    ScreenBufferCursor::new()?.restore_position()?;
+    ScreenBufferCursor::output()?.restore_position()?;
     Ok(())
 }
 
@@ -76,7 +75,7 @@ struct ScreenBufferCursor {
 }
 
 impl ScreenBufferCursor {
-    fn new() -> Result<ScreenBufferCursor> {
+    fn output() -> Result<ScreenBufferCursor> {
         Ok(ScreenBufferCursor {
             screen_buffer: ScreenBuffer::from(Handle::new(HandleType::CurrentOutputHandle)?),
         })
@@ -163,14 +162,6 @@ impl From<Handle> for ScreenBufferCursor {
     }
 }
 
-impl From<HANDLE> for ScreenBufferCursor {
-    fn from(handle: HANDLE) -> Self {
-        ScreenBufferCursor {
-            screen_buffer: ScreenBuffer::from(handle),
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::{

src/input/input/windows.rs

@@ -1,15 +1,6 @@
 //! This is a WINDOWS specific implementation for input related action.
 
-use std::{
-    char, io,
-    sync::{
-        atomic::{AtomicBool, Ordering},
-        mpsc::{self, Receiver, Sender},
-        Arc, Mutex,
-    },
-    thread,
-    time::Duration,
-};
+use std::{char, collections::VecDeque, io, sync::Mutex};
 
 use crossterm_winapi::{
     ButtonState, Console, ConsoleMode, EventFlags, Handle, InputEventType, KeyEventRecord,
@@ -85,41 +76,15 @@ impl Input for WindowsInput {
     }
 
     fn read_async(&self) -> AsyncReader {
-        AsyncReader::new(Box::new(move |event_tx, cancellation_token| loop {
-            for i in read_input_events().unwrap().1 {
-                if event_tx.send(i).is_err() {
-                    return;
-                }
-            }
-
-            if cancellation_token.load(Ordering::SeqCst) {
-                return;
-            }
-
-            thread::sleep(Duration::from_millis(1));
-        }))
+        let handle = Handle::current_in_handle().expect("failed to create console input handle");
+        let console = Console::from(handle);
+        AsyncReader::new(console, None)
     }
 
     fn read_until_async(&self, delimiter: u8) -> AsyncReader {
-        AsyncReader::new(Box::new(move |event_tx, cancellation_token| loop {
-            for event in read_input_events().unwrap().1 {
-                if let InputEvent::Keyboard(KeyEvent::Char(key)) = event {
-                    if (key as u8) == delimiter {
-                        return;
-                    }
-                }
-
-                if cancellation_token.load(Ordering::SeqCst) {
-                    return;
-                } else {
-                    if event_tx.send(event).is_err() {
-                        return;
-                    }
-                }
-
-                thread::sleep(Duration::from_millis(1));
-            }
-        }))
+        let handle = Handle::current_in_handle().expect("failed to create console input handle");
+        let console = Console::from(handle);
+        AsyncReader::new(console, Some(delimiter))
     }
 
     fn read_sync(&self) -> SyncReader {
@@ -271,9 +236,9 @@ impl Iterator for SyncReader {
 /// } // `reader` dropped <- thread cleaned up, `_raw` dropped <- raw mode disabled
 /// ```
 pub struct AsyncReader {
-    event_rx: Receiver<InputEvent>,
-    shutdown: Arc<AtomicBool>,
-    thread: Option<thread::JoinHandle<()>>,
+    console: Console,
+    buffer: VecDeque<InputEvent>,
+    delimiter: Option<u8>,
 }
 
 impl AsyncReader {
@@ -284,44 +249,15 @@ impl AsyncReader {
     ///
     /// * A thread is spawned to read the input.
     /// * The reading thread is cleaned up when you drop the `AsyncReader`.
-    pub fn new(function: Box<dyn Fn(&Sender<InputEvent>, &Arc<AtomicBool>) + Send>) -> AsyncReader {
-        let shutdown_handle = Arc::new(AtomicBool::new(false));
-
-        let (event_tx, event_rx) = mpsc::channel();
-        let thread_shutdown = shutdown_handle.clone();
-
-        let thread = thread::spawn(move || {
-            function(&event_tx, &thread_shutdown);
-        });
-
+    pub fn new(console: Console, delimiter: Option<u8>) -> AsyncReader {
         AsyncReader {
-            event_rx,
-            shutdown: shutdown_handle,
-            thread: Some(thread),
+            console,
+            buffer: VecDeque::new(),
+            delimiter,
         }
     }
 
-    // TODO If we we keep the Drop semantics, do we really need this in the public API? It's useless as
-    //      there's no `start`, etc.
-    /// Stops the input reader.
-    ///
-    /// # Notes
-    ///
-    /// * The reading thread is cleaned up.
-    /// * You don't need to call this method, because it will be automatically called when the
-    ///   `AsyncReader` is dropped.
-    pub fn stop(&mut self) {
-        if let Some(thread) = self.thread.take() {
-            self.shutdown.store(true, Ordering::SeqCst);
-            thread.join().expect("failed to join background thread");
-        }
-    }
-}
-
-impl Drop for AsyncReader {
-    fn drop(&mut self) {
-        self.stop();
-    }
+    pub fn stop(&mut self) {}
 }
 
 impl Iterator for AsyncReader {
@@ -332,8 +268,31 @@ impl Iterator for AsyncReader {
     /// `None` doesn't mean that the iteration is finished. See the
     /// [`AsyncReader`](struct.AsyncReader.html) documentation for more information.
     fn next(&mut self) -> Option<Self::Item> {
-        let mut iterator = self.event_rx.try_iter();
-        iterator.next()
+        loop {
+            if self.buffer.is_empty() {
+                let (_, events) = read_input_events(&self.console).expect("read failed");
+
+                if events.is_empty() {
+                    return None;
+                }
+
+                self.buffer.extend(events);
+            }
+
+            if let Some(delimiter) = self.delimiter {
+                while let Some(e) = self.buffer.pop_front() {
+                    if let InputEvent::Keyboard(KeyEvent::Char(key)) = e {
+                        if (key as u8) == delimiter {
+                            return Some(e);
+                        }
+                    }
+                }
+
+                continue;
+            }
+
+            return self.buffer.pop_front();
+        }
     }
 }
 
@@ -361,9 +320,7 @@ fn read_single_event() -> Result<Option<InputEvent>> {
 }
 
 /// partially inspired by: https://github.com/retep998/wio-rs/blob/master/src/console.rs#L130
-fn read_input_events() -> Result<(u32, Vec<InputEvent>)> {
-    let console = Console::from(Handle::current_in_handle()?);
-
+fn read_input_events(console: &Console) -> Result<(u32, Vec<InputEvent>)> {
     let result = console.read_console_input()?;
 
     let mut input_events = Vec::with_capacity(result.0 as usize);

src/style/sys/windows.rs

@@ -40,8 +40,7 @@ pub(crate) fn set_foreground_color(fg_color: Color) -> Result<()> {
         color = color | wincon::BACKGROUND_INTENSITY as u16;
     }
 
-    Console::from(**screen_buffer.handle()).set_text_attribute(color)?;
-
+    Console::from(screen_buffer.handle().clone()).set_text_attribute(color)?;
     Ok(())
 }
 
@@ -66,8 +65,7 @@ pub(crate) fn set_background_color(bg_color: Color) -> Result<()> {
         color = color | wincon::FOREGROUND_INTENSITY as u16;
     }
 
-    Console::from(**screen_buffer.handle()).set_text_attribute(color)?;
-
+    Console::from(screen_buffer.handle().clone()).set_text_attribute(color)?;
     Ok(())
 }
 

src/terminal/sys/windows.rs

@@ -53,7 +53,7 @@ pub(crate) fn scroll_up(row_count: u16) -> Result<()> {
         window.top -= count; // move top down
         window.bottom -= count; // move bottom down
 
-        Console::new()?.set_console_info(true, window)?;
+        Console::output()?.set_console_info(true, window)?;
     }
     Ok(())
 }
@@ -70,7 +70,7 @@ pub(crate) fn scroll_down(row_count: u16) -> Result<()> {
         window.top += count; // move top down
         window.bottom += count; // move bottom down
 
-        Console::new()?.set_console_info(true, window)?;
+        Console::output()?.set_console_info(true, window)?;
     }
     Ok(())
 }
@@ -91,7 +91,7 @@ pub(crate) fn set_size(width: u16, height: u16) -> Result<()> {
 
     // get the position of the current console window
     let screen_buffer = ScreenBuffer::current()?;
-    let console = Console::from(**screen_buffer.handle());
+    let console = Console::from(screen_buffer.handle().clone());
     let csbi = screen_buffer.info()?;
 
     let current_size = csbi.buffer_size();