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
use crate::Object;
use hyper::body::Buf;
use hyper::Client;
use hyper::{Request, Response, Method, Body, StatusCode};
use serde::Serialize;
use serde_json::Value;
use thiserror::Error;
#[derive(Error, Debug)]
pub enum Error {
#[error("http error: status code {0}")]
HttpError(StatusCode),
#[error("internal http error: {0}")]
InternalHttpError(#[from] hyper::Error),
#[error("invalid json: {0}")]
InternalJsonError(#[from] serde_json::Error),
}
fn status_ok(res: &Response<Body>) -> Result<(), Error> {
if res.status() != StatusCode::OK {
Err(Error::HttpError(res.status()).into())
} else {
Ok(())
}
}
#[derive(Serialize)]
struct EmitRequest {
event: String,
data: Value,
}
pub struct HttpClient {
url: String,
}
impl HttpClient {
pub fn new<S: Into<String>>(url: S) -> Self {
HttpClient {
url: url.into(),
}
}
pub async fn get<S: Into<String>>(&self, pattern: S) -> Result<Vec<Object>, Error> {
let client = Client::new();
let url = self.url.to_owned() + "/query?pattern=" + &pattern.into();
let res = client.get(url.parse().unwrap()).await?;
status_ok(&res)?;
let body = hyper::body::aggregate(res).await?;
let objects = serde_json::from_reader(body.reader())?;
Ok(objects)
}
pub async fn set<S: Into<String>>(&self, name: S, value: Value) -> Result<(), Error> {
let client = Client::new();
let value_json = serde_json::to_string(&value)?;
let req = Request::builder()
.method(Method::POST)
.uri(self.url.to_owned() + "/objects/" + &name.into())
.body(Body::from(value_json)).unwrap();
let res = client.request(req).await?;
status_ok(&res)?;
Ok(())
}
pub async fn patch<S: Into<String>>(&self, name: S, value: Value) -> Result<(), Error> {
let client = Client::new();
let value_json = serde_json::to_string(&value)?;
let req = Request::builder()
.method(Method::PATCH)
.uri(self.url.to_owned() + "/objects/" + &name.into())
.body(Body::from(value_json)).unwrap();
let res = client.request(req).await?;
status_ok(&res)?;
Ok(())
}
pub async fn remove<S: Into<String>>(&self, name: S) -> Result<bool, Error> {
let client = Client::new();
let req = Request::builder()
.method(Method::DELETE)
.uri(self.url.to_owned() + "/objects/" + &name.into())
.body(Body::empty()).unwrap();
let res = client.request(req).await?;
match res.status() {
StatusCode::OK => Ok(true),
StatusCode::NOT_FOUND => Ok(false),
_ => Err(Error::HttpError(res.status()))
}
}
pub async fn emit<S: Into<String>, S2: Into<String>>(&self, object: S, event: S2, data: Value) -> Result<(), Error> {
let client = Client::new();
let emit_req = EmitRequest { event: event.into(), data: data.into() };
let json = serde_json::to_string(&emit_req)?;
let req = Request::builder()
.method(Method::POST)
.uri(self.url.to_owned() + "/events/" + &object.into())
.body(Body::from(json)).unwrap();
let res = client.request(req).await?;
status_ok(&res)?;
Ok(())
}
}