1use crate::error::decode_remote;
2use crate::sse;
3use crate::{Code, Error, Issue, RemoteError};
4use bytes::Bytes;
5use futures_core::Stream;
6use futures_util::StreamExt;
7use reqwest::header::{HeaderMap, HeaderValue, ACCEPT, CONTENT_TYPE};
8use serde::de::DeserializeOwned;
9use serde::{Deserialize, Serialize};
10use std::collections::{BTreeMap, VecDeque};
11use std::pin::Pin;
12use std::sync::Arc;
13use std::time::Duration;
14
15#[derive(Clone, Copy, Debug, PartialEq, Eq)]
16pub enum Method {
17 Get,
18 Post,
19 Put,
20 Patch,
21 Delete,
22}
23
24impl Method {
25 pub fn sends_body(self) -> bool {
26 !matches!(self, Method::Get | Method::Delete)
27 }
28
29 fn verb(self) -> reqwest::Method {
30 match self {
31 Method::Get => reqwest::Method::GET,
32 Method::Post => reqwest::Method::POST,
33 Method::Put => reqwest::Method::PUT,
34 Method::Patch => reqwest::Method::PATCH,
35 Method::Delete => reqwest::Method::DELETE,
36 }
37 }
38}
39
40pub fn encode_segment(value: &impl std::fmt::Display) -> String {
41 const HEX: &[u8; 16] = b"0123456789ABCDEF";
42 let raw = value.to_string();
43 let mut out = String::with_capacity(raw.len());
44 for byte in raw.bytes() {
45 match byte {
46 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
47 out.push(byte as char)
48 }
49 _ => {
50 out.push('%');
51 out.push(HEX[usize::from(byte >> 4)] as char);
52 out.push(HEX[usize::from(byte & 0x0f)] as char);
53 }
54 }
55 }
56 out
57}
58
59#[derive(Clone, Debug, Default)]
60pub struct CallOptions {
61 pub headers: HeaderMap,
62 pub timeout: Option<Duration>,
63}
64
65impl CallOptions {
66 pub fn new() -> Self {
67 CallOptions::default()
68 }
69
70 pub fn header(mut self, name: &'static str, value: &str) -> Self {
71 if let Ok(v) = HeaderValue::from_str(value) {
72 self.headers.insert(name, v);
73 }
74 self
75 }
76
77 pub fn timeout(mut self, d: Duration) -> Self {
78 self.timeout = Some(d);
79 self
80 }
81}
82
83#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
84pub struct Empty {}
85
86#[derive(
87 Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize,
88)]
89#[serde(transparent)]
90pub struct DurationNs(pub i64);
91
92impl DurationNs {
93 pub fn as_duration(&self) -> Duration {
94 Duration::from_nanos(self.0.max(0) as u64)
95 }
96}
97
98#[derive(
99 Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize,
100)]
101#[serde(transparent)]
102pub struct StringInt(#[serde(with = "crate::codec::string_int")] pub i64);
103
104#[derive(
105 Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize,
106)]
107#[serde(transparent)]
108pub struct StringUint(#[serde(with = "crate::codec::string_uint")] pub u64);
109
110#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
111#[serde(transparent)]
112pub struct Base64Bytes(#[serde(with = "crate::codec::base64")] pub Vec<u8>);
113
114pub trait Validate {
115 fn validate(&self, path: &mut Vec<String>, issues: &mut Vec<Issue>);
116
117 fn issues(&self) -> Vec<Issue> {
118 let mut path = Vec::new();
119 let mut issues = Vec::new();
120 self.validate(&mut path, &mut issues);
121 issues
122 }
123}
124
125macro_rules! leaf_validate {
126 ($($t:ty),* $(,)?) => {
127 $(impl Validate for $t {
128 fn validate(&self, _path: &mut Vec<String>, _issues: &mut Vec<Issue>) {}
129 })*
130 };
131}
132
133leaf_validate!(
134 String,
135 str,
136 bool,
137 i8,
138 i16,
139 i32,
140 i64,
141 u8,
142 u16,
143 u32,
144 u64,
145 f32,
146 f64,
147 serde_json::Value,
148 chrono::DateTime<chrono::Utc>,
149 Empty,
150 DurationNs,
151 StringInt,
152 StringUint,
153 Base64Bytes,
154);
155
156impl<T: Validate> Validate for Vec<T> {
157 fn validate(&self, path: &mut Vec<String>, issues: &mut Vec<Issue>) {
158 for (i, item) in self.iter().enumerate() {
159 path.push(i.to_string());
160 item.validate(path, issues);
161 path.pop();
162 }
163 }
164}
165
166impl<T: Validate, const N: usize> Validate for [T; N] {
167 fn validate(&self, path: &mut Vec<String>, issues: &mut Vec<Issue>) {
168 for (i, item) in self.iter().enumerate() {
169 path.push(i.to_string());
170 item.validate(path, issues);
171 path.pop();
172 }
173 }
174}
175
176impl<T: Validate> Validate for Option<T> {
177 fn validate(&self, path: &mut Vec<String>, issues: &mut Vec<Issue>) {
178 if let Some(v) = self {
179 v.validate(path, issues);
180 }
181 }
182}
183
184impl<T: Validate> Validate for Box<T> {
185 fn validate(&self, path: &mut Vec<String>, issues: &mut Vec<Issue>) {
186 (**self).validate(path, issues);
187 }
188}
189
190impl<K: ToString, V: Validate> Validate for BTreeMap<K, V> {
191 fn validate(&self, path: &mut Vec<String>, issues: &mut Vec<Issue>) {
192 for (key, item) in self {
193 path.push(key.to_string());
194 item.validate(path, issues);
195 path.pop();
196 }
197 }
198}
199
200type HeaderSource = Arc<dyn Fn() -> HeaderMap + Send + Sync>;
201
202#[derive(Clone)]
203pub struct Transport {
204 base: String,
205 client: reqwest::Client,
206 headers: Option<HeaderSource>,
207}
208
209impl std::fmt::Debug for Transport {
210 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
211 f.debug_struct("Transport")
212 .field("base", &self.base)
213 .finish()
214 }
215}
216
217impl Transport {
218 pub fn new(base_url: impl Into<String>) -> Self {
219 Transport::with_client(base_url, reqwest::Client::new())
220 }
221
222 pub fn with_client(base_url: impl Into<String>, client: reqwest::Client) -> Self {
223 let base = base_url.into();
224 Transport {
225 base: base.trim_end_matches('/').to_string(),
226 client,
227 headers: None,
228 }
229 }
230
231 pub fn with_headers(mut self, f: impl Fn() -> HeaderMap + Send + Sync + 'static) -> Self {
232 self.headers = Some(Arc::new(f));
233 self
234 }
235
236 pub fn base_url(&self) -> &str {
237 &self.base
238 }
239
240 fn prepare<I: Serialize + Validate>(
241 &self,
242 path: &str,
243 method: Method,
244 input: &I,
245 options: Option<&CallOptions>,
246 accept: &'static str,
247 ) -> Result<reqwest::RequestBuilder, Error> {
248 let issues = input.issues();
249 if !issues.is_empty() {
250 return Err(Error::Invalid(issues));
251 }
252 let body = serde_json::to_string(input)?;
253 let url = format!("{}/{}", self.base, path);
254 let request = self.client.request(method.verb(), url);
255 let mut request = if method.sends_body() {
256 request.header(CONTENT_TYPE, "application/json").body(body)
257 } else {
258 request.query(&[("input", body.as_str())])
259 };
260 request = request.header(ACCEPT, accept);
261 Ok(self.apply(request, options))
262 }
263
264 fn prepare_rest<I: Serialize + Validate>(
265 &self,
266 path: &str,
267 method: Method,
268 input: &I,
269 drop: &[&str],
270 options: Option<&CallOptions>,
271 accept: &'static str,
272 ) -> Result<reqwest::RequestBuilder, Error> {
273 let payload = rest_payload(input, drop)?;
274 let url = format!("{}/{}", self.base, path);
275 let request = self.client.request(method.verb(), url);
276 let mut request = if method.sends_body() {
277 request
278 .header(CONTENT_TYPE, "application/json")
279 .body(serde_json::to_string(&payload)?)
280 } else {
281 request.query(&query_pairs(&payload))
282 };
283 request = request.header(ACCEPT, accept);
284 Ok(self.apply(request, options))
285 }
286
287 fn apply(
288 &self,
289 mut request: reqwest::RequestBuilder,
290 options: Option<&CallOptions>,
291 ) -> reqwest::RequestBuilder {
292 if let Some(source) = &self.headers {
293 request = request.headers(source());
294 }
295 if let Some(options) = options {
296 request = request.headers(options.headers.clone());
297 if let Some(timeout) = options.timeout {
298 request = request.timeout(timeout);
299 }
300 }
301 request
302 }
303
304 async fn send(request: reqwest::RequestBuilder) -> Result<reqwest::Response, Error> {
305 let response = request.send().await.map_err(map_transport)?;
306 let status = response.status();
307 if status.as_u16() >= 400 {
308 let body = response.bytes().await.map_err(map_transport)?;
309 return Err(Error::Remote(decode_remote(status.as_u16(), &body)));
310 }
311 Ok(response)
312 }
313
314 pub async fn call<I: Serialize + Validate, O: DeserializeOwned>(
315 &self,
316 path: &str,
317 method: Method,
318 input: &I,
319 options: Option<&CallOptions>,
320 ) -> Result<O, Error> {
321 let request = self.prepare(path, method, input, options, "application/json")?;
322 let response = Transport::send(request).await?;
323 let body = response.bytes().await.map_err(map_transport)?;
324 decode_body(&body)
325 }
326
327 pub async fn call_rest<I: Serialize + Validate, O: DeserializeOwned>(
328 &self,
329 path: &str,
330 method: Method,
331 input: &I,
332 drop: &[&str],
333 options: Option<&CallOptions>,
334 ) -> Result<O, Error> {
335 let request = self.prepare_rest(path, method, input, drop, options, "application/json")?;
336 let response = Transport::send(request).await?;
337 let body = response.bytes().await.map_err(map_transport)?;
338 decode_body(&body)
339 }
340
341 pub fn subscribe<I: Serialize + Validate, O: DeserializeOwned + Send + 'static>(
342 &self,
343 path: &str,
344 method: Method,
345 input: &I,
346 options: Option<&CallOptions>,
347 ) -> impl Stream<Item = Result<O, Error>> + Send + 'static {
348 events(self.prepare(path, method, input, options, "text/event-stream"))
349 }
350
351 pub fn subscribe_rest<I: Serialize + Validate, O: DeserializeOwned + Send + 'static>(
352 &self,
353 path: &str,
354 method: Method,
355 input: &I,
356 drop: &[&str],
357 options: Option<&CallOptions>,
358 ) -> impl Stream<Item = Result<O, Error>> + Send + 'static {
359 events(self.prepare_rest(path, method, input, drop, options, "text/event-stream"))
360 }
361
362 pub async fn upload<I: Serialize + Validate, O: DeserializeOwned>(
363 &self,
364 path: &str,
365 input: &I,
366 file: reqwest::Body,
367 filename: &str,
368 options: Option<&CallOptions>,
369 ) -> Result<O, Error> {
370 let issues = input.issues();
371 if !issues.is_empty() {
372 return Err(Error::Invalid(issues));
373 }
374 let body = serde_json::to_string(input)?;
375 self.post_form(path, body, file, filename, options).await
376 }
377
378 pub async fn upload_rest<I: Serialize + Validate, O: DeserializeOwned>(
379 &self,
380 path: &str,
381 input: &I,
382 drop: &[&str],
383 file: reqwest::Body,
384 filename: &str,
385 options: Option<&CallOptions>,
386 ) -> Result<O, Error> {
387 let body = serde_json::to_string(&rest_payload(input, drop)?)?;
388 self.post_form(path, body, file, filename, options).await
389 }
390
391 async fn post_form<O: DeserializeOwned>(
392 &self,
393 path: &str,
394 body: String,
395 file: reqwest::Body,
396 filename: &str,
397 options: Option<&CallOptions>,
398 ) -> Result<O, Error> {
399 let input_part = reqwest::multipart::Part::text(body)
400 .mime_str("application/json")
401 .map_err(map_transport)?;
402 let file_part = reqwest::multipart::Part::stream(file).file_name(filename.to_string());
403 let form = reqwest::multipart::Form::new()
404 .part("input", input_part)
405 .part("file", file_part);
406 let request = self
407 .client
408 .post(format!("{}/{}", self.base, path))
409 .header(ACCEPT, "application/json")
410 .multipart(form);
411 let response = Transport::send(self.apply(request, options)).await?;
412 let bytes = response.bytes().await.map_err(map_transport)?;
413 decode_body(&bytes)
414 }
415}
416
417fn events<O: DeserializeOwned + Send + 'static>(
418 prepared: Result<reqwest::RequestBuilder, Error>,
419) -> impl Stream<Item = Result<O, Error>> + Send + 'static {
420 futures_util::stream::unfold(SubscribeState::Start(prepared), |state| async move {
421 let mut state = state;
422 loop {
423 match state {
424 SubscribeState::Start(Err(err)) => return Some((Err(err), SubscribeState::Done)),
425 SubscribeState::Start(Ok(request)) => match Transport::send(request).await {
426 Ok(response) => {
427 state = SubscribeState::Reading {
428 body: Some(Box::pin(response.bytes_stream())),
429 parser: sse::Parser::new(),
430 queue: VecDeque::new(),
431 };
432 }
433 Err(err) => return Some((Err(err), SubscribeState::Done)),
434 },
435 SubscribeState::Reading {
436 mut body,
437 mut parser,
438 mut queue,
439 } => {
440 if let Some(item) = queue.pop_front() {
441 let next = if item.is_err() {
442 SubscribeState::Done
443 } else {
444 SubscribeState::Reading {
445 body,
446 parser,
447 queue,
448 }
449 };
450 return Some((item, next));
451 }
452 let stream = body.as_mut()?;
453 match stream.next().await {
454 Some(Ok(chunk)) => {
455 for event in parser.push(&chunk) {
456 match shape_event::<O>(&event) {
457 Some(item) => queue.push_back(item),
458 None => {
459 body = None;
460 break;
461 }
462 }
463 }
464 }
465 Some(Err(err)) => {
466 queue.push_back(Err(map_transport(err)));
467 body = None;
468 }
469 None => {
470 if let Some(event) = parser.finish() {
471 if let Some(item) = shape_event::<O>(&event) {
472 queue.push_back(item);
473 }
474 }
475 body = None;
476 }
477 }
478 state = SubscribeState::Reading {
479 body,
480 parser,
481 queue,
482 };
483 }
484 SubscribeState::Done => return None,
485 }
486 }
487 })
488}
489
490fn rest_payload<I: Serialize + Validate>(
491 input: &I,
492 drop: &[&str],
493) -> Result<serde_json::Value, Error> {
494 let issues = input.issues();
495 if !issues.is_empty() {
496 return Err(Error::Invalid(issues));
497 }
498 let mut payload = serde_json::to_value(input)?;
499 if let Some(fields) = payload.as_object_mut() {
500 for name in drop {
501 fields.remove(*name);
502 }
503 }
504 Ok(payload)
505}
506
507fn query_pairs(payload: &serde_json::Value) -> Vec<(String, String)> {
508 let Some(fields) = payload.as_object() else {
509 return Vec::new();
510 };
511 let mut pairs = Vec::new();
512 for (name, value) in fields {
513 match value {
514 serde_json::Value::Null => {}
515 serde_json::Value::Array(items) => {
516 for item in items.iter().filter(|item| !item.is_null()) {
517 pairs.push((name.clone(), query_value(item)));
518 }
519 }
520 _ => pairs.push((name.clone(), query_value(value))),
521 }
522 }
523 pairs
524}
525
526fn query_value(value: &serde_json::Value) -> String {
527 match value {
528 serde_json::Value::String(text) => text.clone(),
529 other => other.to_string(),
530 }
531}
532
533type ByteStream = Pin<Box<dyn Stream<Item = Result<Bytes, reqwest::Error>> + Send>>;
534
535enum SubscribeState<O> {
536 Start(Result<reqwest::RequestBuilder, Error>),
537 Reading {
538 body: Option<ByteStream>,
539 parser: sse::Parser,
540 queue: VecDeque<Result<O, Error>>,
541 },
542 Done,
543}
544
545fn shape_event<O: DeserializeOwned>(event: &sse::Event) -> Option<Result<O, Error>> {
546 match event.name.as_str() {
547 "message" => Some(serde_json::from_str(&event.data).map_err(Error::Decode)),
548 "error" => Some(Err(Error::Remote(decode_remote(
549 500,
550 event.data.as_bytes(),
551 )))),
552 "done" => None,
553 _ => Some(Err(Error::Remote(RemoteError::new(
554 Code::Internal,
555 format!("unexpected event {}", event.name),
556 0,
557 )))),
558 }
559}
560
561fn decode_body<O: DeserializeOwned>(body: &[u8]) -> Result<O, Error> {
562 if body.iter().all(|b| b.is_ascii_whitespace()) {
563 return serde_json::from_str("{}").map_err(Error::Decode);
564 }
565 serde_json::from_slice(body).map_err(Error::Decode)
566}
567
568fn map_transport(err: reqwest::Error) -> Error {
569 if err.is_timeout() {
570 return Error::Remote(RemoteError::new(Code::DeadlineExceeded, err.to_string(), 0));
571 }
572 Error::Transport(err)
573}