Repository: crossterm-rs/crossterm (language: rust)

Issue:
/dev/tty does not work on macOS with kqueue
**Describe the bug**
As described in tokio-rs/mio#1377, `kqueue` does not play well with `/dev/tty`. This is a problem for crossterm (as in #407), being unable to work properly.

**To Reproduce**
See reproducer in tokio-rs/mio#1377.

**Expected behavior**
Do not fail to poll on `/dev/tty`.

**OS**
MacOs.

**Terminal/Console**
 Any.

**Possible solutions**
The general solution is to use `select` instead of `kqueue` or `epoll` (`poll` should not work either). Changing Mio to support a `select`-based `Poll` requires major refactoring that is probably never going to happen. Given that `mio::Poll` is used in crossterm only (AFAIK) to poll `SIGWINCH` and `/dev/tty`, I suggest dropping the dependency on Mio and implement a `select`-based polling system with `libc` or `nix`.

Two major problems:
- how to poll from fd and signal on non-linux (w/o `signalfd`);
- how does it play with the waker system?

I'd be open to make a PR. Any suggestion?


Discussion / hints:
To handle signals we could use the `signal_hook` crate with pipes and `select` on `/dev/tty` and the read end of the pipe.

I've never used Rust asynchronous programming features (i.e. `async/await`) so I'm a bit lost for the `event-stream` feature.
Here is some relevant code that uses this approach:

https://github.com/wez/wezterm/blob/f12df0be9b10491bdfdead3c4fc57918b208068d/termwiz/src/terminal/unix.rs#L390-L479

