termit 0.7.0

Terminal UI over crossterm
Documentation 
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

use async_channel::{bounded, Receiver, SendError, Sender};
use futures_lite::{ready, AsyncRead, StreamExt};
use log::{trace, warn};
use std::{
    io,
    task::Poll,
    thread::{self, JoinHandle},
    time::Duration,
};

pub(crate) type DefaultStdIn = AsyncReader<io::Stdin>;
pub(crate) fn async_input() -> DefaultStdIn {
    DefaultStdIn::default()
}

/// An std::io::Read to AsyncRead adapter
///
/// It runs blocking read on a separate thread
/// passing input through a channel to the consumer.
///
/// It's probably not very efficient, but helps avoiding
/// async-std / tokio dependency just for this purpose.
///
/// If you have tokio / async-std dependency anyway,
/// then use that framework's async stdin in setting up
/// the `Terminal`.
pub struct AsyncReader<R> {
    stop_sender: Sender<()>,
    buff_receiver: Receiver<Vec<u8>>,
    join_handle: Option<JoinHandle<(R, io::Result<Vec<u8>>)>>,
    buffer: Vec<u8>,
}

impl Default for AsyncReader<io::Stdin> {
    fn default() -> Self {
        Self::new(io::stdin())
    }
}

impl<R> AsyncReader<R>
where
    R: io::Read,
    R: Send + 'static,
{
    pub fn new(mut read: R) -> Self {
        let (stop_sender, stop_receiver) = bounded(1);
        let (buff_sender, buff_receiver) = bounded(1);
        let mut buf = [0; 1024];

        let join_handle = Some(thread::spawn(move || 'reading: loop {
            match stop_receiver.try_recv() {
                Ok(()) | Err(async_channel::TryRecvError::Closed) => break (read, Ok(vec![])),
                Err(async_channel::TryRecvError::Empty) => {}
            }
            break match read.read(&mut buf) {
                Ok(0) => (read, Ok(vec![])),
                Err(e) => (read, Err(e)),
                Ok(len) => {
                    let mut sendit = buf[0..len].to_vec();
                    loop {
                        if let Ok(()) = stop_receiver.try_recv() {
                            break (read, Ok(sendit));
                        }
                        match buff_sender.send_blocking(sendit) {
                            Ok(()) => {
                                continue 'reading;
                            }
                            Err(SendError(vec)) => {
                                warn!("Unable to send input");
                                sendit = vec;
                            }
                        }
                    }
                }
            };
        }));

        Self {
            stop_sender,
            buff_receiver,
            join_handle,
            buffer: vec![],
        }
    }
}

impl<R> AsyncRead for AsyncReader<R> {
    fn poll_read(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &mut [u8],
    ) -> Poll<io::Result<usize>> {
        Poll::Ready(loop {
            if !self.buffer.is_empty() {
                let len = buf.len().min(self.buffer.len());
                buf[0..len].copy_from_slice(&self.buffer[0..len]);
                if len == self.buffer.len() {
                    self.buffer.clear();
                } else {
                    self.buffer = self.buffer.split_off(len);
                }
                break Ok(len);
            }

            if let Some(buf) = ready!(self.buff_receiver.poll_next(cx)) {
                self.buffer = buf;
            } else {
                break Ok(0);
            }
        })
    }
}

impl<R> Drop for AsyncReader<R> {
    fn drop(&mut self) {
        trace!("closing reader");
        if let Some(join_handle) = self.join_handle.take() {
            self.stop_sender
                .send_blocking(())
                .unwrap_or_else(|_| warn!("could not stop reader"));
            trace!("stopped reader");

            thread::sleep(Duration::from_millis(10));

            if !join_handle.is_finished() {
                warn!("abandoned the reader");
                return;
            }

            match join_handle.join() {
                Ok((_r, Ok(unread))) if unread.is_empty() => {}
                Ok((_r, Ok(unread))) => {
                    warn!("unread data: {unread:?}")
                }
                Ok((_r, Err(e))) => {
                    warn!("reader error on drop: {e:?}")
                }
                Err(e) => warn!("could not join reader thread - {e:?}"),
            }
            trace!("closed reader");
        }
    }
}