wooooo! switch to using tokio; make ^C work properly

1 files changed, 46 insertions, 43 deletions

diff --git a/src/terminal.rs b/src/terminal.rs
index 73ba035..f42d153 100644
--- a/src/terminal.rs
+++ b/src/terminal.rs
@@ -1,15 +1,15 @@
 #![forbid(unsafe_code)]
-use crate::prelude::*;
 use crate::terminal::prelude::*;
 
 use crate::terminal::decoding::CharBufReader;
 
-use async_std::io::{self, Read};
-use async_std::os::unix::io::{AsRawFd, RawFd};
-use async_std::sync::{Arc, Mutex};
 use nix::sys::termios::{self, Termios};
-use signal_hook::consts::TERM_SIGNALS;
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::os::unix::io::{AsRawFd, RawFd};
+use std::sync::Arc;
+use tokio::io::{self, AsyncRead, AsyncWriteExt};
+use tokio::signal::unix::{signal, SignalKind};
+use tokio::sync::{mpsc, Mutex};
+use tokio::task;
 
 pub mod decoding;
 pub mod error;
@@ -87,31 +87,41 @@ enum InputAction {
 }
 
 
-pub struct Terminal<InputStream: Read + Unpin> {
+pub struct Terminal<InputStream: AsyncRead + Unpin> {
   reader: Arc<Mutex<CharBufReader<InputStream>>>,
-  //reader: Arc<CharBufReader<InputStream>>,
   line_buffer: LineBuffer,
   file_descriptor: RawFd,
   initial_termios: Termios,
-  is_interrupted: Arc<AtomicBool>,
+  interrupt_receiver: mpsc::Receiver<()>,
 }
 
 
-impl<InputStream: Read + AsRawFd + Unpin> Terminal<InputStream> {
+async fn handle_signals(interrupt_sender: mpsc::Sender<()>) -> Result<()>
+{
+  let mut stream = signal(SignalKind::interrupt()).map_err(error::internal)?;
+  /* TODO make it work on other signals:
+    SignalKind::hangup();
+    SignalKind::interrupt();
+    SignalKind::terminate();
+    SignalKind::quit();
+    */
+
+  loop {
+    stream.recv().await;
+    interrupt_sender.send(()).await.map_err(error::internal)?;
+  }
+}
+
+
+impl<InputStream: AsyncRead + AsRawFd + Unpin> Terminal<InputStream> {
   pub fn init(input_stream: InputStream) -> Result<Terminal<InputStream>> {
-    let is_interrupted = Arc::new(AtomicBool::new(false));
+    let (interrupt_sender, interrupt_receiver) = mpsc::channel(1);
 
-    for signal in TERM_SIGNALS {
-      signal_hook::flag::register(*signal, Arc::clone(&is_interrupted))
-          .map_err(error::internal)?;
-    }
+    let _ = task::spawn(handle_signals(interrupt_sender));
 
     let fd = input_stream.as_raw_fd();
     let termios = termios::tcgetattr(fd).map_err(error::mode_setting)?;
-    let reader = Arc::new(Mutex::new(CharBufReader::new(
-          input_stream, Arc::clone(&is_interrupted))));
-    //let reader = Arc::new(CharBufReader::new(
-          //input_stream, Arc::clone(&is_interrupted)));
+    let reader = Arc::new(Mutex::new(CharBufReader::new(input_stream)));
     let line_buffer = LineBuffer::new();
 
     let terminal = Terminal {
@@ -119,7 +129,7 @@ impl<InputStream: Read + AsRawFd + Unpin> Terminal<InputStream> {
       line_buffer: line_buffer,
       file_descriptor: fd,
       initial_termios: termios,
-      is_interrupted: is_interrupted,
+      interrupt_receiver: interrupt_receiver,
     };
 
     terminal.init_modes()?;
@@ -158,8 +168,6 @@ impl<InputStream: Read + AsRawFd + Unpin> Terminal<InputStream> {
 
 
   pub fn cleanup_modes(&self) -> Result<()> {
-    println!("de-initializing"); // DO NOT SUBMIT
-
     let termios = self.initial_termios.clone();
 
     termios::tcsetattr(self.file_descriptor,
@@ -171,23 +179,20 @@ impl<InputStream: Read + AsRawFd + Unpin> Terminal<InputStream> {
   }
 
 
-  pub fn is_exiting(&self) -> bool {
-    self.is_interrupted.load(Ordering::Relaxed)
-  }
-
-
   pub async fn handle_input(&mut self) -> Result<Input>
   {
-    let is_interrupted = Arc::clone(&self.is_interrupted);
-
     loop {
       let mut action: Option<InputAction> = None;
 
-      let string = CharBufReader::fill_buf(Arc::clone(&self.reader)).await.map_err(error::input)?;
-
-      if is_interrupted.load(Ordering::Relaxed) {
-        break;
-      }
+      let string = tokio::select! {
+        result = CharBufReader::fill_buf(Arc::clone(&self.reader)) => {
+          let string: String = result.map_err(error::input)?;
+          string
+        }
+        _ = self.interrupt_receiver.recv() => {
+          return Ok(Input::End);
+        }
+      };
 
       let mut chars = string.char_indices();
 
@@ -201,8 +206,11 @@ impl<InputStream: Read + AsRawFd + Unpin> Terminal<InputStream> {
           }
           _ => {
             self.line_buffer.insert(&c.to_string());
-            print!("{}", c);
-            io::stdout().flush();
+
+            let mut stdout = io::stdout();
+            stdout.write_all(format!("{}", c).as_bytes()).await
+                .map_err(error::internal)?;
+            stdout.flush().await.map_err(error::internal)?;
           }
         }
 
@@ -230,13 +238,8 @@ impl<InputStream: Read + AsRawFd + Unpin> Terminal<InputStream> {
     }
 
     println!("line buffer {:?}", self.line_buffer);
-    if self.is_interrupted.load(Ordering::Relaxed) {
-      println!("exiting 3");
-      Ok(Input::End)
-    } else {
-      let input = Input::String(self.line_buffer.as_string());
-      Ok(input)
-    }
+    let input = Input::String(self.line_buffer.as_string());
+    Ok(input)
   }
 }