Skip to content

Commit 46ec14e

Browse files
raphael-goetzraphael-goetz
committed
feat: adjusted parsing & content type processing for bugfixes
1 parent 3f5deb6 commit 46ec14e

1 file changed

Lines changed: 89 additions & 154 deletions

File tree

‎adapter/rest/src/main.rs‎

Lines changed: 89 additions & 154 deletions
Original file line numberDiff line numberDiff line change
@@ -4,27 +4,30 @@ use base::{
44
traits::Server as ServerTrait,
55
};
66
use http_body_util::{BodyExt, Full};
7+
use hyper::server::conn::http1;
78
use hyper::{Request, Response};
89
use hyper::{
910
StatusCode,
1011
body::{Bytes, Incoming},
1112
};
12-
use hyper::{
13-
header::{HeaderName, HeaderValue},
14-
server::conn::http1,
15-
};
1613
use hyper_util::rt::TokioIo;
17-
use std::collections::HashMap;
14+
use serde_json::json;
1815
use std::convert::Infallible;
1916
use std::net::SocketAddr;
2017
use std::sync::Arc;
2118
use tokio::net::TcpListener;
2219
use tonic::async_trait;
23-
use tucana::shared::{AdapterConfiguration, RuntimeFeature, value::Kind};
24-
use tucana::shared::{Struct, ValidationFlow, Value};
25-
use tucana::shared::{Translation, value::Kind::StructValue};
20+
use tucana::shared::{
21+
AdapterConfiguration, RuntimeFeature, Struct, Translation, ValidationFlow, Value,
22+
helper::{path::get_string, value::ToValue},
23+
value::Kind,
24+
};
25+
26+
use crate::response::{error_to_http_response, value_to_http_response};
2627

2728
mod config;
29+
mod content_type;
30+
mod response;
2831
mod route;
2932

3033
#[tokio::main]
@@ -72,163 +75,43 @@ struct HttpServer {
7275
addr: Option<SocketAddr>,
7376
}
7477

75-
fn json_error(status: StatusCode, msg: &str) -> Response<Full<Bytes>> {
76-
let body = format!(r#"{{"error": "{}"}}"#, msg);
77-
Response::builder()
78-
.status(status)
79-
.header("content-type", "application/json")
80-
.body(Full::new(Bytes::from(body)))
81-
.unwrap()
82-
}
83-
84-
fn build_response(
85-
status: StatusCode,
86-
headers: HashMap<String, String>,
87-
body: Vec<u8>,
88-
) -> Response<Full<Bytes>> {
89-
let mut builder = Response::builder().status(status);
90-
91-
{
92-
let h = builder.headers_mut().unwrap();
93-
for (k, v) in headers {
94-
let name = match HeaderName::from_bytes(k.as_bytes()) {
95-
Ok(n) => n,
96-
Err(_) => {
97-
log::warn!("Dropping invalid header name: {}", k);
98-
continue;
99-
}
100-
};
101-
102-
let value = match HeaderValue::from_str(&v) {
103-
Ok(v) => v,
104-
Err(_) => {
105-
log::warn!("Dropping invalid header value for {}: {:?}", k, v);
106-
continue;
107-
}
108-
};
109-
110-
h.insert(name, value);
111-
}
112-
}
113-
114-
builder.body(Full::new(Bytes::from(body))).unwrap()
115-
}
116-
11778
async fn execute_flow_to_hyper_response(
11879
flow: ValidationFlow,
119-
body: Vec<u8>,
80+
body: Value,
12081
store: Arc<base::store::AdapterStore>,
12182
) -> Response<Full<Bytes>> {
122-
let value: Option<Value> = if body.is_empty() {
123-
None
124-
} else {
125-
match prost::Message::decode(body.as_slice()) {
126-
Ok(v) => Some(v),
127-
Err(e) => {
128-
log::warn!("Failed to decode request body as protobuf Value: {}", e);
129-
return json_error(
130-
StatusCode::BAD_REQUEST,
131-
"Failed to decode request body as protobuf Value",
132-
);
133-
}
134-
}
135-
};
136-
137-
match store.validate_and_execute_flow(flow, value).await {
83+
match store.validate_and_execute_flow(flow, Some(body)).await {
13884
Some(result) => {
13985
log::debug!("Received Result: {:?}", result);
140-
let Value {
141-
kind: Some(StructValue(Struct { fields })),
142-
} = result
143-
else {
144-
return json_error(
145-
StatusCode::INTERNAL_SERVER_ERROR,
146-
"Flow result was not a struct",
147-
);
148-
};
14986

150-
let Some(headers_val) = fields.get("headers") else {
151-
return json_error(
152-
StatusCode::INTERNAL_SERVER_ERROR,
153-
"Flow result missing headers",
154-
);
155-
};
156-
let Some(status_code_val) = fields.get("status_code") else {
157-
return json_error(
158-
StatusCode::INTERNAL_SERVER_ERROR,
159-
"Flow result missing status_code",
160-
);
161-
};
162-
let Some(payload_val) = fields.get("payload") else {
163-
return json_error(
164-
StatusCode::INTERNAL_SERVER_ERROR,
165-
"Flow result missing payload",
166-
);
167-
};
168-
169-
// headers struct
170-
let Value {
87+
if let Value {
17188
kind:
17289
Some(Kind::StructValue(Struct {
173-
fields: header_fields,
90+
fields: result_fields,
17491
})),
175-
} = headers_val
176-
else {
177-
return json_error(
178-
StatusCode::INTERNAL_SERVER_ERROR,
179-
"headers was not a list of header entries",
180-
);
181-
};
182-
183-
let http_headers: HashMap<String, String> = header_fields
184-
.iter()
185-
.filter_map(|(k, v)| {
186-
if let Value {
187-
kind: Some(Kind::StringValue(x)),
188-
} = v
189-
{
190-
Some((k.clone(), x.clone()))
191-
} else {
192-
None
193-
}
194-
})
195-
.collect();
196-
197-
// status_code number
198-
let Some(Kind::NumberValue(code)) = status_code_val.kind else {
199-
return json_error(
200-
StatusCode::INTERNAL_SERVER_ERROR,
201-
"status_code was not a number",
202-
);
203-
};
204-
205-
// payload -> json bytes
206-
let json_val = tucana::shared::helper::value::to_json_value(payload_val.clone());
207-
let json = serde_json::to_vec_pretty(&json_val).unwrap_or_else(|err| {
208-
let fallback = serde_json::json!({
209-
"error": format!("Serialization failed: {}", err),
210-
});
211-
serde_json::to_vec(&fallback)
212-
.unwrap_or_else(|_| br#"{"error":"Serialization failed"}"#.to_vec())
213-
});
214-
215-
let http_code = match code.number {
216-
Some(num) => match num {
217-
tucana::shared::number_value::Number::Integer(int) => int as u16,
218-
tucana::shared::number_value::Number::Float(float) => float as u16,
219-
},
220-
None => {
221-
return json_error(StatusCode::INTERNAL_SERVER_ERROR, "Flow execution failed");
92+
} = &result
93+
{
94+
if result_fields.contains_key("name")
95+
&& result_fields.contains_key("message")
96+
&& !result_fields.contains_key("payload")
97+
&& !result_fields.contains_key("headers")
98+
{
99+
log::debug!("Detected a RuntimeError");
100+
let name = get_string("name", &result);
101+
let message = get_string("message", &result);
102+
103+
return error_to_http_response(
104+
StatusCode::INTERNAL_SERVER_ERROR,
105+
format!("{}: {}", name.unwrap(), message.unwrap()).as_str(),
106+
);
222107
}
223-
};
108+
}
224109

225-
let status =
226-
StatusCode::from_u16(http_code).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
227-
build_response(status, http_headers, json)
110+
value_to_http_response(result)
228111
}
229112
None => {
230113
log::error!("flow execution failed");
231-
json_error(StatusCode::INTERNAL_SERVER_ERROR, "Flow execution failed")
114+
error_to_http_response(StatusCode::INTERNAL_SERVER_ERROR, "Flow execution failed")
232115
}
233116
}
234117
}
@@ -239,22 +122,41 @@ pub async fn handle_request(
239122
) -> Result<Response<Full<Bytes>>, Infallible> {
240123
let method = req.method().clone();
241124
let path = req.uri().path().to_string();
125+
let headers = req.headers().clone();
242126

243127
// Read full body
244128
let body_bytes = match BodyExt::collect(req.into_body()).await {
245129
Ok(collected) => collected.to_bytes().to_vec(),
246130
Err(err) => {
247131
log::error!("Failed to read request body: {}", err);
248-
return Ok(json_error(
132+
return Ok(error_to_http_response(
249133
StatusCode::BAD_REQUEST,
250134
"Failed to read request body",
251135
));
252136
}
253137
};
254138

139+
let request_body_value = match content_type::parse_body_from_headers(&headers, &body_bytes) {
140+
Ok(value) => value,
141+
Err(err) => {
142+
log::warn!("Failed to parse request body: {}", err);
143+
let status_code = match err {
144+
content_type::BodyParseError::UnsupportedContentType { .. } => {
145+
StatusCode::UNSUPPORTED_MEDIA_TYPE
146+
}
147+
_ => StatusCode::BAD_REQUEST,
148+
};
149+
150+
return Ok(error_to_http_response(status_code, &err.to_string()));
151+
}
152+
};
153+
255154
// slug matching
256155
let Some(slug) = route::extract_slug_from_path(&path) else {
257-
return Ok(json_error(StatusCode::BAD_REQUEST, "Missing slug in path"));
156+
return Ok(error_to_http_response(
157+
StatusCode::BAD_REQUEST,
158+
"Missing slug in path",
159+
));
258160
};
259161

260162
let pattern = format!("REST.{}.*", slug);
@@ -265,9 +167,42 @@ pub async fn handle_request(
265167

266168
let resp = match store.get_possible_flow_match(pattern, route).await {
267169
FlowIdentifyResult::Single(flow) => {
268-
execute_flow_to_hyper_response(flow, body_bytes, store).await
170+
let mut header_fields = std::collections::HashMap::new();
171+
let mut fields = std::collections::HashMap::new();
172+
173+
for (name, value) in headers.iter() {
174+
let key = name.as_str().to_owned();
175+
let value_str = value
176+
.to_str()
177+
.map(str::to_owned)
178+
.unwrap_or_else(|_| String::from_utf8_lossy(value.as_bytes()).into_owned());
179+
180+
header_fields.insert(key, value_str.to_value());
181+
}
182+
183+
match request_body_value {
184+
Some(v) => {
185+
fields.insert(String::from("payload"), v);
186+
}
187+
None => {}
188+
};
189+
190+
fields.insert(
191+
String::from("headers"),
192+
Value {
193+
kind: Some(Kind::StructValue(Struct {
194+
fields: header_fields,
195+
})),
196+
},
197+
);
198+
199+
let input = Value {
200+
kind: Some(Kind::StructValue(Struct { fields })),
201+
};
202+
203+
execute_flow_to_hyper_response(flow, input, store).await
269204
}
270-
_ => json_error(StatusCode::NOT_FOUND, "No flow found for path"),
205+
_ => error_to_http_response(StatusCode::NOT_FOUND, "No flow found for path"),
271206
};
272207

273208
Ok(resp)

0 commit comments

Comments
 (0)