tyler
/
proxmox-backup
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

src/api2/admin/datastore/h2upload.rs: improve http2 service

This commit is contained in:
Dietmar Maurer 2019-05-07 11:24:44 +02:00
parent 78f7e225e4
commit 92ac375aff
1 changed files with 126 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

152
src/api2/admin/datastore/h2upload.rs
View File

@ -1,16 +1,21 @@
use failure::*; use failure::*;
use lazy_static::lazy_static;
use futures::{Future, Stream}; use std::collections::HashMap;
use h2::server;
use futures::*;
use hyper::header::{HeaderValue, UPGRADE}; use hyper::header::{HeaderValue, UPGRADE};
use hyper::{Body, Response, StatusCode}; use hyper::{Body, Request, Response, StatusCode};
use hyper::http::request::Parts; use hyper::http::request::Parts;
use hyper::rt; use hyper::rt;
use serde_json::Value; use serde_json::Value;
use crate::tools;
use crate::api_schema::router::*; use crate::api_schema::router::*;
use crate::api_schema::*; use crate::api_schema::*;
use crate::server::formatter::*;
use crate::server::RestEnvironment;
pub fn api_method_upgrade_h2upload() -> ApiAsyncMethod { pub fn api_method_upgrade_h2upload() -> ApiAsyncMethod {
ApiAsyncMethod::new( ApiAsyncMethod::new(
@ -20,12 +25,95 @@ pub fn api_method_upgrade_h2upload() -> ApiAsyncMethod {
) )
} }
lazy_static!{
static ref BACKUP_ROUTER: Router = backup_api();
}
pub struct BackupService {
rpcenv: RestEnvironment,
}
impl BackupService {
fn new(rpcenv: &RpcEnvironment) -> Self {
let mut rpcenv = RestEnvironment::new(rpcenv.env_type());
rpcenv.set_user(rpcenv.get_user());
Self { rpcenv }
}
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()))),
};
let formatter = &JSON_FORMATTER;
println!("H2 REQUEST {} {}", method, path);
println!("H2 COMPO {:?}", components);
let mut uri_param = HashMap::new();
match BACKUP_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);
}
}
}
}
impl hyper::service::Service for BackupService {
type ReqBody = Body;
type ResBody = Body;
type Error = hyper::Error;
type Future = Box<Future<Item = Response<Body>, Error = Self::Error> + Send>;
fn call(&mut self, req: Request<Self::ReqBody>) -> Self::Future {
let _path = req.uri().path().to_owned();
let _method = req.method().clone();
Box::new(self.handle_request(req).then(move |result| {
match result {
Ok(res) => {
//log_response(method, &path, &res);
Ok::<_, hyper::Error>(res)
}
Err(err) => {
if let Some(apierr) = err.downcast_ref::<HttpError>() {
let mut resp = Response::new(Body::from(apierr.message.clone()));
*resp.status_mut() = apierr.code;
//log_response(method, &path, &resp);
Ok(resp)
} else {
let mut resp = Response::new(Body::from(err.to_string()));
*resp.status_mut() = StatusCode::BAD_REQUEST;
//log_response(method, &path, &resp);
Ok(resp)
}
}
}
}))
}
}
fn upgrade_h2upload( fn upgrade_h2upload(
parts: Parts, parts: Parts,
req_body: Body, req_body: Body,
param: Value, _param: Value,
_info: &ApiAsyncMethod, _info: &ApiAsyncMethod,
_rpcenv: &mut RpcEnvironment, rpcenv: &mut RpcEnvironment,
) -> Result<BoxFut, Error> { ) -> Result<BoxFut, Error> {
let expected_protocol: &'static str = "proxmox-backup-protocol-h2"; let expected_protocol: &'static str = "proxmox-backup-protocol-h2";
@ -43,34 +131,19 @@ fn upgrade_h2upload(
bail!("unexpected http version '{:?}' (expected version < 2)", parts.version); bail!("unexpected http version '{:?}' (expected version < 2)", parts.version);
} }
let service = BackupService::new(rpcenv);
rt::spawn( rt::spawn(
req_body req_body
.on_upgrade() .on_upgrade()
.map_err(|e| Error::from(e)) .map_err(Error::from)
.and_then(move |conn| { .and_then(move |conn| {
println!("upgrade done"); println!("upgrade done");
server::handshake(conn)
.and_then(|h2| {
println!("Accept h2");
// Accept all inbound HTTP/2.0 streams sent over the
// connection.
h2.for_each(|(request, mut respond)| {
println!("Received request: {:?}", request);
// Build a response with no body let mut http = hyper::server::conn::Http::new();
let response = Response::builder() http.http2_only(true);
.status(StatusCode::OK)
.body(())
.unwrap();
// Send the response back to the client http.serve_connection(conn, service).map_err(Error::from)
respond.send_response(response, true)
.unwrap();
Ok(())
})
})
.map_err(Error::from)
}) })
.map_err(|e| eprintln!("error during upgrade: {}", e)) .map_err(|e| eprintln!("error during upgrade: {}", e))
); );
@ -83,3 +156,30 @@ fn upgrade_h2upload(
.unwrap() .unwrap()
))) )))
} }
fn backup_api() -> Router {
let test1 = Router::new()
.get(
ApiMethod::new(
test1_get,
ObjectSchema::new("Test something.")
)
);
let router = Router::new()
.subdir("test1", test1)
.list_subdirs();
router
}
fn test1_get (
_param: Value,
_info: &ApiMethod,
_rpcenv: &mut RpcEnvironment,
) -> Result<Value, Error> {
Ok(Value::Null)
}