/
tty.rs
262 lines (232 loc) · 8.61 KB
/
tty.rs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
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::{
source::EventSource,
sys::unix::{
file_descriptor::{tty_fd, FileDesc},
parse::parse_event,
},
InternalEvent,
};
/// 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()
}
}