use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use anyhow::{Context, bail};
use bytes::{Bytes, BytesMut};
use carte_api::ErrorBody;
use http::{HeaderMap, Method, Request, StatusCode, header};
use http_body_util::{BodyExt, Full};
use hyper::body::Incoming;
use hyper::upgrade::Upgraded;
use hyper_util::rt::TokioIo;
use serde::Serialize;
use serde::de::DeserializeOwned;
use tokio::net::UnixStream;
use crate::exit::{Exit, Failure};
pub struct Client {
socket: PathBuf,
}
static WARN_ABOUT_VERSION: AtomicBool = AtomicBool::new(false);
pub fn warn_about_version() {
WARN_ABOUT_VERSION.store(true, Ordering::Relaxed);
}
fn version_hint(headers: &HeaderMap) -> Option<String> {
let this = env!("CARGO_PKG_VERSION");
let daemon = headers
.get(carte_api::VERSION_HEADER)
.and_then(|value| value.to_str().ok());
match daemon {
Some(daemon) if daemon == this => None,
Some(daemon) => Some(format!(
"the daemon runs carte {daemon}, this is {this}; run `carte daemon restart`"
)),
None => Some(format!(
"the daemon runs an older carte than this one ({this}); run `carte daemon restart`"
)),
}
}
fn check_version(headers: &HeaderMap) {
warn_once(&WARN_ABOUT_VERSION, headers, &mut std::io::stderr());
}
fn warn_once(armed: &AtomicBool, headers: &HeaderMap, out: &mut impl std::io::Write) {
if armed.swap(false, Ordering::Relaxed)
&& let Some(hint) = version_hint(headers)
{
let _ = writeln!(out, "warning: {hint}");
}
}
impl Client {
pub fn new(socket: &Path) -> Self {
Client {
socket: socket.to_path_buf(),
}
}
pub async fn get<T: DeserializeOwned>(&self, path: &str) -> anyhow::Result<T> {
let response = self.send(Method::GET, path, Bytes::new()).await?;
decode(response).await
}
pub async fn post<B: Serialize, T: DeserializeOwned>(
&self,
path: &str,
body: &B,
) -> anyhow::Result<T> {
let response = self
.send(Method::POST, path, Bytes::from(serde_json::to_vec(body)?))
.await?;
decode(response).await
}
pub async fn stream<B: Serialize, T: DeserializeOwned>(
&self,
method: Method,
path: &str,
body: Option<&B>,
mut each: impl FnMut(T),
) -> anyhow::Result<()> {
let bytes = match body {
Some(body) => Bytes::from(serde_json::to_vec(body)?),
None => Bytes::new(),
};
let response = self.send(method, path, bytes).await?;
if !response.status().is_success() {
return decode::<serde_json::Value>(response).await.map(|_| ());
}
check_version(response.headers());
let mut body = response.into_body();
let mut buf = BytesMut::new();
while let Some(frame) = body.frame().await {
if let Ok(data) = frame?.into_data() {
buf.extend_from_slice(&data);
}
while let Some(pos) = buf.iter().position(|&b| b == b'\n') {
let line = buf.split_to(pos + 1);
each(serde_json::from_slice(&line[..pos])?);
}
}
Ok(())
}
pub async fn upgrade<B: Serialize>(
&self,
path: &str,
protocol: &str,
body: &B,
) -> anyhow::Result<TokioIo<Upgraded>> {
let stream = self.connect().await?;
let (mut sender, connection) =
hyper::client::conn::http1::handshake(TokioIo::new(stream)).await?;
tokio::spawn(connection.with_upgrades());
let request = Request::builder()
.method(Method::POST)
.uri(path)
.header("host", "carte")
.header("content-type", "application/json")
.header(header::CONNECTION, "upgrade")
.header(header::UPGRADE, protocol)
.body(Full::new(Bytes::from(serde_json::to_vec(body)?)))?;
let response = sender.send_request(request).await?;
if response.status() != StatusCode::SWITCHING_PROTOCOLS {
return decode::<serde_json::Value>(response).await.and_then(|_| {
bail!("the daemon didn't switch to {protocol}");
});
}
check_version(response.headers());
Ok(TokioIo::new(hyper::upgrade::on(response).await?))
}
async fn connect(&self) -> anyhow::Result<UnixStream> {
UnixStream::connect(&self.socket)
.await
.map_err(|err| unreachable(&self.socket, &err).into())
}
async fn send(
&self,
method: Method,
path: &str,
body: Bytes,
) -> anyhow::Result<http::Response<Incoming>> {
let stream = self.connect().await?;
let (mut sender, connection) =
hyper::client::conn::http1::handshake(TokioIo::new(stream)).await?;
tokio::spawn(connection);
let request = Request::builder()
.method(method)
.uri(path)
.header("host", "carte")
.header("content-type", "application/json")
.body(Full::new(body))?;
Ok(sender.send_request(request).await?)
}
}
fn unreachable(socket: &Path, err: &std::io::Error) -> Failure {
let message = if err.kind() == std::io::ErrorKind::ConnectionRefused && socket.exists() {
format!(
"the carte daemon is starting or stopping ({}: {err}); try again in a moment, and if this persists it may have crashed: start it with `carte daemon`",
socket.display()
)
} else {
format!(
"the carte daemon isn't running ({}: {err}); start it with `carte daemon`",
socket.display()
)
};
Failure::new(Exit::DaemonDown, message)
}
async fn decode<T: DeserializeOwned>(response: http::Response<Incoming>) -> anyhow::Result<T> {
let status = response.status();
let headers = response.headers().clone();
let bytes = response.into_body().collect().await?.to_bytes();
if status == StatusCode::OK {
check_version(&headers);
return serde_json::from_slice(&bytes).context("reading the daemon's answer");
}
let error = serde_json::from_slice::<ErrorBody>(&bytes).ok();
let outdated = matches!(
status,
StatusCode::METHOD_NOT_ALLOWED
| StatusCode::UNSUPPORTED_MEDIA_TYPE
| StatusCode::UNPROCESSABLE_ENTITY
) || (status == StatusCode::NOT_FOUND && error.is_none());
let found = error.is_some();
let mut message = match error {
Some(error) => error.error,
None => match String::from_utf8_lossy(&bytes).trim() {
"" => format!("the daemon answered {status}"),
text => format!("the daemon answered {status}: {text}"),
},
};
if outdated && let Some(hint) = version_hint(&headers) {
message = format!("{message}; {hint}");
}
if status == StatusCode::NOT_FOUND && found {
return Err(Failure::new(Exit::NotFound, message).into());
}
bail!("{message}")
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn a_socket_nobody_listens_on_means_starting_or_stopping() {
let dir = tempfile::tempdir().unwrap();
let socket = dir.path().join("carte.sock");
leave_a_socket_nobody_listens_on(&socket);
let err = Client::new(&socket)
.get::<serde_json::Value>("/")
.await
.unwrap_err();
let failure = err.downcast_ref::<Failure>().unwrap();
assert_eq!(failure.exit, Exit::DaemonDown);
assert!(failure.message.contains("is starting or stopping"), "{err}");
std::fs::remove_file(&socket).unwrap();
let err = Client::new(&socket)
.get::<serde_json::Value>("/")
.await
.unwrap_err();
assert!(err.to_string().contains("isn't running"), "{err}");
}
fn leave_a_socket_nobody_listens_on(path: &Path) {
use nix::sys::socket::{AddressFamily, SockFlag, SockType, UnixAddr, bind, socket};
use nix::sys::wait::{WaitStatus, waitpid};
use nix::unistd::{ForkResult, fork};
let addr = UnixAddr::new(path).unwrap();
match unsafe { fork() }.unwrap() {
ForkResult::Child => {
let bound = socket(
AddressFamily::Unix,
SockType::Stream,
SockFlag::empty(),
None,
)
.and_then(|fd| bind(std::os::fd::AsRawFd::as_raw_fd(&fd), &addr));
unsafe { libc::_exit(i32::from(bound.is_err())) }
}
ForkResult::Parent { child } => {
assert_eq!(waitpid(child, None).unwrap(), WaitStatus::Exited(child, 0));
}
}
}
fn stub(dir: &Path, version: Option<&'static str>, app: axum::Router) -> Client {
let socket = dir.join("test.sock");
let listener = tokio::net::UnixListener::bind(&socket).unwrap();
let app = app.layer(axum::middleware::map_response(
move |mut response: axum::response::Response| async move {
if let Some(version) = version {
response.headers_mut().insert(
carte_api::VERSION_HEADER,
http::HeaderValue::from_static(version),
);
}
response
},
));
tokio::spawn(async move { axum::serve(listener, app).await });
Client::new(&socket)
}
#[test]
fn warns_once_and_only_about_another_version() {
let headers = |version: &'static str| {
let mut headers = HeaderMap::new();
headers.insert(
carte_api::VERSION_HEADER,
http::HeaderValue::from_static(version),
);
headers
};
let this = env!("CARGO_PKG_VERSION");
let mut out = Vec::new();
let armed = AtomicBool::new(true);
warn_once(&armed, &headers("0.0.1"), &mut out);
warn_once(&armed, &headers("0.0.1"), &mut out);
assert_eq!(
String::from_utf8(std::mem::take(&mut out)).unwrap(),
format!(
"warning: the daemon runs carte 0.0.1, this is {this}; run `carte daemon restart`\n"
)
);
let armed = AtomicBool::new(true);
warn_once(&armed, &headers(this), &mut out);
warn_once(&armed, &headers("0.0.1"), &mut out);
assert!(out.is_empty(), "{}", String::from_utf8_lossy(&out));
let armed = AtomicBool::new(false);
warn_once(&armed, &headers("0.0.1"), &mut out);
assert!(out.is_empty(), "{}", String::from_utf8_lossy(&out));
}
#[tokio::test]
async fn a_daemon_without_a_route_is_named_as_another_version() {
let dir = tempfile::tempdir().unwrap();
let app = axum::Router::new()
.route("/v1/snapshot", axum::routing::get(|| async { "{}" }))
.fallback(|| async { StatusCode::NOT_FOUND });
let client = stub(dir.path(), Some("0.0.1"), app);
let err = client
.upgrade(
carte_api::paths::ATTACH,
carte_api::attach::PROTOCOL,
&serde_json::json!({}),
)
.await
.unwrap_err()
.to_string();
let this = env!("CARGO_PKG_VERSION");
assert!(
err.contains(&format!(
"the daemon runs carte 0.0.1, this is {this}; run `carte daemon restart`"
)),
"{err}"
);
assert!(
err.starts_with("the daemon answered 404 Not Found"),
"{err}"
);
}
#[tokio::test]
async fn a_daemon_without_the_version_header_is_older() {
let dir = tempfile::tempdir().unwrap();
let app = axum::Router::new().route(
"/v1/control",
axum::routing::post(|| async {
(
StatusCode::UNPROCESSABLE_ENTITY,
"Failed to deserialize the JSON body: unknown variant `nap`",
)
}),
);
let client = stub(dir.path(), None, app);
let err = client
.post::<_, serde_json::Value>("/v1/control", &serde_json::json!({}))
.await
.unwrap_err()
.to_string();
assert!(err.contains("unknown variant `nap`"), "{err}");
assert!(
err.contains("the daemon runs an older carte than this one"),
"{err}"
);
}
#[tokio::test]
async fn errors_from_the_same_version_have_no_hint() {
let dir = tempfile::tempdir().unwrap();
let app = axum::Router::new()
.route(
"/v1/control",
axum::routing::post(|| async {
(
StatusCode::NOT_FOUND,
axum::Json(ErrorBody {
error: "no app nova".into(),
}),
)
}),
)
.fallback(|| async { StatusCode::NOT_FOUND });
let client = stub(dir.path(), Some(env!("CARGO_PKG_VERSION")), app);
let err = client
.post::<_, serde_json::Value>("/v1/control", &serde_json::json!({}))
.await
.unwrap_err();
assert_eq!(err.downcast_ref::<Failure>().unwrap().exit, Exit::NotFound);
assert_eq!(err.to_string(), "no app nova");
let err = client
.post::<_, serde_json::Value>("/v1/attach", &serde_json::json!({}))
.await
.unwrap_err();
assert_eq!(err.to_string(), "the daemon answered 404 Not Found");
}
#[tokio::test]
async fn reads_bodies_larger_than_one_chunk() {
let dir = tempfile::tempdir().unwrap();
let socket = dir.path().join("test.sock");
let listener = tokio::net::UnixListener::bind(&socket).unwrap();
let big: Vec<String> = (0..20_000).map(|i| format!("item {i}")).collect();
let body = big.clone();
let app = axum::Router::new().route(
"/big",
axum::routing::get(move || async move { axum::Json(body) }),
);
tokio::spawn(async move { axum::serve(listener, app).await });
let got: Vec<String> = Client::new(&socket).get("/big").await.unwrap();
assert_eq!(got, big);
}
}