mirror of
https://github.com/l0ng-ai/tty7.git
synced 2026-11-06 00:15:06 +00:00
fix(mobile): say when typing doesn't reach the pane
The gateway's input thread gave up silently on a failed send_input, and the app's input task did the same on a failed write, so the phone kept showing a live pane while keystrokes went nowhere. Both now report the failure as a pane error. Keys after it are dropped rather than ending the stream, which the phone would read as the pane closing.
This commit is contained in:
@@ -349,12 +349,16 @@ async fn pane_stream(
|
||||
backend: Arc<dyn Backend>,
|
||||
) -> io::Result<()> {
|
||||
let (down_tx, mut down) = mpsc::channel::<Down>(PANE_BACKLOG);
|
||||
// Weak, so the stream still ends when the pane's feed does.
|
||||
let refused = down_tx.downgrade();
|
||||
std::thread::Builder::new()
|
||||
.name(format!("gateway-pane-{pane_id}"))
|
||||
.spawn(move || read_pane(feed, down_tx))?;
|
||||
|
||||
// Keystrokes go through one thread so they reach the pane in the order
|
||||
// they were typed; each `send_input` is its own daemon round trip.
|
||||
// they were typed; each `send_input` is its own daemon round trip. One
|
||||
// that fails is said to the phone, which would otherwise go on showing a
|
||||
// live pane while what it types goes nowhere.
|
||||
let (input_tx, input_rx) = std_mpsc::channel::<Vec<u8>>();
|
||||
std::thread::Builder::new()
|
||||
.name(format!("gateway-input-{pane_id}"))
|
||||
@@ -362,6 +366,10 @@ async fn pane_stream(
|
||||
for bytes in input_rx {
|
||||
if let Err(e) = backend.send_input(machine.as_deref(), pane_id, &bytes) {
|
||||
log::debug!("mobile gateway: input to pane {pane_id}: {e}");
|
||||
let message = format!("typing didn't reach the pane: {e}");
|
||||
if let Some(down) = refused.upgrade() {
|
||||
let _ = down.blocking_send(Down::Event(PaneEvent::Error { message }));
|
||||
}
|
||||
return;
|
||||
}
|
||||
}
|
||||
@@ -370,10 +378,11 @@ async fn pane_stream(
|
||||
loop {
|
||||
tokio::select! {
|
||||
frame = read_frame(&mut recv) => match frame? {
|
||||
// Once typing has failed, the phone has been told; later
|
||||
// keys are dropped rather than ending the stream, which would
|
||||
// read as the pane closing.
|
||||
Some(Frame::Bytes(bytes)) => {
|
||||
if input_tx.send(bytes).is_err() {
|
||||
return Ok(());
|
||||
}
|
||||
let _ = input_tx.send(bytes);
|
||||
}
|
||||
// No upstream messages yet; one from a newer app is ignored
|
||||
// rather than fatal, so the terminal keeps working.
|
||||
|
||||
@@ -111,6 +111,9 @@ impl Backend for FakeMachine {
|
||||
}
|
||||
|
||||
fn send_input(&self, _machine: Option<&str>, _pane_id: u64, bytes: &[u8]) -> io::Result<()> {
|
||||
if bytes == b"refuse" {
|
||||
return Err(io::Error::other("the daemon hung up"));
|
||||
}
|
||||
let typed = self.typed.lock().unwrap();
|
||||
typed
|
||||
.as_ref()
|
||||
@@ -296,6 +299,32 @@ async fn a_paired_phone_reads_the_tree_and_drives_a_pane() {
|
||||
assert!(err.to_string().contains("no such pane 99"), "{err}");
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn typing_that_fails_is_reported_not_dropped() {
|
||||
let rig = Rig::new().await;
|
||||
let session = within(rig.paired()).await;
|
||||
let (mut keys, mut screen) = within(session.pane(None, 1)).await.unwrap();
|
||||
within(screen.next()).await.unwrap(); // size
|
||||
within(screen.next()).await.unwrap(); // prompt
|
||||
|
||||
keys.input(b"refuse").await.unwrap();
|
||||
let Some(PaneItem::Event(PaneEvent::Error { message })) = within(screen.next()).await.unwrap()
|
||||
else {
|
||||
panic!("expected the failure to be reported");
|
||||
};
|
||||
assert!(message.contains("the daemon hung up"), "{message}");
|
||||
|
||||
// Keys after it go nowhere, but the stream stays up: its end would say
|
||||
// the pane closed, which it did not.
|
||||
keys.input(b"ls\r").await.unwrap();
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_millis(300), screen.next())
|
||||
.await
|
||||
.is_err(),
|
||||
"nothing more arrives"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn a_revoked_phone_is_cut_off_at_its_next_stream() {
|
||||
let rig = Rig::new().await;
|
||||
|
||||
@@ -254,9 +254,19 @@ async fn pane_open(
|
||||
let (input_tx, mut input_rx) = mpsc::unbounded_channel::<Vec<u8>>();
|
||||
state.panes.lock().await.insert(handle, input_tx);
|
||||
|
||||
// A write that fails is said on the pane's channel, and the handle goes,
|
||||
// so the next key is refused too rather than queued behind a dead stream.
|
||||
let refused = on_output.clone();
|
||||
let app_state = state.inner().clone();
|
||||
tokio::spawn(async move {
|
||||
while let Some(bytes) = input_rx.recv().await {
|
||||
if keys.input(&bytes).await.is_err() {
|
||||
if let Err(e) = keys.input(&bytes).await {
|
||||
app_state.panes.lock().await.remove(&handle);
|
||||
let event = PaneEvent::Error {
|
||||
message: format!("typing didn't reach the computer: {}", err(e)),
|
||||
};
|
||||
let json = serde_json::to_string(&event).expect("events serialize");
|
||||
let _ = refused.send(InvokeResponseBody::Json(json));
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user