command_stream/terminal/
buffer.rs

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

command_stream/terminal/
buffer.rs
