aboutsummaryrefslogtreecommitdiff
path: root/src/api
diff options
context:
space:
mode:
Diffstat (limited to 'src/api')
-rw-r--r--src/api/client.rs43
-rw-r--r--src/api/server.rs60
-rw-r--r--src/api/wait.rs4
3 files changed, 74 insertions, 33 deletions
diff --git a/src/api/client.rs b/src/api/client.rs
index 7d966965..38baf807 100644
--- a/src/api/client.rs
+++ b/src/api/client.rs
@@ -1,15 +1,16 @@
use std::fmt;
use std::io::{self, BufRead, BufReader, Write};
-use std::os::unix::net::UnixStream;
use std::path::PathBuf;
use std::time::Duration;
+use interprocess::local_socket::traits::Stream as _;
use serde::de::DeserializeOwned;
use crate::api::schema::{
ErrorResponse, EventsSubscribeParams, Method, PingParams, Request, ResponseResult,
SubscriptionEventEnvelope, SuccessResponse,
};
+use crate::ipc::LocalStream;
/// API connection target resolved by clients at the process edge.
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -66,8 +67,8 @@ impl ApiClient {
timeout: Duration,
) -> Result<serde_json::Value, ApiClientError> {
let mut stream = self.connect()?;
- stream.set_write_timeout(Some(timeout))?;
- stream.set_read_timeout(Some(timeout))?;
+ set_timeout_best_effort(&stream, TimeoutKind::Send, timeout)?;
+ set_timeout_best_effort(&stream, TimeoutKind::Recv, timeout)?;
write_request(&mut stream, request)?;
let mut reader = BufReader::new(stream);
@@ -97,7 +98,7 @@ impl ApiClient {
let mut stream = self.connect()?;
write_request(&mut stream, request)?;
if let Some(timeout) = read_timeout {
- stream.set_read_timeout(Some(timeout))?;
+ set_timeout_best_effort(&stream, TimeoutKind::Recv, timeout)?;
}
let mut reader = BufReader::new(stream);
@@ -124,13 +125,35 @@ impl ApiClient {
}
}
- fn connect(&self) -> io::Result<UnixStream> {
- UnixStream::connect(self.socket_path())
+ fn connect(&self) -> io::Result<LocalStream> {
+ crate::ipc::connect_local_stream(&self.socket_path())
+ }
+}
+
+enum TimeoutKind {
+ Send,
+ Recv,
+}
+
+fn set_timeout_best_effort(
+ stream: &LocalStream,
+ kind: TimeoutKind,
+ timeout: Duration,
+) -> io::Result<()> {
+ let result = match kind {
+ TimeoutKind::Send => stream.set_send_timeout(Some(timeout)),
+ TimeoutKind::Recv => stream.set_recv_timeout(Some(timeout)),
+ };
+ match result {
+ Ok(()) => Ok(()),
+ #[cfg(windows)]
+ Err(err) if err.kind() == io::ErrorKind::Unsupported => Ok(()),
+ Err(err) => Err(err),
}
}
pub struct EventStream {
- reader: BufReader<UnixStream>,
+ reader: BufReader<LocalStream>,
}
impl EventStream {
@@ -181,7 +204,7 @@ impl From<serde_json::Error> for ApiClientError {
}
}
-fn write_request(stream: &mut UnixStream, request: &Request) -> Result<(), ApiClientError> {
+fn write_request(stream: &mut LocalStream, request: &Request) -> Result<(), ApiClientError> {
stream.write_all(serde_json::to_string(request)?.as_bytes())?;
stream.write_all(b"\n")?;
stream.flush()?;
@@ -189,7 +212,7 @@ fn write_request(stream: &mut UnixStream, request: &Request) -> Result<(), ApiCl
}
fn read_json_line<T: DeserializeOwned>(
- reader: &mut BufReader<UnixStream>,
+ reader: &mut BufReader<LocalStream>,
) -> Result<T, ApiClientError> {
let mut line = String::new();
let read = reader.read_line(&mut line)?;
@@ -200,7 +223,7 @@ fn read_json_line<T: DeserializeOwned>(
}
fn read_optional_json_line<T: DeserializeOwned>(
- reader: &mut BufReader<UnixStream>,
+ reader: &mut BufReader<LocalStream>,
) -> Result<Option<T>, ApiClientError> {
let mut line = String::new();
let read = reader.read_line(&mut line)?;
diff --git a/src/api/server.rs b/src/api/server.rs
index e2c4b4a1..94e8c725 100644
--- a/src/api/server.rs
+++ b/src/api/server.rs
@@ -1,13 +1,13 @@
use std::io::{self, Read, Write};
-use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
+use interprocess::local_socket::traits::{ListenerExt as _, Stream as _};
use tracing::{debug, error, info, warn};
-#[cfg(test)]
+#[cfg(all(test, unix))]
use std::fs;
use crate::api::schema::{
@@ -16,7 +16,10 @@ use crate::api::schema::{
use crate::api::subscriptions::ActiveSubscription;
use crate::api::wait::wait_for_output;
use crate::api::{request_changes_ui, socket_path, ApiRequestMessage, ApiRequestSender, EventHub};
-use crate::ipc::{remove_socket_file_if_owned, socket_file_identity, SocketFileIdentity};
+use crate::ipc::{
+ bind_local_listener, remove_socket_file_if_owned, socket_file_identity, LocalStream,
+ SocketFileIdentity,
+};
const SOCKET_PERMISSION_MODE: u32 = 0o600;
pub(super) const CONNECTION_POLL_INTERVAL: Duration = Duration::from_millis(100);
@@ -46,7 +49,7 @@ impl Drop for ServerHandle {
impl ServerHandle {
pub(crate) fn remove_socket_file_if_owned(&self) -> std::io::Result<()> {
- remove_socket_file_if_owned(&self.path, self.identity)
+ remove_socket_file_if_owned(&self.path, &self.identity)
}
}
@@ -57,7 +60,9 @@ pub fn start_server(
start_server_with_capabilities(
api_tx,
event_hub,
- Some(ServerCapabilities { live_handoff: true }),
+ Some(ServerCapabilities {
+ live_handoff: crate::platform::capabilities().live_handoff,
+ }),
)
}
@@ -69,7 +74,7 @@ pub fn start_server_with_capabilities(
let path = socket_path();
prepare_socket_path(&path)?;
- let listener = UnixListener::bind(&path)?;
+ let listener = bind_local_listener(&path)?;
restrict_socket_permissions(&path)?;
let identity = socket_file_identity(&path)?;
info!(path = %path.display(), "api server listening");
@@ -127,13 +132,13 @@ fn restrict_socket_permissions(path: &Path) -> std::io::Result<()> {
}
fn handle_connection(
- mut stream: UnixStream,
+ mut stream: LocalStream,
api_tx: &ApiRequestSender,
event_hub: &EventHub,
running: &Arc<AtomicBool>,
capabilities: Option<ServerCapabilities>,
) -> std::io::Result<()> {
- if let Err(err) = stream.set_write_timeout(Some(STREAM_WRITE_TIMEOUT)) {
+ if let Err(err) = stream.set_send_timeout(Some(STREAM_WRITE_TIMEOUT)) {
debug!(err = %err, "api connection write timeout unavailable");
}
@@ -340,7 +345,7 @@ fn api_response_outcome(response: &str) -> &'static str {
}
}
-fn read_initial_request_line(stream: &mut UnixStream) -> std::io::Result<Option<String>> {
+fn read_initial_request_line(stream: &mut LocalStream) -> std::io::Result<Option<String>> {
stream.set_nonblocking(true)?;
let deadline = Instant::now() + INITIAL_REQUEST_TIMEOUT;
let mut bytes = Vec::new();
@@ -387,7 +392,7 @@ fn read_initial_request_line(stream: &mut UnixStream) -> std::io::Result<Option<
}
fn stream_subscriptions(
- mut stream: UnixStream,
+ mut stream: LocalStream,
request_id: String,
params: crate::api::schema::EventsSubscribeParams,
api_tx: &ApiRequestSender,
@@ -444,27 +449,30 @@ fn stream_subscriptions(
}
}
-fn write_text_line(stream: &mut UnixStream, value: &str) -> std::io::Result<()> {
+fn write_text_line(stream: &mut LocalStream, value: &str) -> std::io::Result<()> {
stream.write_all(value.as_bytes())?;
stream.write_all(b"\n")?;
stream.flush()
}
-fn write_text_line_allow_disconnect(stream: &mut UnixStream, value: &str) -> std::io::Result<()> {
+fn write_text_line_allow_disconnect(stream: &mut LocalStream, value: &str) -> std::io::Result<()> {
match write_text_line(stream, value) {
Err(err) if is_connection_closed_error(&err) => Ok(()),
result => result,
}
}
-fn write_json_line<T: serde::Serialize>(stream: &mut UnixStream, value: &T) -> std::io::Result<()> {
+fn write_json_line<T: serde::Serialize>(
+ stream: &mut LocalStream,
+ value: &T,
+) -> std::io::Result<()> {
let encoded = serde_json::to_string(value)
.map_err(|err| std::io::Error::other(format!("failed to encode json: {err}")))?;
write_text_line(stream, &encoded)
}
fn write_json_line_allow_disconnect<T: serde::Serialize>(
- stream: &mut UnixStream,
+ stream: &mut LocalStream,
value: &T,
) -> std::io::Result<()> {
let encoded = serde_json::to_string(value)
@@ -473,7 +481,7 @@ fn write_json_line_allow_disconnect<T: serde::Serialize>(
}
pub(super) fn should_stop_connection(
- stream: &mut UnixStream,
+ stream: &mut LocalStream,
running: &Arc<AtomicBool>,
) -> std::io::Result<bool> {
if !running.load(Ordering::Relaxed) {
@@ -483,7 +491,7 @@ pub(super) fn should_stop_connection(
probe_stream_closed(stream)
}
-fn probe_stream_closed(stream: &mut UnixStream) -> std::io::Result<bool> {
+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) {
@@ -581,11 +589,13 @@ fn error_response_json(id: String, code: &str, message: String) -> String {
})
}
-#[cfg(test)]
+#[cfg(all(test, unix))]
mod tests {
use super::*;
+ use interprocess::local_socket::traits::Listener as _;
use std::io::{BufRead, BufReader};
use std::os::unix::fs::PermissionsExt;
+ use std::os::unix::net::UnixListener;
use std::sync::{Mutex, OnceLock};
use tokio::sync::mpsc;
@@ -602,13 +612,21 @@ mod tests {
std::env::temp_dir().join(format!("herdr-{name}-{}-{nanos}", std::process::id()))
}
- fn read_line(stream: &mut UnixStream) -> String {
+ fn read_line(stream: &mut LocalStream) -> String {
let mut reader = BufReader::new(stream);
let mut line = String::new();
reader.read_line(&mut line).unwrap();
line
}
+ fn local_stream_pair(name: &str) -> (LocalStream, LocalStream, PathBuf) {
+ let path = unique_test_path(name);
+ let listener = crate::ipc::bind_local_listener(&path).unwrap();
+ let client = crate::ipc::connect_local_stream(&path).unwrap();
+ let server = listener.accept().unwrap();
+ (client, server, path)
+ }
+
#[test]
fn socket_path_prefers_explicit_env_override() {
let _guard = env_lock().lock().unwrap();
@@ -770,7 +788,7 @@ mod tests {
}
});
- let (mut client, server) = UnixStream::pair().unwrap();
+ let (mut client, server, _path) = local_stream_pair("api-wait-disconnect");
client
.write_all(br#"{"id":"req_wait","method":"pane.wait_for_output","params":{"pane_id":"pane_1","source":"recent","match":{"type":"substring","value":"never"}}}"#)
.unwrap();
@@ -800,7 +818,7 @@ mod tests {
#[test]
fn subscriptions_stop_when_client_disconnects() {
let (api_tx, _api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
- let (mut client, server) = UnixStream::pair().unwrap();
+ let (mut client, server, _path) = local_stream_pair("api-sub-disconnect");
client
.write_all(
br#"{"id":"sub_1","method":"events.subscribe","params":{"subscriptions":[{"type":"workspace.created"}]}}"#,
@@ -832,7 +850,7 @@ mod tests {
#[test]
fn subscriptions_stop_when_server_shuts_down() {
let (api_tx, _api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
- let (mut client, server) = UnixStream::pair().unwrap();
+ let (mut client, server, _path) = local_stream_pair("api-sub-shutdown");
client
.write_all(
br#"{"id":"sub_2","method":"events.subscribe","params":{"subscriptions":[{"type":"workspace.created"}]}}"#,
diff --git a/src/api/wait.rs b/src/api/wait.rs
index 4bce14f2..cbb39b58 100644
--- a/src/api/wait.rs
+++ b/src/api/wait.rs
@@ -1,4 +1,3 @@
-use std::os::unix::net::UnixStream;
use std::sync::atomic::AtomicBool;
use std::sync::Arc;
@@ -13,11 +12,12 @@ use crate::api::server::{
};
use crate::api::subscriptions::{match_output, output_match_read_source};
use crate::api::ApiRequestSender;
+use crate::ipc::LocalStream;
pub(super) fn wait_for_output(
request_id: String,
params: crate::api::schema::PaneWaitForOutputParams,
- stream: &mut UnixStream,
+ stream: &mut LocalStream,
api_tx: &ApiRequestSender,
running: &Arc<AtomicBool>,
) -> std::io::Result<Option<String>> {
generated by cgit v1.2.3 (git 2.39.1) at 2026-11-06 22:40:52 +0000