From 319e42552bd1547cd8df30ec2eb2786b49c1318b Mon Sep 17 00:00:00 2001 From: Dietmar Maurer Date: Wed, 26 Jun 2019 17:29:12 +0200 Subject: [PATCH] src/server/h2service.rs: implement generic h2 service --- src/server/h2service.rs | 121 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 121 insertions(+) create mode 100644 src/server/h2service.rs diff --git a/src/server/h2service.rs b/src/server/h2service.rs new file mode 100644 index 00000000..4535f39b --- /dev/null +++ b/src/server/h2service.rs @@ -0,0 +1,121 @@ +use failure::*; + +use std::collections::HashMap; +use std::sync::Arc; + +use futures::*; +use hyper::{Body, Request, Response, StatusCode}; + +use crate::tools; +use crate::api_schema::router::*; +use crate::server::formatter::*; +use crate::server::WorkerTask; + +/// 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 { + router: &'static Router, + rpcenv: E, + worker: Arc, + debug: bool, +} + +impl H2Service { + + pub fn new(rpcenv: E, worker: Arc, router: &'static Router, debug: bool) -> Self { + Self { rpcenv, worker, router, debug } + } + + pub fn debug>(&self, msg: S) { + if self.debug { self.worker.log(msg); } + } + + fn handle_request(&self, req: Request) -> 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()))), + }; + + self.debug(format!("{} {}", method, path)); + + let mut uri_param = HashMap::new(); + + let formatter = &JSON_FORMATTER; + + match self.router.find_method(&components, method, &mut uri_param) { + MethodDefinition::None => { + let err = http_err!(NOT_FOUND, "Path not found.".to_string()); + return Box::new(future::ok((formatter.format_error)(err))); + } + MethodDefinition::Simple(api_method) => { + return crate::server::rest::handle_sync_api_request( + self.rpcenv.clone(), api_method, formatter, parts, body, uri_param); + } + MethodDefinition::Async(async_method) => { + return crate::server::rest::handle_async_api_request( + self.rpcenv.clone(), async_method, formatter, parts, body, uri_param); + } + } + } + + fn log_response(worker: Arc, method: hyper::Method, path: &str, resp: &Response) { + + 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::() { + message = &data.0; + } + + worker.log(format!("{} {}: {} {}: {}", method.as_str(), path, status.as_str(), reason, message)); + } + } +} + +impl hyper::service::Service for H2Service { + type ReqBody = Body; + type ResBody = Body; + type Error = Error; + type Future = Box, Error = Self::Error> + Send>; + + fn call(&mut self, req: Request) -> Self::Future { + let path = req.uri().path().to_owned(); + let method = req.method().clone(); + let worker = self.worker.clone(); + + Box::new(self.handle_request(req).then(move |result| { + match result { + Ok(res) => { + Self::log_response(worker, method, &path, &res); + Ok::<_, Error>(res) + } + Err(err) => { + if let Some(apierr) = err.downcast_ref::() { + let mut resp = Response::new(Body::from(apierr.message.clone())); + resp.extensions_mut().insert(ErrorMessageExtension(apierr.message.clone())); + *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())); + resp.extensions_mut().insert(ErrorMessageExtension(err.to_string())); + *resp.status_mut() = StatusCode::BAD_REQUEST; + Self::log_response(worker, method, &path, &resp); + Ok(resp) + } + } + } + })) + } +}