diff options
Diffstat (limited to 'src/api')
| -rw-r--r-- | src/api/server.rs | 186 | ||||
| -rw-r--r-- | src/api/wait.rs | 121 |
2 files changed, 268 insertions, 39 deletions
diff --git a/src/api/server.rs b/src/api/server.rs index 37b464ac..df0f8fc3 100644 --- a/src/api/server.rs +++ b/src/api/server.rs @@ -14,11 +14,11 @@ use crate::api::schema::{ ErrorBody, ErrorResponse, Method, Request, ResponseResult, ServerCapabilities, SuccessResponse, }; use crate::api::subscriptions::ActiveSubscription; -use crate::api::wait::wait_for_output; +use crate::api::wait::{wait_for_event, wait_for_output}; use crate::api::{request_changes_ui, socket_path, ApiRequestMessage, ApiRequestSender, EventHub}; use crate::ipc::{ - bind_local_listener, remove_socket_file_if_owned, socket_file_identity, LocalStream, - SocketFileIdentity, + bind_local_listener, is_connection_closed_error, local_stream_peer_closed, + remove_socket_file_if_owned, socket_file_identity, LocalStream, SocketFileIdentity, }; const SOCKET_PERMISSION_MODE: u32 = 0o600; @@ -197,6 +197,38 @@ fn handle_connection( } result } + Method::EventsWait(params) => { + let Some(response) = wait_for_event( + request_id.clone(), + params, + &mut stream, + api_tx, + event_hub, + running, + )? + else { + crate::logging::api_request_completed( + &request_id, + method, + "client_disconnected", + changes_ui, + ); + return Ok(()); + }; + let result = write_text_line_allow_disconnect(&mut stream, &response); + match &result { + Ok(()) => crate::logging::api_request_completed( + &request_id, + method, + api_response_outcome(&response), + changes_ui, + ), + Err(err) => { + crate::logging::api_request_failed(&request_id, method, &err.to_string()) + } + } + result + } Method::PaneWaitForOutput(params) => { let Some(response) = wait_for_output(request_id.clone(), params, &mut stream, api_tx, running)? @@ -515,40 +547,7 @@ pub(super) fn should_stop_connection( return Ok(true); } - probe_stream_closed(stream) -} - -fn probe_stream_closed(stream: &mut LocalStream) -> std::io::Result<bool> { - stream.set_nonblocking(true)?; - let mut probe = [0u8; 1]; - let status = match stream.read(&mut probe) { - Ok(0) => Ok(true), - Ok(_) => Ok(true), - Err(err) - if matches!( - err.kind(), - std::io::ErrorKind::WouldBlock | std::io::ErrorKind::Interrupted - ) => - { - Ok(false) - } - Err(err) if is_connection_closed_error(&err) => Ok(true), - Err(err) => Err(err), - }; - stream.set_nonblocking(false)?; - status -} - -fn is_connection_closed_error(err: &std::io::Error) -> bool { - matches!( - err.kind(), - std::io::ErrorKind::BrokenPipe - | std::io::ErrorKind::ConnectionAborted - | std::io::ErrorKind::ConnectionReset - | std::io::ErrorKind::NotConnected - | std::io::ErrorKind::UnexpectedEof - | std::io::ErrorKind::WriteZero - ) + local_stream_peer_closed(stream) } fn dispatch_to_app(request: Request, api_tx: &ApiRequestSender) -> String { @@ -620,6 +619,7 @@ fn error_response_json(id: String, code: &str, message: String) -> String { mod tests { use super::*; use interprocess::local_socket::traits::Listener as _; + use std::collections::HashMap; use std::io::{BufRead, BufReader}; use std::os::unix::fs::PermissionsExt; use std::os::unix::net::UnixListener; @@ -654,6 +654,64 @@ mod tests { (client, server, path) } + fn pane_info( + pane_id: &str, + agent_status: crate::api::schema::AgentStatus, + ) -> crate::api::schema::PaneInfo { + crate::api::schema::PaneInfo { + pane_id: pane_id.into(), + terminal_id: "term_1".into(), + workspace_id: "ws_1".into(), + tab_id: "tab_1".into(), + focused: true, + cwd: None, + foreground_cwd: None, + label: None, + agent: Some("pi".into()), + title: None, + display_agent: None, + agent_status, + custom_status: None, + state_labels: HashMap::new(), + agent_session: None, + revision: 0, + } + } + + fn spawn_pane_get_responder( + agent_status: crate::api::schema::AgentStatus, + ) -> (ApiRequestSender, std::thread::JoinHandle<()>) { + let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); + let responder = std::thread::spawn(move || { + while let Some(msg) = api_rx.blocking_recv() { + match msg.request.method { + Method::PaneGet(_) => msg + .respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: msg.request.id, + result: ResponseResult::PaneInfo { + pane: pane_info("pane_1", agent_status), + }, + }) + .unwrap(), + ) + .unwrap(), + Method::EventsWait(_) => msg + .respond_to + .send(error_response_json( + msg.request.id, + "unexpected_dispatch", + "events.wait should be handled by the api server".into(), + )) + .unwrap(), + other => panic!("unexpected request: {other:?}"), + } + } + }); + (api_tx, responder) + } + #[test] fn socket_path_prefers_explicit_env_override() { let _guard = env_lock().lock().unwrap(); @@ -784,6 +842,60 @@ mod tests { } #[test] + fn events_wait_agent_status_returns_initial_match() { + let (api_tx, responder) = + spawn_pane_get_responder(crate::api::schema::AgentStatus::Blocked); + + let (mut client, server, _path) = local_stream_pair("api-events-wait-initial"); + client + .write_all(br#"{"id":"wait_1","method":"events.wait","params":{"match_event":{"event":"pane_agent_status_changed","pane_id":"pane_1","agent_status":"blocked"},"timeout_ms":1000}}"#) + .unwrap(); + client.write_all(b"\n").unwrap(); + client.flush().unwrap(); + + let running = Arc::new(AtomicBool::new(true)); + let event_hub = EventHub::default(); + handle_connection(server, &api_tx, &event_hub, &running, None).unwrap(); + + let response: serde_json::Value = serde_json::from_str(&read_line(&mut client)).unwrap(); + assert_eq!(response["id"], "wait_1"); + assert_eq!(response["result"]["type"], "wait_matched"); + assert_eq!( + response["result"]["event"]["data"]["agent_status"], + "blocked" + ); + drop(api_tx); + responder.join().unwrap(); + } + + #[test] + fn events_wait_agent_status_times_out_server_side() { + let (api_tx, responder) = + spawn_pane_get_responder(crate::api::schema::AgentStatus::Unknown); + + let (mut client, server, _path) = local_stream_pair("api-events-wait-timeout"); + client + .write_all(br#"{"id":"wait_2","method":"events.wait","params":{"match_event":{"event":"pane_agent_status_changed","pane_id":"pane_1","agent_status":"blocked"},"timeout_ms":30}}"#) + .unwrap(); + client.write_all(b"\n").unwrap(); + client.flush().unwrap(); + + let running = Arc::new(AtomicBool::new(true)); + let event_hub = EventHub::default(); + handle_connection(server, &api_tx, &event_hub, &running, None).unwrap(); + + let response: serde_json::Value = serde_json::from_str(&read_line(&mut client)).unwrap(); + assert_eq!(response["id"], "wait_2"); + assert_eq!(response["error"]["code"], "timeout"); + assert_eq!( + response["error"]["message"], + "timed out waiting for event match" + ); + drop(api_tx); + responder.join().unwrap(); + } + + #[test] fn wait_for_output_stops_when_client_disconnects() { let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); let (first_read_tx, first_read_rx) = std::sync::mpsc::channel(); diff --git a/src/api/wait.rs b/src/api/wait.rs index 3ed08d70..e70472c3 100644 --- a/src/api/wait.rs +++ b/src/api/wait.rs @@ -4,14 +4,17 @@ use std::sync::Arc; use regex::Regex; use crate::api::schema::{ - ErrorBody, ErrorResponse, Method, Request, ResponseResult, SuccessResponse, + ErrorBody, ErrorResponse, EventData, EventEnvelope, EventKind, EventMatch, EventsWaitParams, + Method, Request, ResponseResult, Subscription, SubscriptionEventData, + SubscriptionEventEnvelope, SuccessResponse, }; use crate::api::server::{ dispatch_to_app_with_timeout, should_stop_connection, APP_RESPONSE_TIMEOUT, CONNECTION_POLL_INTERVAL, }; +use crate::api::subscriptions::ActiveSubscription; use crate::api::subscriptions::{match_output, output_match_read_source}; -use crate::api::ApiRequestSender; +use crate::api::{ApiRequestSender, EventHub}; use crate::ipc::LocalStream; pub(super) fn wait_for_output( @@ -122,3 +125,117 @@ pub(super) fn wait_for_output( std::thread::sleep(CONNECTION_POLL_INTERVAL); } } + +pub(super) fn wait_for_event( + request_id: String, + params: EventsWaitParams, + stream: &mut LocalStream, + api_tx: &ApiRequestSender, + event_hub: &EventHub, + running: &Arc<AtomicBool>, +) -> std::io::Result<Option<String>> { + let deadline = params + .timeout_ms + .map(|ms| std::time::Instant::now() + std::time::Duration::from_millis(ms)); + + let subscription = match event_match_subscription(&request_id, params.match_event) { + Ok(subscription) => subscription, + Err(response) => return Ok(Some(serde_json::to_string(&response).unwrap())), + }; + let mut active = match ActiveSubscription::new(subscription, &request_id, 0, api_tx, event_hub) + { + Ok(active) => active, + Err(response) => return Ok(Some(serde_json::to_string(&response).unwrap())), + }; + + loop { + if should_stop_connection(stream, running)? { + return Ok(None); + } + + if let Some(event) = active.poll(api_tx, event_hub) { + return Ok(Some(wait_matched_response(&request_id, event))); + } + + if deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) { + return Ok(Some( + serde_json::to_string(&ErrorResponse { + id: request_id, + error: ErrorBody { + code: "timeout".into(), + message: "timed out waiting for event match".into(), + }, + }) + .unwrap(), + )); + } + + std::thread::sleep(CONNECTION_POLL_INTERVAL); + } +} + +fn event_match_subscription( + request_id: &str, + match_event: EventMatch, +) -> Result<Subscription, ErrorResponse> { + match match_event { + EventMatch::PaneAgentStatusChanged { + pane_id, + agent_status, + } => Ok(Subscription::PaneAgentStatusChanged { + pane_id, + agent_status: Some(agent_status), + }), + _ => Err(ErrorResponse { + id: request_id.into(), + error: ErrorBody { + code: "unsupported_event_wait_match".into(), + message: "events.wait currently supports pane agent status matches".into(), + }, + }), + } +} + +fn wait_matched_response(request_id: &str, event: serde_json::Value) -> String { + let Ok(event) = serde_json::from_value::<SubscriptionEventEnvelope>(event) else { + return serde_json::to_string(&ErrorResponse { + id: request_id.into(), + error: ErrorBody { + code: "internal_error".into(), + message: "failed to decode matched event".into(), + }, + }) + .unwrap(); + }; + + let SubscriptionEventData::PaneAgentStatusChanged(data) = event.data else { + return serde_json::to_string(&ErrorResponse { + id: request_id.into(), + error: ErrorBody { + code: "unsupported_event_wait_match".into(), + message: "events.wait currently supports pane agent status matches".into(), + }, + }) + .unwrap(); + }; + + serde_json::to_string(&SuccessResponse { + id: request_id.into(), + result: ResponseResult::WaitMatched { + event: EventEnvelope { + event: EventKind::PaneAgentStatusChanged, + data: EventData::PaneAgentStatusChanged { + pane_id: data.pane_id, + workspace_id: data.workspace_id, + agent_status: data.agent_status, + agent: data.agent, + title: data.title, + display_agent: data.display_agent, + custom_status: data.custom_status, + state_labels: data.state_labels, + }, + }, + }, + }) + .unwrap() +} |