https://github.com/wez/wezterm/blob/f12df0be9b10491bdfdead3c4fc57918b208068d/filedescriptor/src/unix.rs#L421-L476
That is sad to hear :(

There are 3 things at stake when speaking about rewrites:
- Crossterm uses the signal crate with the mio extension for [terminal size](https://github.com/crossterm-rs/crossterm/blob/master/src/event/source/unix.rs#L129).
- Crossterm uses MIO for polling event readiness on unix systems.
- Crossterm has a `Waker` type which throws an event and wakes up a pending `poll` operation from MIO.

Also as for async input system, only the `Waker` and `EventSource` needs to function correctly. All other async logic is not of importance in this issue (as far I know).

There are some options here. First, given that this use case of `/dev/tty` only doesnt work on macos we could limit this feature and make macos use STDOUT by default. Which might bring some sad faces along. 

Secondly, rewrite/extend the input system. The area of code I'm talking about can be found [here](https://github.com/crossterm-rs/crossterm/blob/master/src/event/source/unix.rs#L74)

It 'could' be easy to replace the system. Since the logic responsible for all input/eventing logic is hidden behind a trait that is currently implemented for [windows and unix](https://github.com/crossterm-rs/crossterm/tree/master/src/event/source). We can create a new implementation called  `MacOsEventSource` for example. If this implementation works well, according to that which is expected behavior, it could be added [here]( https://github.com/crossterm-rs/crossterm/blob/master/src/event/read.rs#L24) with a `cfg(macos)` macro. 

Steps to create a new input system 
- Create a new `EventSource` like `MacOsEventSource`
- Implement the `try_read` functionality, this function must be able to:
     - Wait for input for a certain duration
     - Return Some (`TerminalResizeEvent`, `Input`) or `None`  if the duration elapsed. 
     - Being able to wake up the read
- Add the newly created `EventSource` to the InternalEventReader` [here](https://github.com/crossterm-rs/crossterm/blob/master/src/event/read.rs#L24).  This should make crossterm use a select based approach and therefore fixing all problems for macos. 

This `select` based `EventSource` could serve as the default implementation for all Unix systems or we can limit it to MacOs (as spoken about above).

As far `select` goes, I am not that familiar with it and cant estimate how difficult it is to make it work within the `EventSource` trait. 
> There are 3 things at stake when speaking about rewrites:
> 
>     * Crossterm uses the signal crate with the mio extension for [terminal size](https://github.com/crossterm-rs/crossterm/blob/master/src/event/source/unix.rs#L129).
> 
>     * Crossterm uses MIO for polling event readiness on unix systems.
> 
>     * Crossterm has a `Waker` type which throws an event and wakes up a pending `poll` operation from MIO.

All of the above can be done w/o Mio, with just the `signal_hook` crate, `select` syscall and pipes.

> There are some options here. First, given that this use case of `/dev/tty` only doesnt work on macos we could limit this feature and make macos use STDOUT by default. Which might bring some sad faces along.

I'm guessing you meant `STDIN`. The problem there, which is one I already faced while using crossterm, is that a downstream crate may want to read, e.g., a file from standard input.

> This `select` based `EventSource` could serve as the default implementation for all Unix systems or we can limit it to MacOs (as spoken about above).

Indeed, an event source using `select` would work just fine on all UNIX systems: it's not used for polling from 100s of FDs and is the most supported syscall for polling (since the '80s). I'd be more keen to drop the dependency on Mio and rewrite the `UnixEventSource`.

Which of the two options would you prefer?

Also, I don't see tests around. How is this crate tested?


Testing is mainly done with hand :). Just press some keys and see if it works. It is very hard to write tests for the input module since terminals differ from key combinations they support and we need to physically interact with the STDOUT/STDIN/ or /dev/tty. Which is not available in a CI environment. So its tricky. 

If we can drop the MIO and are able to implement the same thing with 'select' I'm all the way in. It would make crossterm way smaller. So that's cool. Anyhow, as far I can tell it is quite easy to create a custom `EventSource` and create an implementation. If that's done we can test it and if it works we can drop it and switch to it completely. The async part should not be a deal as long the `EventSource` does its job well as described.
Great, I'll work on a PR.

Do you have plans to merge #407 or shall I open a new PR?

That would just use `tty_fd`  instead of `STDIN` in `crossterm::terminal`. E.g. here:
https://github.com/crossterm-rs/crossterm/blob/4b1c8579142da89d0189f85aefe457d643c71491/src/terminal/sys/unix.rs#L122
Nice!

This branch is waiting for the mac problem to be fixed, though that's what you going to do now. Thus I think closing the branch is better, it contains lots of debug code, and the important changes are not that big of a change. We should use the `tty_fd` function. And this function should first try `/dev/tty` but if not possible it should try `stdin` at the following places:

- https://github.com/crossterm-rs/crossterm/blob/4b1c8579142da89d0189f85aefe457d643c71491/src/terminal/sys/unix.rs#L122
- https://github.com/crossterm-rs/crossterm/blob/4b1c8579142da89d0189f85aefe457d643c71491/src/terminal/sys/unix.rs#L128



Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 56558baef..7ec03d44f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -29,6 +29,7 @@ all-features = true
 default = ["bracketed-paste"]
 bracketed-paste = []
 event-stream = ["futures-core"]
+use-dev-tty = ["filedescriptor"]
 
 #
 # Shared dependencies
@@ -56,8 +57,9 @@ crossterm_winapi = "0.9"
 #
 [target.'cfg(unix)'.dependencies]
 libc = "0.2"
-mio = { version = "0.8", features = ["os-poll"] }
 signal-hook = { version = "0.3.13" }
+filedescriptor = { version = "0.8", optional = true }
+mio = { version = "0.8", features = ["os-poll"] }
 signal-hook-mio = { version = "0.2.3", features = ["support-v0_8"] }
 
 #
diff --git a/src/event/source/unix.rs b/src/event/source/unix.rs
index 8a44d5a5d..b866d7d93 100644
--- a/src/event/source/unix.rs
+++ b/src/event/source/unix.rs
@@ -1,241 +1,11 @@
-use std::{collections::VecDeque, io, time::Duration};
+#[cfg(feature = "use-dev-tty")]
+pub(crate) mod tty;
 
-use mio::{unix::SourceFd, Events, Interest, Poll, Token};
-use signal_hook_mio::v0_8::Signals;
+#[cfg(not(feature = "use-dev-tty"))]
+pub(crate) mod mio;
 
-use crate::Result;
+#[cfg(feature = "use-dev-tty")]
+pub(crate) use self::tty::UnixInternalEventSource;
 
-#[cfg(feature = "event-stream")]
-use super::super::sys::Waker;
-use super::super::{
-    source::EventSource,
-    sys::unix::{
-        file_descriptor::{tty_fd, FileDesc},
-        parse::parse_event,
-    },
-    timeout::PollTimeout,
-    Event, InternalEvent,
-};
-
-// Tokens to identify file descriptor
-const TTY_TOKEN: Token = Token(0);
-const SIGNAL_TOKEN: Token = Token(1);
-#[cfg(feature = "event-stream")]
-const WAKE_TOKEN: Token = Token(2);
-
-// I (@zrzka) wasn't able to read more than 1_022 bytes when testing
-// reading on macOS/Linux -> we don't need bigger buffer and 1k of bytes
-// is enough.
-const TTY_BUFFER_SIZE: usize = 1_024;
-
-pub(crate) struct UnixInternalEventSource {
-    poll: Poll,
-    events: Events,
-    parser: Parser,
-    tty_buffer: [u8; TTY_BUFFER_SIZE],
-    tty_fd: FileDesc,
-    signals: Signals,
-    #[cfg(feature = "event-stream")]
-    waker: Waker,
-}
-
-impl UnixInternalEventSource {
-    pub fn new() -> Result<Self> {
-        UnixInternalEventSource::from_file_descriptor(tty_fd()?)
-    }
-
-    pub(crate) fn from_file_descriptor(input_fd: FileDesc) -> Result<Self> {
-        let poll = Poll::new()?;
-        let registry = poll.registry();
-
-        let tty_raw_fd = input_fd.raw_fd();
-        let mut tty_ev = SourceFd(&tty_raw_fd);
-        registry.register(&mut tty_ev, TTY_TOKEN, Interest::READABLE)?;
-
-        let mut signals = Signals::new(&[signal_hook::consts::SIGWINCH])?;
-        registry.register(&mut signals, SIGNAL_TOKEN, Interest::READABLE)?;
-
-        #[cfg(feature = "event-stream")]
-        let waker = Waker::new(registry, WAKE_TOKEN)?;
-
-        Ok(UnixInternalEventSource {
-            poll,
-            events: Events::with_capacity(3),
-            parser: Parser::default(),
-            tty_buffer: [0u8; TTY_BUFFER_SIZE],
-            tty_fd: input_fd,
-            signals,
-            #[cfg(feature = "event-stream")]
-            waker,
-        })
-    }
-}
-
-impl EventSource for UnixInternalEventSource {
-    fn try_read(&mut self, timeout: Option<Duration>) -> Result<Option<InternalEvent>> {
-        if let Some(event) = self.parser.next() {
-            return Ok(Some(event));
-        }
-
-        let timeout = PollTimeout::new(timeout);
-
-        loop {
-            if let Err(e) = self.poll.poll(&mut self.events, timeout.leftover()) {
-                // Mio will throw an interrupted error in case of cursor position retrieval. We need to retry until it succeeds.
-                // Previous versions of Mio (< 0.7) would automatically retry the poll call if it was interrupted (if EINTR was returned).
-                // https://docs.rs/mio/0.7.0/mio/struct.Poll.html#notes
-                if e.kind() == io::ErrorKind::Interrupted {
-                    continue;
-                } else {
-                    return Err(e);
-                }
-            };
-
-            if self.events.is_empty() {
-                // No readiness events = timeout
-                return Ok(None);
-            }
-
-            for token in self.events.iter().map(|x| x.token()) {
-                match token {
-                    TTY_TOKEN => {
-                        loop {
-                            match self.tty_fd.read(&mut self.tty_buffer, TTY_BUFFER_SIZE) {
-                                Ok(read_count) => {
-                                    if read_count > 0 {
-                                        self.parser.advance(
-                                            &self.tty_buffer[..read_count],
-                                            read_count == TTY_BUFFER_SIZE,
-                                        );
-                                    }
-                                }
-                                Err(e) => {
-                                    // No more data to read at the moment. We will receive another event
-                                    if e.kind() == io::ErrorKind::WouldBlock {
-                                        break;
-                                    }
-                                    // once more data is available to read.
-                                    else if e.kind() == io::ErrorKind::Interrupted {
-                                        continue;
-                                    }
-                                }
-                            };
-
-                            if let Some(event) = self.parser.next() {
-                                return Ok(Some(event));
-                            }
-                        }
-                    }
-                    SIGNAL_TOKEN => {
-                        for signal in self.signals.pending() {
-                            match signal {
-                                signal_hook::consts::SIGWINCH => {
-                                    // TODO Should we remove tput?
-                                    //
-                                    // This can take a really long time, because terminal::size can
-                                    // launch new process (tput) and then it parses its output. It's
-                                    // not a really long time from the absolute time point of view, but
-                                    // it's a really long time from the mio, async-std/tokio executor, ...
-                                    // point of view.
-                                    let new_size = crate::terminal::size()?;
-                                    return Ok(Some(InternalEvent::Event(Event::Resize(
-                                        new_size.0, new_size.1,
-                                    ))));
-                                }
-                                _ => unreachable!("Synchronize signal registration & handling"),
-                            };
-                        }
-                    }
-                    #[cfg(feature = "event-stream")]
-                    WAKE_TOKEN => {
-                        return Err(std::io::Error::new(
-                            std::io::ErrorKind::Interrupted,
-                            "Poll operation was woken up by `Waker::wake`",
-                        ));
-                    }
-                    _ => unreachable!("Synchronize Evented handle registration & token handling"),
-                }
-            }
-
-            // Processing above can take some time, check if timeout expired
-            if timeout.elapsed() {
-                return Ok(None);
-            }
-        }
-    }
-
-    #[cfg(feature = "event-stream")]
-    fn waker(&self) -> Waker {
-        self.waker.clone()
-    }
-}
-
-//
-// Following `Parser` structure exists for two reasons:
-//
-//  * mimic anes Parser interface
-//  * move the advancing, parsing, ... stuff out of the `try_read` method
-//
-#[derive(Debug)]
-struct Parser {
-    buffer: Vec<u8>,
-    internal_events: VecDeque<InternalEvent>,
-}
-
-impl Default for Parser {
-    fn default() -> Self {
-        Parser {
-            // This buffer is used for -> 1 <- ANSI escape sequence. Are we
-            // aware of any ANSI escape sequence that is bigger? Can we make
-            // it smaller?
-            //
-            // Probably not worth spending more time on this as "there's a plan"
-            // to use the anes crate parser.
-            buffer: Vec::with_capacity(256),
-            // TTY_BUFFER_SIZE is 1_024 bytes. How many ANSI escape sequences can
-            // fit? What is an average sequence length? Let's guess here
-            // and say that the average ANSI escape sequence length is 8 bytes. Thus
-            // the buffer size should be 1024/8=128 to avoid additional allocations
-            // when processing large amounts of data.
-            //
-            // There's no need to make it bigger, because when you look at the `try_read`
-            // method implementation, all events are consumed before the next TTY_BUFFER
-            // is processed -> events pushed.
-            internal_events: VecDeque::with_capacity(128),
-        }
-    }
-}
-
-impl Parser {
-    fn advance(&mut self, buffer: &[u8], more: bool) {
-        for (idx, byte) in buffer.iter().enumerate() {
-            let more = idx + 1 < buffer.len() || more;
-
-            self.buffer.push(*byte);
-
-            match parse_event(&self.buffer, more) {
-                Ok(Some(ie)) => {
-                    self.internal_events.push_back(ie);
-                    self.buffer.clear();
-                }
-                Ok(None) => {
-                    // Event can't be parsed, because we don't have enough bytes for
-                    // the current sequence. Keep the buffer and process next bytes.
-                }
-                Err(_) => {
-                    // Event can't be parsed (not enough parameters, parameter is not a number, ...).
-                    // Clear the buffer and continue with another sequence.
-                    self.buffer.clear();
-                }
-            }
-        }
-    }
-}
-
-impl Iterator for Parser {
-    type Item = InternalEvent;
-
-    fn next(&mut self) -> Option<Self::Item> {
-        self.internal_events.pop_front()
-    }
-}
+#[cfg(not(feature = "use-dev-tty"))]
+pub(crate) use self::mio::UnixInternalEventSource;
\ No newline at end of file
diff --git a/src/event/source/unix/mio.rs b/src/event/source/unix/mio.rs
new file mode 100644
index 000000000..67bda3735
--- /dev/null
+++ b/src/event/source/unix/mio.rs
@@ -0,0 +1,241 @@
+use std::{collections::VecDeque, io, time::Duration};
+
+use mio::{unix::SourceFd, Events, Interest, Poll, Token};
+use signal_hook_mio::v0_8::Signals;
+
+use crate::Result;
+
+#[cfg(feature = "event-stream")]
+use crate::event::sys::Waker;
+use crate::event::{
+    source::EventSource,
+    sys::unix::{
+        file_descriptor::{tty_fd, FileDesc},
+        parse::parse_event,
+    },
+    timeout::PollTimeout,
+    Event, InternalEvent,
+};
+
+// Tokens to identify file descriptor
+const TTY_TOKEN: Token = Token(0);
+const SIGNAL_TOKEN: Token = Token(1);
+#[cfg(feature = "event-stream")]
+const WAKE_TOKEN: Token = Token(2);
+
+// I (@zrzka) wasn't able to read more than 1_022 bytes when testing
+// reading on macOS/Linux -> we don't need bigger buffer and 1k of bytes
+// is enough.
+const TTY_BUFFER_SIZE: usize = 1_024;
+
+pub(crate) struct UnixInternalEventSource {
+    poll: Poll,
+    events: Events,
+    parser: Parser,
+    tty_buffer: [u8; TTY_BUFFER_SIZE],
+    tty_fd: FileDesc,
+    signals: Signals,
+    #[cfg(feature = "event-stream")]
+    waker: Waker,
+}
+
+impl UnixInternalEventSource {
+    pub fn new() -> Result<Self> {
+        UnixInternalEventSource::from_file_descriptor(tty_fd()?)
+    }
+
+    pub(crate) fn from_file_descriptor(input_fd: FileDesc) -> Result<Self> {
+        let poll = Poll::new()?;
+        let registry = poll.registry();
+
+        let tty_raw_fd = input_fd.raw_fd();
+        let mut tty_ev = SourceFd(&tty_raw_fd);
+        registry.register(&mut tty_ev, TTY_TOKEN, Interest::READABLE)?;
+
+        let mut signals = Signals::new([signal_hook::consts::SIGWINCH])?;
+        registry.register(&mut signals, SIGNAL_TOKEN, Interest::READABLE)?;
+
+        #[cfg(feature = "event-stream")]
+        let waker = Waker::new(registry, WAKE_TOKEN)?;
+
+        Ok(UnixInternalEventSource {
+            poll,
+            events: Events::with_capacity(3),
+            parser: Parser::default(),
+            tty_buffer: [0u8; TTY_BUFFER_SIZE],
+            tty_fd: input_fd,
+            signals,
+            #[cfg(feature = "event-stream")]
+            waker,
+        })
+    }
+}
+
+impl EventSource for UnixInternalEventSource {
+    fn try_read(&mut self, timeout: Option<Duration>) -> Result<Option<InternalEvent>> {
+        if let Some(event) = self.parser.next() {
+            return Ok(Some(event));
+        }
+
+        let timeout = PollTimeout::new(timeout);
+
+        loop {
+            if let Err(e) = self.poll.poll(&mut self.events, timeout.leftover()) {
+                // Mio will throw an interrupted error in case of cursor position retrieval. We need to retry until it succeeds.
+                // Previous versions of Mio (< 0.7) would automatically retry the poll call if it was interrupted (if EINTR was returned).
+                // https://docs.rs/mio/0.7.0/mio/struct.Poll.html#notes
+                if e.kind() == io::ErrorKind::Interrupted {
+                    continue;
+                } else {
+                    return Err(e);
+                }
+            };
+
+            if self.events.is_empty() {
+                // No readiness events = timeout
+                return Ok(None);
+            }
+
+            for token in self.events.iter().map(|x| x.token()) {
+                match token {
+                    TTY_TOKEN => {
+                        loop {
+                            match self.tty_fd.read(&mut self.tty_buffer, TTY_BUFFER_SIZE) {
+                                Ok(read_count) => {
+                                    if read_count > 0 {
+                                        self.parser.advance(
+                                            &self.tty_buffer[..read_count],
+                                            read_count == TTY_BUFFER_SIZE,
+                                        );
+                                    }
+                                }
+                                Err(e) => {
+                                    // No more data to read at the moment. We will receive another event
+                                    if e.kind() == io::ErrorKind::WouldBlock {
+                                        break;
+                                    }
+                                    // once more data is available to read.
+                                    else if e.kind() == io::ErrorKind::Interrupted {
+                                        continue;
+                                    }
+                                }
+                            };
+
+                            if let Some(event) = self.parser.next() {
+                                return Ok(Some(event));
+                            }
+                        }
+                    }
+                    SIGNAL_TOKEN => {
+                        for signal in self.signals.pending() {
+                            match signal {
+                                signal_hook::consts::SIGWINCH => {
+                                    // TODO Should we remove tput?
+                                    //
+                                    // This can take a really long time, because terminal::size can
+                                    // launch new process (tput) and then it parses its output. It's
+                                    // not a really long time from the absolute time point of view, but
+                                    // it's a really long time from the mio, async-std/tokio executor, ...
+                                    // point of view.
+                                    let new_size = crate::terminal::size()?;
+                                    return Ok(Some(InternalEvent::Event(Event::Resize(
+                                        new_size.0, new_size.1,
+                                    ))));
+                                }
+                                _ => unreachable!("Synchronize signal registration & handling"),
+                            };
+                        }
+                    }
+                    #[cfg(feature = "event-stream")]
+                    WAKE_TOKEN => {
+                        return Err(std::io::Error::new(
+                            std::io::ErrorKind::Interrupted,
+                            "Poll operation was woken up by `Waker::wake`",
+                        ));
+                    }
+                    _ => unreachable!("Synchronize Evented handle registration & token handling"),
+                }
+            }
+
+            // Processing above can take some time, check if timeout expired
+            if timeout.elapsed() {
+                return Ok(None);
+            }
+        }
+    }
+
+    #[cfg(feature = "event-stream")]
+    fn waker(&self) -> Waker {
+        self.waker.clone()
+    }
+}
+
+//
+// Following `Parser` structure exists for two reasons:
+//
+//  * mimic anes Parser interface
+//  * move the advancing, parsing, ... stuff out of the `try_read` method
+//
+#[derive(Debug)]
+struct Parser {
+    buffer: Vec<u8>,
+    internal_events: VecDeque<InternalEvent>,
+}
+
+impl Default for Parser {
+    fn default() -> Self {
+        Parser {
+            // This buffer is used for -> 1 <- ANSI escape sequence. Are we
+            // aware of any ANSI escape sequence that is bigger? Can we make
+            // it smaller?
+            //
+            // Probably not worth spending more time on this as "there's a plan"
+            // to use the anes crate parser.
+            buffer: Vec::with_capacity(256),
+            // TTY_BUFFER_SIZE is 1_024 bytes. How many ANSI escape sequences can
+            // fit? What is an average sequence length? Let's guess here
+            // and say that the average ANSI escape sequence length is 8 bytes. Thus
+            // the buffer size should be 1024/8=128 to avoid additional allocations
+            // when processing large amounts of data.
+            //
+            // There's no need to make it bigger, because when you look at the `try_read`
+            // method implementation, all events are consumed before the next TTY_BUFFER
+            // is processed -> events pushed.
+            internal_events: VecDeque::with_capacity(128),
+        }
+    }
+}
+
+impl Parser {
+    fn advance(&mut self, buffer: &[u8], more: bool) {
+        for (idx, byte) in buffer.iter().enumerate() {
+            let more = idx + 1 < buffer.len() || more;
+
+            self.buffer.push(*byte);
+
+            match parse_event(&self.buffer, more) {
+                Ok(Some(ie)) => {
+                    self.internal_events.push_back(ie);
+                    self.buffer.clear();
+                }
+                Ok(None) => {
+                    // Event can't be parsed, because we don't have enough bytes for
+                    // the current sequence. Keep the buffer and process next bytes.
+                }
+                Err(_) => {
+                    // Event can't be parsed (not enough parameters, parameter is not a number, ...).
+                    // Clear the buffer and continue with another sequence.
+                    self.buffer.clear();
+                }
+            }
+        }
+    }
+}
+
+impl Iterator for Parser {
+    type Item = InternalEvent;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.internal_events.pop_front()
+    }
+}
diff --git a/src/event/source/unix/tty.rs b/src/event/source/unix/tty.rs
new file mode 100644
index 000000000..a4ab1f24f
--- /dev/null
+++ b/src/event/source/unix/tty.rs
@@ -0,0 +1,264 @@
+use std::os::unix::prelude::AsRawFd;
+use std::{collections::VecDeque, io, os::unix::net::UnixStream, time::Duration};
+
+use signal_hook::low_level::pipe;
+
+use crate::event::timeout::PollTimeout;
+use crate::event::Event;
+use crate::Result;
+
+use filedescriptor::{poll, pollfd, POLLIN};
+
+#[cfg(feature = "event-stream")]
+use crate::event::sys::Waker;
+
+use crate::{
+    event::{
+        sys::unix::{
+            file_descriptor::{tty_fd, FileDesc},
+            parse::parse_event,
+        },
+        InternalEvent,
+        source::EventSource,
+    },
+};
+
+/// Holds a prototypical Waker and a receiver we can wait on when doing select().
+#[cfg(feature = "event-stream")]
+struct WakePipe {
+    receiver: UnixStream,
+    waker: Waker,
+}
+
+#[cfg(feature = "event-stream")]
+impl WakePipe {
+    fn new() -> Result<Self> {
+        let (receiver, sender) = nonblocking_unix_pair()?;
+        Ok(WakePipe {
+            receiver,
+            waker: Waker::new(sender),
+        })
+    }
+}
+
+// I (@zrzka) wasn't able to read more than 1_022 bytes when testing
+// reading on macOS/Linux -> we don't need bigger buffer and 1k of bytes
+// is enough.
+const TTY_BUFFER_SIZE: usize = 1_024;
+
+pub(crate) struct UnixInternalEventSource {
+    parser: Parser,
+    tty_buffer: [u8; TTY_BUFFER_SIZE],
+    tty: FileDesc,
+    winch_signal_receiver: UnixStream,
+    #[cfg(feature = "event-stream")]
+    wake_pipe: WakePipe,
+}
+
+fn nonblocking_unix_pair() -> Result<(UnixStream, UnixStream)> {
+    let (receiver, sender) = UnixStream::pair()?;
+    receiver.set_nonblocking(true)?;
+    sender.set_nonblocking(true)?;
+    Ok((receiver, sender))
+}
+
+impl UnixInternalEventSource {
+    pub fn new() -> Result<Self> {
+        UnixInternalEventSource::from_file_descriptor(tty_fd()?)
+    }
+
+    pub(crate) fn from_file_descriptor(input_fd: FileDesc) -> Result<Self> {
+        Ok(UnixInternalEventSource {
+            parser: Parser::default(),
+            tty_buffer: [0u8; TTY_BUFFER_SIZE],
+            tty: input_fd,
+            winch_signal_receiver: {
+                let (receiver, sender) = nonblocking_unix_pair()?;
+                // Unregistering is unnecessary because EventSource is a singleton
+                pipe::register(libc::SIGWINCH, sender)?;
+                receiver
+            },
+            #[cfg(feature = "event-stream")]
+            wake_pipe: WakePipe::new()?,
+        })
+    }
+}
+
+/// read_complete reads from a non-blocking file descriptor
+/// until the buffer is full or it would block.
+///
+/// Similar to `std::io::Read::read_to_end`, except this function
+/// only fills the given buffer and does not read beyond that.
+fn read_complete(fd: &FileDesc, buf: &mut [u8]) -> Result<usize> {
+    loop {
+        match fd.read(buf, buf.len()) {
+            Ok(x) => return Ok(x),
+            Err(e) => match e.kind() {
+                io::ErrorKind::WouldBlock => return Ok(0),
+                io::ErrorKind::Interrupted => continue,
+                _ => return Err(e),
+            },
+        }
+    }
+}
+
+impl EventSource for UnixInternalEventSource {
+    fn try_read(&mut self, timeout: Option<Duration>) -> Result<Option<InternalEvent>> {
+        let timeout = PollTimeout::new(timeout);
+
+        fn make_pollfd<F: AsRawFd>(fd: &F) -> pollfd {
+            pollfd {
+                fd: fd.as_raw_fd(),
+                events: POLLIN,
+                revents: 0,
+            }
+        }
+
+        #[cfg(not(feature = "event-stream"))]
+        let mut fds = [
+            make_pollfd(&self.tty),
+            make_pollfd(&self.winch_signal_receiver),
+        ];
+
+        #[cfg(feature = "event-stream")]
+        let mut fds = [
+            make_pollfd(&self.tty),
+            make_pollfd(&self.winch_signal_receiver),
+            make_pollfd(&self.wake_pipe.receiver),
+        ];
+
+        while timeout.leftover().map_or(true, |t| !t.is_zero()) {
+            // check if there are buffered events from the last read
+            if let Some(event) = self.parser.next() {
+                return Ok(Some(event));
+            }
+            match poll(&mut fds, timeout.leftover()) {
+                Err(filedescriptor::Error::Io(e)) => return Err(e),
+                res => res.expect("polling tty"),
+            };
+            if fds[0].revents & POLLIN != 0 {
+                loop {
+                    let read_count = read_complete(&self.tty, &mut self.tty_buffer)?;
+                    if read_count > 0 {
+                        self.parser.advance(
+                            &self.tty_buffer[..read_count],
+                            read_count == TTY_BUFFER_SIZE,
+                        );
+                    }
+
+                    if let Some(event) = self.parser.next() {
+                        return Ok(Some(event));
+                    }
+
+                    if read_count == 0 {
+                        break;
+                    }
+                }
+            }
+            if fds[1].revents & POLLIN != 0 {
+                let fd = FileDesc::new(self.winch_signal_receiver.as_raw_fd(), false);
+                // drain the pipe
+                while read_complete(&fd, &mut [0; 1024])? != 0 {}
+                // TODO Should we remove tput?
+                //
+                // This can take a really long time, because terminal::size can
+                // launch new process (tput) and then it parses its output. It's
+                // not a really long time from the absolute time point of view, but
+                // it's a really long time from the mio, async-std/tokio executor, ...
+                // point of view.
+                let new_size = crate::terminal::size()?;
+                return Ok(Some(InternalEvent::Event(Event::Resize(
+                    new_size.0, new_size.1,
+                ))));
+            }
+
+            #[cfg(feature = "event-stream")]
+            if fds[2].revents & POLLIN != 0 {
+                let fd = FileDesc::new(self.wake_pipe.receiver.as_raw_fd(), false);
+                // drain the pipe
+                while read_complete(&fd, &mut [0; 1024])? != 0 {}
+
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::Interrupted,
+                    "Poll operation was woken up by `Waker::wake`",
+                ));
+            }
+        }
+        Ok(None)
+    }
+
+    #[cfg(feature = "event-stream")]
+    fn waker(&self) -> Waker {
+        self.wake_pipe.waker.clone()
+    }
+}
+
+//
+// Following `Parser` structure exists for two reasons:
+//
+//  * mimic anes Parser interface
+//  * move the advancing, parsing, ... stuff out of the `try_read` method
+//
+#[derive(Debug)]
+struct Parser {
+    buffer: Vec<u8>,
+    internal_events: VecDeque<InternalEvent>,
+}
+
+impl Default for Parser {
+    fn default() -> Self {
+        Parser {
+            // This buffer is used for -> 1 <- ANSI escape sequence. Are we
+            // aware of any ANSI escape sequence that is bigger? Can we make
+            // it smaller?
+            //
+            // Probably not worth spending more time on this as "there's a plan"
+            // to use the anes crate parser.
+            buffer: Vec::with_capacity(256),
+            // TTY_BUFFER_SIZE is 1_024 bytes. How many ANSI escape sequences can
+            // fit? What is an average sequence length? Let's guess here
+            // and say that the average ANSI escape sequence length is 8 bytes. Thus
+            // the buffer size should be 1024/8=128 to avoid additional allocations
+            // when processing large amounts of data.
+            //
+            // There's no need to make it bigger, because when you look at the `try_read`
+            // method implementation, all events are consumed before the next TTY_BUFFER
+            // is processed -> events pushed.
+            internal_events: VecDeque::with_capacity(128),
+        }
+    }
+}
+
+impl Parser {
+    fn advance(&mut self, buffer: &[u8], more: bool) {
+        for (idx, byte) in buffer.iter().enumerate() {
+            let more = idx + 1 < buffer.len() || more;
+
+            self.buffer.push(*byte);
+
+            match parse_event(&self.buffer, more) {
+                Ok(Some(ie)) => {
+                    self.internal_events.push_back(ie);
+                    self.buffer.clear();
+                }
+                Ok(None) => {
+                    // Event can't be parsed, because we don't have enough bytes for
+                    // the current sequence. Keep the buffer and process next bytes.
+                }
+                Err(_) => {
+                    // Event can't be parsed (not enough parameters, parameter is not a number, ...).
+                    // Clear the buffer and continue with another sequence.
+                    self.buffer.clear();
+                }
+            }
+        }
+    }
+}
+
+impl Iterator for Parser {
+    type Item = InternalEvent;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.internal_events.pop_front()
+    }
+}
diff --git a/src/event/sys/unix/file_descriptor.rs b/src/event/sys/unix/file_descriptor.rs
index 8b481766b..268d57ebf 100644
--- a/src/event/sys/unix/file_descriptor.rs
+++ b/src/event/sys/unix/file_descriptor.rs
@@ -1,6 +1,9 @@
 use std::{
     fs, io,
-    os::unix::io::{IntoRawFd, RawFd},
+    os::unix::{
+        io::{IntoRawFd, RawFd},
+        prelude::AsRawFd,
+    },
 };
 
 use libc::size_t;
@@ -34,7 +37,7 @@ impl FileDesc {
                 self.fd,
                 buffer.as_mut_ptr() as *mut libc::c_void,
                 size as size_t,
-            ) as isize
+            ) 
         };
 
         if result < 0 {
@@ -63,6 +66,12 @@ impl Drop for FileDesc {
     }
 }
 
+impl AsRawFd for FileDesc {
+    fn as_raw_fd(&self) -> RawFd {
+        self.raw_fd()
+    }
+}
+
 /// Creates a file descriptor pointing to the standard input or `/dev/tty`.
 pub fn tty_fd() -> Result<FileDesc> {
     let (fd, close_on_drop) = if unsafe { libc::isatty(libc::STDIN_FILENO) == 1 } {
diff --git a/src/event/sys/unix/parse.rs b/src/event/sys/unix/parse.rs
index 0cfbd1a5e..5d0aa1716 100644
--- a/src/event/sys/unix/parse.rs
+++ b/src/event/sys/unix/parse.rs
@@ -104,11 +104,11 @@ pub(crate) fn parse_event(buffer: &[u8], input_available: bool) -> Result<Option
             KeyCode::Backspace.into(),
         )))),
         c @ b'\x01'..=b'\x1A' => Ok(Some(InternalEvent::Event(Event::Key(KeyEvent::new(
-            KeyCode::Char((c as u8 - 0x1 + b'a') as char),
+            KeyCode::Char((c - 0x1 + b'a') as char),
             KeyModifiers::CONTROL,
         ))))),
         c @ b'\x1C'..=b'\x1F' => Ok(Some(InternalEvent::Event(Event::Key(KeyEvent::new(
-            KeyCode::Char((c as u8 - 0x1C + b'4') as char),
+            KeyCode::Char((c - 0x1C + b'4') as char),
             KeyModifiers::CONTROL,
         ))))),
         b'\0' => Ok(Some(InternalEvent::Event(Event::Key(KeyEvent::new(
diff --git a/src/event/sys/unix/waker.rs b/src/event/sys/unix/waker.rs
index 4509e9ba3..8ac63676c 100644
--- a/src/event/sys/unix/waker.rs
+++ b/src/event/sys/unix/waker.rs
@@ -1,36 +1,11 @@
-use std::sync::{Arc, Mutex};
+#[cfg(feature = "use-dev-tty")]
+pub(crate) mod tty;
 
-use mio::{Registry, Token};
+#[cfg(not(feature = "use-dev-tty"))]
+pub(crate) mod mio;
 
-use crate::Result;
+#[cfg(feature = "use-dev-tty")]
+pub(crate) use self::tty::Waker;
 
-/// Allows to wake up the `mio::Poll::poll()` method.
-/// This type wraps `mio::Waker`, for more information see its documentation.
-#[derive(Clone, Debug)]
-pub(crate) struct Waker {
-    inner: Arc<Mutex<mio::Waker>>,
-}
-
-impl Waker {
-    /// Create a new `Waker`.
-    pub(crate) fn new(registry: &Registry, waker_token: Token) -> Result<Self> {
-        Ok(Self {
-            inner: Arc::new(Mutex::new(mio::Waker::new(registry, waker_token)?)),
-        })
-    }
-
-    /// Wake up the [`Poll`] associated with this `Waker`.
-    ///
-    /// Readiness is set to `Ready::readable()`.
-    pub(crate) fn wake(&self) -> Result<()> {
-        self.inner.lock().unwrap().wake()
-    }
-
-    /// Resets the state so the same waker can be reused.
-    ///
-    /// This function is not impl
-    #[allow(dead_code, clippy::clippy::unnecessary_wraps)]
-    pub(crate) fn reset(&self) -> Result<()> {
-        Ok(())
-    }
-}
+#[cfg(not(feature = "use-dev-tty"))]
+pub(crate) use self::mio::Waker;
\ No newline at end of file
diff --git a/src/event/sys/unix/waker/mio.rs b/src/event/sys/unix/waker/mio.rs
new file mode 100644
index 000000000..891f006ac
--- /dev/null
+++ b/src/event/sys/unix/waker/mio.rs
@@ -0,0 +1,36 @@
+use std::sync::{Arc, Mutex};
+
+use mio::{Registry, Token};
+
+use crate::Result;
+
+/// Allows to wake up the `mio::Poll::poll()` method.
+/// This type wraps `mio::Waker`, for more information see its documentation.
+#[derive(Clone, Debug)]
+pub(crate) struct Waker {
+    inner: Arc<Mutex<mio::Waker>>,
+}
+
+impl Waker {
+    /// Create a new `Waker`.
+    pub(crate) fn new(registry: &Registry, waker_token: Token) -> Result<Self> {
+        Ok(Self {
+            inner: Arc::new(Mutex::new(mio::Waker::new(registry, waker_token)?)),
+        })
+    }
+
+    /// Wake up the [`Poll`] associated with this `Waker`.
+    ///
+    /// Readiness is set to `Ready::readable()`.
+    pub(crate) fn wake(&self) -> Result<()> {
+        self.inner.lock().unwrap().wake()
+    }
+
+    /// Resets the state so the same waker can be reused.
+    ///
+    /// This function is not impl
+    #[allow(dead_code, clippy::clippy::unnecessary_wraps)]
+    pub(crate) fn reset(&self) -> Result<()> {
+        Ok(())
+    }
+}
\ No newline at end of file
diff --git a/src/event/sys/unix/waker/tty.rs b/src/event/sys/unix/waker/tty.rs
new file mode 100644
index 000000000..47868f129
--- /dev/null
+++ b/src/event/sys/unix/waker/tty.rs
@@ -0,0 +1,38 @@
+use std::{
+    io::Write,
+    os::unix::net::UnixStream,
+    sync::{Arc, Mutex},
+};
+
+use crate::Result;
+
+/// Allows to wake up the EventSource::try_read() method.
+#[derive(Clone, Debug)]
+pub(crate) struct Waker {
+    inner: Arc<Mutex<UnixStream>>,
+}
+
+impl Waker {
+    /// Create a new `Waker`.
+    pub(crate) fn new(writer: UnixStream) -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(writer)),
+        }
+    }
+
+    /// Wake up the [`Poll`] associated with this `Waker`.
+    ///
+    /// Readiness is set to `Ready::readable()`.
+    pub(crate) fn wake(&self) -> Result<()> {
+        self.inner.lock().unwrap().write(&[0])?;
+        Ok(())
+    }
+
+    /// Resets the state so the same waker can be reused.
+    ///
+    /// This function is not impl
+    #[allow(dead_code, clippy::clippy::unnecessary_wraps)]
+    pub(crate) fn reset(&self) -> Result<()> {
+        Ok(())
+    }
+}


Test patch: