2019-05-08 15:36:19 +00:00
|
|
|
use failure::*;
|
|
|
|
|
|
|
|
|
|
use std::collections::HashMap;
|
|
|
|
|
use std::sync::Arc;
|
2019-08-23 12:11:14 +00:00
|
|
|
use std::task::{Context, Poll};
|
2019-05-08 15:36:19 +00:00
|
|
|
|
|
|
|
|
use futures::*;
|
|
|
|
|
use hyper::{Body, Request, Response, StatusCode};
|
|
|
|
|
|
|
|
|
|
use crate::tools;
|
|
|
|
|
use crate::api_schema::router::*;
|
|
|
|
|
use crate::server::formatter::*;
|
|
|
|
|
use crate::server::WorkerTask;
|
|
|
|
|
|
2019-06-26 15:29:12 +00:00
|
|
|
/// Hyper Service implementation to handle stateful H2 connections.
|
|
|
|
|
///
|
|
|
|
|
/// We use this kind of service to handle backup protocol
|
|
|
|
|
/// connections. State is stored inside the generic ``rpcenv``. Logs
|
|
|
|
|
/// goes into the ``WorkerTask`` log.
|
|
|
|
|
pub struct H2Service<E> {
|
|
|
|
|
router: &'static Router,
|
|
|
|
|
rpcenv: E,
|
2019-05-08 15:36:19 +00:00
|
|
|
worker: Arc<WorkerTask>,
|
2019-05-29 07:35:21 +00:00
|
|
|
debug: bool,
|
2019-05-08 15:36:19 +00:00
|
|
|
}
|
|
|
|
|
|
2019-06-26 15:29:12 +00:00
|
|
|
impl <E: RpcEnvironment + Clone> H2Service<E> {
|
2019-05-08 15:36:19 +00:00
|
|
|
|
2019-06-26 15:29:12 +00:00
|
|
|
pub fn new(rpcenv: E, worker: Arc<WorkerTask>, router: &'static Router, debug: bool) -> Self {
|
|
|
|
|
Self { rpcenv, worker, router, debug }
|
2019-05-29 07:35:21 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
pub fn debug<S: AsRef<str>>(&self, msg: S) {
|
|
|
|
|
if self.debug { self.worker.log(msg); }
|
2019-05-08 15:36:19 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
fn handle_request(&self, req: Request<Body>) -> BoxFut {
|
|
|
|
|
|
|
|
|
|
let (parts, body) = req.into_parts();
|
|
|
|
|
|
|
|
|
|
let method = parts.method.clone();
|
|
|
|
|
|
|
|
|
|
let (path, components) = match tools::normalize_uri_path(parts.uri.path()) {
|
|
|
|
|
Ok((p,c)) => (p, c),
|
|
|
|
|
Err(err) => return Box::new(future::err(http_err!(BAD_REQUEST, err.to_string()))),
|
|
|
|
|
};
|
|
|
|
|
|
2019-05-29 08:17:38 +00:00
|
|
|
self.debug(format!("{} {}", method, path));
|
2019-05-08 15:36:19 +00:00
|
|
|
|
|
|
|
|
let mut uri_param = HashMap::new();
|
|
|
|
|
|
2019-06-26 15:29:12 +00:00
|
|
|
let formatter = &JSON_FORMATTER;
|
|
|
|
|
|
|
|
|
|
match self.router.find_method(&components, method, &mut uri_param) {
|
2019-05-08 15:36:19 +00:00
|
|
|
MethodDefinition::None => {
|
|
|
|
|
let err = http_err!(NOT_FOUND, "Path not found.".to_string());
|
2019-06-26 15:29:12 +00:00
|
|
|
return Box::new(future::ok((formatter.format_error)(err)));
|
2019-05-08 15:36:19 +00:00
|
|
|
}
|
|
|
|
|
MethodDefinition::Simple(api_method) => {
|
2019-05-09 16:01:24 +00:00
|
|
|
return crate::server::rest::handle_sync_api_request(
|
2019-06-26 15:29:12 +00:00
|
|
|
self.rpcenv.clone(), api_method, formatter, parts, body, uri_param);
|
2019-05-08 15:36:19 +00:00
|
|
|
}
|
|
|
|
|
MethodDefinition::Async(async_method) => {
|
2019-05-09 16:01:24 +00:00
|
|
|
return crate::server::rest::handle_async_api_request(
|
2019-06-26 15:29:12 +00:00
|
|
|
self.rpcenv.clone(), async_method, formatter, parts, body, uri_param);
|
2019-05-08 15:36:19 +00:00
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
fn log_response(worker: Arc<WorkerTask>, method: hyper::Method, path: &str, resp: &Response<Body>) {
|
|
|
|
|
|
|
|
|
|
let status = resp.status();
|
|
|
|
|
|
|
|
|
|
if !status.is_success() {
|
|
|
|
|
let reason = status.canonical_reason().unwrap_or("unknown reason");
|
|
|
|
|
|
|
|
|
|
let mut message = "request failed";
|
|
|
|
|
if let Some(data) = resp.extensions().get::<ErrorMessageExtension>() {
|
|
|
|
|
message = &data.0;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
worker.log(format!("{} {}: {} {}: {}", method.as_str(), path, status.as_str(), reason, message));
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
2019-08-23 12:11:14 +00:00
|
|
|
impl <E: RpcEnvironment + Clone> tower_service::Service<Request<Body>> for H2Service<E> {
|
|
|
|
|
type Response = Response<Body>;
|
2019-05-30 06:10:06 +00:00
|
|
|
type Error = Error;
|
2019-08-23 12:11:14 +00:00
|
|
|
type Future =
|
|
|
|
|
std::pin::Pin<Box<dyn Future<Output = Result<Response<Body>, Self::Error>> + Send>>;
|
2019-05-08 15:36:19 +00:00
|
|
|
|
2019-08-23 12:11:14 +00:00
|
|
|
fn poll_ready(&mut self, _cx: &mut Context) -> Poll<Result<(), Self::Error>> {
|
|
|
|
|
Poll::Ready(Ok(()))
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
fn call(&mut self, req: Request<Body>) -> Self::Future {
|
2019-05-08 15:36:19 +00:00
|
|
|
let path = req.uri().path().to_owned();
|
|
|
|
|
let method = req.method().clone();
|
|
|
|
|
let worker = self.worker.clone();
|
|
|
|
|
|
2019-08-23 12:11:14 +00:00
|
|
|
std::pin::Pin::from(self.handle_request(req))
|
|
|
|
|
.map(move |result| match result {
|
2019-05-08 15:36:19 +00:00
|
|
|
Ok(res) => {
|
|
|
|
|
Self::log_response(worker, method, &path, &res);
|
2019-05-30 06:10:06 +00:00
|
|
|
Ok::<_, Error>(res)
|
2019-05-08 15:36:19 +00:00
|
|
|
}
|
|
|
|
|
Err(err) => {
|
2019-05-30 06:10:06 +00:00
|
|
|
if let Some(apierr) = err.downcast_ref::<HttpError>() {
|
2019-05-08 15:36:19 +00:00
|
|
|
let mut resp = Response::new(Body::from(apierr.message.clone()));
|
2019-05-23 06:05:39 +00:00
|
|
|
resp.extensions_mut().insert(ErrorMessageExtension(apierr.message.clone()));
|
2019-05-08 15:36:19 +00:00
|
|
|
*resp.status_mut() = apierr.code;
|
|
|
|
|
Self::log_response(worker, method, &path, &resp);
|
|
|
|
|
Ok(resp)
|
|
|
|
|
} else {
|
|
|
|
|
let mut resp = Response::new(Body::from(err.to_string()));
|
2019-05-23 06:05:39 +00:00
|
|
|
resp.extensions_mut().insert(ErrorMessageExtension(err.to_string()));
|
2019-05-08 15:36:19 +00:00
|
|
|
*resp.status_mut() = StatusCode::BAD_REQUEST;
|
|
|
|
|
Self::log_response(worker, method, &path, &resp);
|
|
|
|
|
Ok(resp)
|
|
|
|
|
}
|
|
|
|
|
}
|
2019-08-23 12:11:14 +00:00
|
|
|
})
|
|
|
|
|
.boxed()
|
2019-05-08 15:36:19 +00:00
|
|
|
}
|
|
|
|
|
}
|
