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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
use eyre::{Context, Result};
#[cfg(windows)]
use tokio::net::TcpStream;
use tonic::Code;
use tonic::transport::{Channel, Endpoint, Uri};
use tower::service_fn;
use hyper_util::rt::TokioIo;
#[cfg(unix)]
use tokio::net::UnixStream;
use atuin_client::history::History;
use crate::history::{
EndHistoryReply, EndHistoryRequest, ShutdownRequest, StartHistoryReply, StartHistoryRequest,
StatusReply, StatusRequest, history_client::HistoryClient as HistoryServiceClient,
};
pub struct HistoryClient {
client: HistoryServiceClient<Channel>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DaemonClientErrorKind {
Connect,
Unavailable,
Unimplemented,
Other,
}
#[must_use]
pub fn classify_error(error: &eyre::Report) -> DaemonClientErrorKind {
for cause in error.chain() {
if cause.downcast_ref::<tonic::transport::Error>().is_some() {
return DaemonClientErrorKind::Connect;
}
if let Some(status) = cause.downcast_ref::<tonic::Status>() {
return match status.code() {
Code::Unavailable => DaemonClientErrorKind::Unavailable,
Code::Unimplemented => DaemonClientErrorKind::Unimplemented,
_ => DaemonClientErrorKind::Other,
};
}
}
DaemonClientErrorKind::Other
}
// Wrap the grpc client
impl HistoryClient {
#[cfg(unix)]
pub async fn new(path: String) -> Result<Self> {
let log_path = path.clone();
let channel = Endpoint::try_from("http://atuin_local_daemon:0")?
.connect_with_connector(service_fn(move |_: Uri| {
let path = path.clone();
async move {
Ok::<_, std::io::Error>(TokioIo::new(UnixStream::connect(path.clone()).await?))
}
}))
.await
.wrap_err_with(|| {
format!(
"failed to connect to local atuin daemon at {}. Is it running?",
&log_path
)
})?;
let client = HistoryServiceClient::new(channel);
Ok(HistoryClient { client })
}
#[cfg(not(unix))]
pub async fn new(port: u64) -> Result<Self> {
let channel = Endpoint::try_from("http://atuin_local_daemon:0")?
.connect_with_connector(service_fn(move |_: Uri| {
let url = format!("127.0.0.1:{port}");
async move {
Ok::<_, std::io::Error>(TokioIo::new(TcpStream::connect(url.clone()).await?))
}
}))
.await
.wrap_err_with(|| {
format!(
"failed to connect to local atuin daemon at 127.0.0.1:{port}. Is it running?"
)
})?;
let client = HistoryServiceClient::new(channel);
Ok(HistoryClient { client })
}
pub async fn start_history(&mut self, h: History) -> Result<StartHistoryReply> {
let req = StartHistoryRequest {
command: h.command,
cwd: h.cwd,
hostname: h.hostname,
session: h.session,
timestamp: h.timestamp.unix_timestamp_nanos() as u64,
};
Ok(self.client.start_history(req).await?.into_inner())
}
pub async fn end_history(
&mut self,
id: String,
duration: u64,
exit: i64,
) -> Result<EndHistoryReply> {
let req = EndHistoryRequest { id, duration, exit };
Ok(self.client.end_history(req).await?.into_inner())
}
pub async fn status(&mut self) -> Result<StatusReply> {
Ok(self.client.status(StatusRequest {}).await?.into_inner())
}
pub async fn shutdown(&mut self) -> Result<bool> {
let resp = self.client.shutdown(ShutdownRequest {}).await?.into_inner();
Ok(resp.accepted)
}
}
|
