command_stream/terminal/
buffer.rs

1//! Independent virtual terminal buffer with bounded frame notifications.
2use serde::{Deserialize, Serialize};
3use tokio::sync::broadcast;
4
5#[derive(Default)]
6pub(crate) struct TerminalResponses {
7    pending: Vec<u8>,
8}
9
10impl vt100::Callbacks for TerminalResponses {
11    fn unhandled_csi(
12        &mut self,
13        screen: &mut vt100::Screen,
14        intermediate: Option<u8>,
15        _: Option<u8>,
16        params: &[&[u16]],
17        command: char,
18    ) {
19        let parameter = params
20            .first()
21            .and_then(|value| value.first())
22            .copied()
23            .unwrap_or(0);
24        let response = match (intermediate, parameter, command) {
25            (None, 5, 'n') => "\x1b[0n".to_owned(),
26            (None, 6, 'n') => {
27                let (row, col) = screen.cursor_position();
28                format!("\x1b[{};{}R", row + 1, col + 1)
29            }
30            (None, 0, 'c') => "\x1b[?1;2c".to_owned(),
31            (None, 18, 't') => {
32                let (rows, cols) = screen.size();
33                format!("\x1b[8;{rows};{cols}t")
34            }
35            _ => return,
36        };
37        self.pending.extend_from_slice(response.as_bytes());
38    }
39}
40
41pub(crate) type TerminalParser = vt100::Parser<TerminalResponses>;
42
43pub(crate) fn terminal_parser(rows: u16, cols: u16, scrollback: usize) -> TerminalParser {
44    vt100::Parser::new_with_callbacks(rows, cols, scrollback, TerminalResponses::default())
45}
46
47pub(crate) fn take_terminal_input(parser: &mut TerminalParser) -> Vec<u8> {
48    std::mem::take(&mut parser.callbacks_mut().pending)
49}
50
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct BufferFrame {
53    pub text: String,
54    pub ansi: Vec<u8>,
55    pub rows: u16,
56    pub cols: u16,
57    pub cursor: (u16, u16),
58    pub alternate: bool,
59}
60
61/// A replayable ANSI update relative to the preceding sequence.
62/// Resize the receiving terminal before applying an update with `reset` set.
63#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
64pub struct BufferChange {
65    pub sequence: u64,
66    pub rows: u16,
67    pub cols: u16,
68    pub cursor: (u16, u16),
69    pub alternate: bool,
70    pub reset: bool,
71    pub ansi: Vec<u8>,
72}
73
74pub(crate) fn valid_dimensions(rows: u16, cols: u16) -> bool {
75    rows > 0 && cols > 0 && u32::from(rows) * u32::from(cols) <= 1_000_000
76}
77
78pub(crate) struct BufferPublisher {
79    previous: vt100::Screen,
80    sequence: u64,
81    pub frames: broadcast::Sender<BufferFrame>,
82    pub changes: broadcast::Sender<BufferChange>,
83}
84
85impl BufferPublisher {
86    pub fn new(screen: &vt100::Screen) -> Self {
87        Self {
88            previous: screen.clone(),
89            sequence: 0,
90            frames: broadcast::channel(16).0,
91            changes: broadcast::channel(16).0,
92        }
93    }
94
95    pub fn frame(screen: &vt100::Screen) -> BufferFrame {
96        let (rows, cols) = screen.size();
97        BufferFrame {
98            text: screen.contents(),
99            ansi: screen.state_formatted(),
100            rows,
101            cols,
102            cursor: screen.cursor_position(),
103            alternate: screen.alternate_screen(),
104        }
105    }
106
107    pub fn publish(&mut self, screen: &vt100::Screen) {
108        let frame = Self::frame(screen);
109        if frame == Self::frame(&self.previous) {
110            return;
111        }
112        let reset = screen.size() != self.previous.size()
113            || screen.alternate_screen() != self.previous.alternate_screen();
114        self.sequence += 1;
115        let change = BufferChange {
116            sequence: self.sequence,
117            rows: frame.rows,
118            cols: frame.cols,
119            cursor: frame.cursor,
120            alternate: frame.alternate,
121            reset,
122            ansi: if reset {
123                let mut ansi = if frame.alternate {
124                    b"\x1b[?1049h".to_vec()
125                } else {
126                    b"\x1b[?1049l".to_vec()
127                };
128                ansi.extend_from_slice(&frame.ansi);
129                ansi
130            } else {
131                screen.state_diff(&self.previous)
132            },
133        };
134        self.previous = screen.clone();
135        let _ = self.frames.send(frame);
136        let _ = self.changes.send(change);
137    }
138}
139
140pub struct TerminalBuffer {
141    parser: TerminalParser,
142    publisher: BufferPublisher,
143}
144
145impl TerminalBuffer {
146    pub fn new(rows: u16, cols: u16) -> Self {
147        assert!(valid_dimensions(rows, cols));
148        let parser = terminal_parser(rows, cols, 1000);
149        let publisher = BufferPublisher::new(parser.screen());
150        Self { parser, publisher }
151    }
152    pub fn snapshot(&self) -> BufferFrame {
153        BufferPublisher::frame(self.parser.screen())
154    }
155    pub fn text(&self, color: bool) -> String {
156        if color {
157            String::from_utf8_lossy(&self.parser.screen().contents_formatted()).into_owned()
158        } else {
159            self.parser.screen().contents()
160        }
161    }
162    pub fn subscribe(&self) -> broadcast::Receiver<BufferFrame> {
163        self.publisher.frames.subscribe()
164    }
165    /// Bounded changes only. A slow receiver gets `RecvError::Lagged`; take a
166    /// new snapshot before applying subsequent deltas in that case.
167    pub fn subscribe_changes(&self) -> broadcast::Receiver<BufferChange> {
168        self.publisher.changes.subscribe()
169    }
170    pub fn write(&mut self, data: &[u8]) {
171        self.parser.process(data);
172        self.flush();
173    }
174    /// Query responses to forward to the application or remote PTY input.
175    pub fn take_input(&mut self) -> Vec<u8> {
176        take_terminal_input(&mut self.parser)
177    }
178    pub fn resize(&mut self, rows: u16, cols: u16) {
179        assert!(valid_dimensions(rows, cols));
180        self.parser.screen_mut().set_size(rows, cols);
181        self.flush();
182    }
183    pub fn flush(&mut self) {
184        self.publisher.publish(self.parser.screen());
185    }
186}
command_stream/terminal/buffer.rs

command_stream/terminal/
buffer.rs
