2019-04-27 06:57:35 +00:00
|
|
|
use failure::*;
|
2019-05-07 09:24:44 +00:00
|
|
|
use lazy_static::lazy_static;
|
2019-04-27 06:57:35 +00:00
|
|
|
|
2019-05-07 09:24:44 +00:00
|
|
|
use std::collections::HashMap;
|
2019-05-07 12:07:00 +00:00
|
|
|
use std::sync::Arc;
|
2019-05-07 09:24:44 +00:00
|
|
|
|
|
|
|
|
use futures::*;
|
2019-04-27 06:57:35 +00:00
|
|
|
use hyper::header::{HeaderValue, UPGRADE};
|
2019-05-07 09:24:44 +00:00
|
|
|
use hyper::{Body, Request, Response, StatusCode};
|
2019-04-27 06:57:35 +00:00
|
|
|
use hyper::http::request::Parts;
|
|
|
|
|
|
|
|
|
|
use serde_json::Value;
|
|
|
|
|
|
2019-05-07 09:24:44 +00:00
|
|
|
use crate::tools;
|
2019-04-27 06:57:35 +00:00
|
|
|
use crate::api_schema::router::*;
|
|
|
|
|
use crate::api_schema::*;
|
2019-05-07 09:24:44 +00:00
|
|
|
use crate::server::formatter::*;
|
2019-05-07 11:42:00 +00:00
|
|
|
use crate::server::{WorkerTask, RestEnvironment};
|
2019-04-27 06:57:35 +00:00
|
|
|
|
|
|
|
|
pub fn api_method_upgrade_h2upload() -> ApiAsyncMethod {
|
|
|
|
|
ApiAsyncMethod::new(
|
|
|
|
|
upgrade_h2upload,
|
|
|
|
|
ObjectSchema::new("Experimental h2 server")
|
|
|
|
|
.required("store", StringSchema::new("Datastore name.")),
|
|
|
|
|
)
|
|
|
|
|
}
|
|
|
|
|
|
2019-05-07 09:24:44 +00:00
|
|
|
lazy_static!{
|
|
|
|
|
static ref BACKUP_ROUTER: Router = backup_api();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
pub struct BackupService {
|
|
|
|
|
rpcenv: RestEnvironment,
|
2019-05-07 12:07:00 +00:00
|
|
|
worker: Arc<WorkerTask>,
|
2019-05-07 09:24:44 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
impl BackupService {
|
|
|
|
|
|
2019-05-07 12:07:00 +00:00
|
|
|
fn new(rpcenv: RestEnvironment, worker: Arc<WorkerTask>) -> Self {
|
|
|
|
|
Self { rpcenv, worker }
|
2019-05-07 09:24:44 +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()))),
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
let formatter = &JSON_FORMATTER;
|
|
|
|
|
|
2019-05-07 12:07:00 +00:00
|
|
|
self.worker.log(format!("H2 REQUEST {} {}", method, path));
|
|
|
|
|
self.worker.log(format!("H2 COMPO {:?}", components));
|
2019-05-07 09:24:44 +00:00
|
|
|
|
|
|
|
|
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);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
2019-05-07 12:07:00 +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 client = "unknown"; // fixme: howto get peer_addr ?
|
|
|
|
|
|
|
|
|
|
let mut message = "request failed";
|
|
|
|
|
if let Some(data) = resp.extensions().get::<ErrorMessageExtension>() {
|
|
|
|
|
message = &data.0;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
worker.log(format!("{} {}: {} {}: [client {}] {}", method.as_str(), path, status.as_str(), reason, client, message));
|
|
|
|
|
}
|
|
|
|
|
}
|
2019-05-07 09:24:44 +00:00
|
|
|
}
|
|
|
|
|
|
2019-05-07 11:42:00 +00:00
|
|
|
impl Drop for BackupService {
|
|
|
|
|
fn drop(&mut self) {
|
|
|
|
|
println!("SERVER DROP");
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
2019-05-07 09:24:44 +00:00
|
|
|
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 {
|
2019-05-07 12:07:00 +00:00
|
|
|
let path = req.uri().path().to_owned();
|
|
|
|
|
let method = req.method().clone();
|
|
|
|
|
let worker = self.worker.clone();
|
2019-05-07 09:24:44 +00:00
|
|
|
|
|
|
|
|
Box::new(self.handle_request(req).then(move |result| {
|
|
|
|
|
match result {
|
|
|
|
|
Ok(res) => {
|
2019-05-07 12:07:00 +00:00
|
|
|
Self::log_response(worker, method, &path, &res);
|
2019-05-07 09:24:44 +00:00
|
|
|
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;
|
2019-05-07 12:07:00 +00:00
|
|
|
Self::log_response(worker, method, &path, &resp);
|
2019-05-07 09:24:44 +00:00
|
|
|
Ok(resp)
|
|
|
|
|
} else {
|
|
|
|
|
let mut resp = Response::new(Body::from(err.to_string()));
|
|
|
|
|
*resp.status_mut() = StatusCode::BAD_REQUEST;
|
2019-05-07 12:07:00 +00:00
|
|
|
Self::log_response(worker, method, &path, &resp);
|
2019-05-07 09:24:44 +00:00
|
|
|
Ok(resp)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}))
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
2019-04-27 06:57:35 +00:00
|
|
|
fn upgrade_h2upload(
|
|
|
|
|
parts: Parts,
|
|
|
|
|
req_body: Body,
|
2019-05-07 09:24:44 +00:00
|
|
|
_param: Value,
|
2019-04-27 06:57:35 +00:00
|
|
|
_info: &ApiAsyncMethod,
|
2019-05-07 09:24:44 +00:00
|
|
|
rpcenv: &mut RpcEnvironment,
|
2019-04-27 06:57:35 +00:00
|
|
|
) -> Result<BoxFut, Error> {
|
|
|
|
|
let expected_protocol: &'static str = "proxmox-backup-protocol-h2";
|
|
|
|
|
|
|
|
|
|
let protocols = parts
|
|
|
|
|
.headers
|
|
|
|
|
.get("UPGRADE")
|
|
|
|
|
.ok_or_else(|| format_err!("missing Upgrade header"))?
|
|
|
|
|
.to_str()?;
|
|
|
|
|
|
|
|
|
|
if protocols != expected_protocol {
|
|
|
|
|
bail!("invalid protocol name");
|
|
|
|
|
}
|
|
|
|
|
|
2019-05-06 08:29:34 +00:00
|
|
|
if parts.version >= http::version::Version::HTTP_2 {
|
|
|
|
|
bail!("unexpected http version '{:?}' (expected version < 2)", parts.version);
|
|
|
|
|
}
|
|
|
|
|
|
2019-05-07 11:42:00 +00:00
|
|
|
let worker_id = String::from("test2workerid");
|
|
|
|
|
|
2019-05-07 09:24:44 +00:00
|
|
|
|
2019-05-07 12:07:00 +00:00
|
|
|
let mut rpcenv1 = RestEnvironment::new(rpcenv.env_type());
|
|
|
|
|
rpcenv1.set_user(rpcenv.get_user());
|
|
|
|
|
|
|
|
|
|
WorkerTask::spawn("test2_download", Some(worker_id), &rpcenv.get_user().unwrap(), true, move |worker| {
|
|
|
|
|
let service = BackupService::new(rpcenv1, worker.clone());
|
|
|
|
|
|
2019-05-08 05:14:40 +00:00
|
|
|
let abort_future = worker.abort_future();
|
|
|
|
|
|
2019-04-27 06:57:35 +00:00
|
|
|
req_body
|
|
|
|
|
.on_upgrade()
|
2019-05-07 09:24:44 +00:00
|
|
|
.map_err(Error::from)
|
2019-04-27 06:57:35 +00:00
|
|
|
.and_then(move |conn| {
|
2019-05-07 11:42:00 +00:00
|
|
|
worker.log("upgrade done");
|
2019-05-07 09:24:44 +00:00
|
|
|
|
|
|
|
|
let mut http = hyper::server::conn::Http::new();
|
|
|
|
|
http.http2_only(true);
|
|
|
|
|
|
2019-05-07 11:42:00 +00:00
|
|
|
http.serve_connection(conn, service)
|
|
|
|
|
.map_err(Error::from)
|
|
|
|
|
.then(|x| {
|
|
|
|
|
println!("H2 END");
|
|
|
|
|
x
|
|
|
|
|
})
|
2019-04-27 06:57:35 +00:00
|
|
|
})
|
2019-05-08 05:14:40 +00:00
|
|
|
.select(abort_future.map_err(|_| {}).then(move |_| { bail!("task aborted"); }))
|
2019-05-08 06:12:49 +00:00
|
|
|
.and_then(|(result, _)| Ok(result))
|
|
|
|
|
.map_err(|(err, _)| err)
|
2019-05-07 11:42:00 +00:00
|
|
|
}).unwrap();
|
2019-04-27 06:57:35 +00:00
|
|
|
|
|
|
|
|
Ok(Box::new(futures::future::ok(
|
|
|
|
|
Response::builder()
|
|
|
|
|
.status(StatusCode::SWITCHING_PROTOCOLS)
|
|
|
|
|
.header(UPGRADE, HeaderValue::from_static(expected_protocol))
|
|
|
|
|
.body(Body::empty())
|
|
|
|
|
.unwrap()
|
|
|
|
|
)))
|
|
|
|
|
}
|
2019-05-07 09:24:44 +00:00
|
|
|
|
|
|
|
|
fn backup_api() -> Router {
|
|
|
|
|
|
|
|
|
|
let test1 = Router::new()
|
|
|
|
|
.get(
|
|
|
|
|
ApiMethod::new(
|
|
|
|
|
test1_get,
|
2019-05-07 10:26:55 +00:00
|
|
|
ObjectSchema::new("Test sync callback.")
|
|
|
|
|
)
|
|
|
|
|
);
|
|
|
|
|
|
|
|
|
|
let test2 = Router::new()
|
|
|
|
|
.download(
|
|
|
|
|
ApiAsyncMethod::new(
|
|
|
|
|
test2_get,
|
|
|
|
|
ObjectSchema::new("Test async callback.")
|
2019-05-07 09:24:44 +00:00
|
|
|
)
|
|
|
|
|
);
|
|
|
|
|
|
|
|
|
|
let router = Router::new()
|
|
|
|
|
.subdir("test1", test1)
|
2019-05-07 10:26:55 +00:00
|
|
|
.subdir("test2", test2)
|
2019-05-07 09:24:44 +00:00
|
|
|
.list_subdirs();
|
|
|
|
|
|
|
|
|
|
router
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
fn test1_get (
|
|
|
|
|
_param: Value,
|
|
|
|
|
_info: &ApiMethod,
|
|
|
|
|
_rpcenv: &mut RpcEnvironment,
|
|
|
|
|
) -> Result<Value, Error> {
|
|
|
|
|
|
|
|
|
|
Ok(Value::Null)
|
|
|
|
|
}
|
2019-05-07 10:26:55 +00:00
|
|
|
|
|
|
|
|
fn test2_get(
|
2019-05-07 11:42:00 +00:00
|
|
|
_parts: Parts,
|
|
|
|
|
_req_body: Body,
|
|
|
|
|
_param: Value,
|
2019-05-07 10:26:55 +00:00
|
|
|
_info: &ApiAsyncMethod,
|
2019-05-07 11:42:00 +00:00
|
|
|
_rpcenv: &mut RpcEnvironment,
|
2019-05-07 10:26:55 +00:00
|
|
|
) -> Result<BoxFut, Error> {
|
|
|
|
|
|
|
|
|
|
let fut = tokio::timer::Interval::new_interval(std::time::Duration::from_millis(300))
|
|
|
|
|
.map_err(|err| http_err!(INTERNAL_SERVER_ERROR, format!("tokio timer interval error: {}", err)))
|
2019-05-08 05:14:40 +00:00
|
|
|
.take(50)
|
2019-05-07 10:26:55 +00:00
|
|
|
.for_each(|tv| {
|
|
|
|
|
println!("LOOP {:?}", tv);
|
|
|
|
|
Ok(())
|
|
|
|
|
})
|
|
|
|
|
.and_then(|_| {
|
|
|
|
|
println!("TASK DONE");
|
|
|
|
|
Ok(Response::builder()
|
|
|
|
|
.status(StatusCode::OK)
|
|
|
|
|
.body(Body::empty())
|
|
|
|
|
.unwrap())
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
Ok(Box::new(fut))
|
|
|
|
|
}
|
