1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
|
use crate::OutputProtocol;
use crate::debug_log;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::io::AsyncWriteExt;
use tokio::net::UnixStream;
/// IPC output protocol.
///
/// Connects to a Unix domain socket and sends messages to it.
/// On Windows, this will fail with a clear error message since
/// Unix domain sockets are not supported.
#[derive(Debug)]
pub struct IPCOutputProtocol {
socket_path: PathBuf,
stream: tokio::sync::Mutex<Option<UnixStream>>,
}
impl IPCOutputProtocol {
pub fn new(socket_path: PathBuf) -> Self {
Self {
socket_path,
stream: tokio::sync::Mutex::new(None),
}
}
}
impl OutputProtocol for IPCOutputProtocol {
async fn init(&self) {
let path = self.socket_path.clone();
let stream_lock = &self.stream;
match UnixStream::connect(&path).await {
Ok(stream) => {
debug_log!("[IPC] Connected to {}", path.display());
*stream_lock.lock().await = Some(stream);
}
Err(e) => {
debug_log!("[IPC] Failed to connect to {}: {}", path.display(), e);
}
}
}
async fn send(self: Arc<Self>, message: &str) {
let mut guard = self.stream.lock().await;
if let Some(ref mut stream) = *guard {
let bytes = format!("{}\n", message);
if stream.write_all(bytes.as_bytes()).await.is_err() {
debug_log!("[IPC] Write error, reconnecting...");
*guard = None;
// Try to reconnect
if let Ok(new_stream) = UnixStream::connect(&self.socket_path).await {
debug_log!("[IPC] Reconnected to {}", self.socket_path.display());
*guard = Some(new_stream);
}
}
} else {
// Try to connect
if let Ok(stream) = UnixStream::connect(&self.socket_path).await {
debug_log!("[IPC] Connected to {}", self.socket_path.display());
let bytes = format!("{}\n", message);
let _ = stream.writable().await;
// We can't use the stream directly here since we need to store it
// for future sends. Let's store and send.
let _ = stream.try_write(bytes.as_bytes());
*guard = Some(stream);
}
}
}
}
|
