diff --git a/src/api2/admin/datastore/h2upload.rs b/src/api2/admin/datastore/h2upload.rs index 7d700f68..4e03b665 100644 --- a/src/api2/admin/datastore/h2upload.rs +++ b/src/api2/admin/datastore/h2upload.rs @@ -1,16 +1,21 @@ use failure::*; +use lazy_static::lazy_static; -use futures::{Future, Stream}; -use h2::server; +use std::collections::HashMap; + +use futures::*; use hyper::header::{HeaderValue, UPGRADE}; -use hyper::{Body, Response, StatusCode}; +use hyper::{Body, Request, Response, StatusCode}; use hyper::http::request::Parts; use hyper::rt; use serde_json::Value; +use crate::tools; use crate::api_schema::router::*; use crate::api_schema::*; +use crate::server::formatter::*; +use crate::server::RestEnvironment; pub fn api_method_upgrade_h2upload() -> ApiAsyncMethod { 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) -> 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, Error = Self::Error> + Send>; + + fn call(&mut self, req: Request) -> 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::() { + 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( parts: Parts, req_body: Body, - param: Value, + _param: Value, _info: &ApiAsyncMethod, - _rpcenv: &mut RpcEnvironment, + rpcenv: &mut RpcEnvironment, ) -> Result { 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); } + let service = BackupService::new(rpcenv); + rt::spawn( req_body .on_upgrade() - .map_err(|e| Error::from(e)) + .map_err(Error::from) .and_then(move |conn| { 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 response = Response::builder() - .status(StatusCode::OK) - .body(()) - .unwrap(); + let mut http = hyper::server::conn::Http::new(); + http.http2_only(true); - // Send the response back to the client - respond.send_response(response, true) - .unwrap(); - - Ok(()) - }) - }) - .map_err(Error::from) + http.serve_connection(conn, service).map_err(Error::from) }) .map_err(|e| eprintln!("error during upgrade: {}", e)) ); @@ -83,3 +156,30 @@ fn upgrade_h2upload( .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 { + + + Ok(Value::Null) +}