src/client/http_client.rs

   1 use failure::*;
   2
   3 use http::Uri;
   4 use hyper::Body;
   5 use hyper::client::Client;
   6 use xdg::BaseDirectories;
   7 use chrono::Utc;
   8 use std::collections::HashSet;
   9 use std::sync::{Arc, Mutex};
  10
  11 use http::{Request, Response};
  12 use http::header::HeaderValue;
  13
  14 use futures::*;
  15 use futures::stream::Stream;
  16 use std::sync::atomic::{AtomicUsize, Ordering};
  17 use tokio::sync::mpsc;
  18
  19 use serde_json::{json, Value};
  20 use url::percent_encoding::{percent_encode,  DEFAULT_ENCODE_SET};
  21
  22 use crate::tools::{self, BroadcastFuture, tty};
  23 use super::pipe_to_stream::*;
  24
  25 #[derive(Clone)]
  26 struct AuthInfo {
  27     username: String,
  28     ticket: String,
  29     token: String,
  30 }
  31
  32 /// HTTP(S) API client
  33 pub struct HttpClient {
  34     client: Client<hyper_tls::HttpsConnector<hyper::client::HttpConnector>>,
  35     server: String,
  36     auth: BroadcastFuture<AuthInfo>,
  37 }
  38
  39 fn store_ticket_info(server: &str, username: &str, ticket: &str, token: &str) -> Result<(), Error> {
  40
  41     let base = BaseDirectories::with_prefix("proxmox-backup")?;
  42
  43     // usually /run/user/<uid>/...
  44     let path = base.place_runtime_file("tickets")?;
  45
  46     let mode = nix::sys::stat::Mode::from_bits_truncate(0o0600);
  47
  48     let mut data = tools::file_get_json(&path, Some(json!({})))?;
  49
  50     let now = Utc::now().timestamp();
  51
  52     data[server][username] = json!({ "timestamp": now, "ticket": ticket, "token": token});
  53
  54     let mut new_data = json!({});
  55
  56     let ticket_lifetime = tools::ticket::TICKET_LIFETIME - 60;
  57
  58     let empty = serde_json::map::Map::new();
  59     for (server, info) in data.as_object().unwrap_or(&empty) {
  60         for (_user, uinfo) in info.as_object().unwrap_or(&empty) {
  61             if let Some(timestamp) = uinfo["timestamp"].as_i64() {
  62                 let age = now - timestamp;
  63                 if age < ticket_lifetime {
  64                     new_data[server][username] = uinfo.clone();
  65                 }
  66             }
  67         }
  68     }
  69
  70     tools::file_set_contents(path, new_data.to_string().as_bytes(), Some(mode))?;
  71
  72     Ok(())
  73 }
  74
  75 fn load_ticket_info(server: &str, username: &str) -> Option<(String, String)> {
  76     let base = match BaseDirectories::with_prefix("proxmox-backup") {
  77         Ok(b) => b,
  78         _ => return None,
  79     };
  80
  81     // usually /run/user/<uid>/...
  82     let path = match base.place_runtime_file("tickets") {
  83         Ok(p) => p,
  84         _ => return None,
  85     };
  86
  87     let data = match tools::file_get_json(&path, None) {
  88         Ok(v) => v,
  89         _ => return None,
  90     };
  91
  92     let now = Utc::now().timestamp();
  93
  94     let ticket_lifetime = tools::ticket::TICKET_LIFETIME - 60;
  95
  96     if let Some(uinfo) = data[server][username].as_object() {
  97         if let Some(timestamp) = uinfo["timestamp"].as_i64() {
  98             let age = now - timestamp;
  99             if age < ticket_lifetime {
 100                 let ticket = match uinfo["ticket"].as_str() {
 101                     Some(t) => t,
 102                     None => return None,
 103                 };
 104                 let token = match uinfo["token"].as_str() {
 105                     Some(t) => t,
 106                     None => return None,
 107                 };
 108                 return Some((ticket.to_owned(), token.to_owned()));
 109             }
 110         }
 111     }
 112
 113     None
 114 }
 115
 116 impl HttpClient {
 117
 118     pub fn new(server: &str, username: &str) -> Result<Self, Error> {
 119         let client = Self::build_client();
 120
 121         let password = if let Some((ticket, _token)) = load_ticket_info(server, username) {
 122             ticket
 123         } else {
 124             Self::get_password(&username)?
 125         };
 126
 127         let login = Self::credentials(client.clone(), server.to_owned(), username.to_owned(), password);
 128
 129         Ok(Self {
 130             client,
 131             server: String::from(server),
 132             auth: BroadcastFuture::new(login),
 133         })
 134     }
 135
 136     fn get_password(_username: &str) -> Result<String, Error> {
 137         use std::env::VarError::*;
 138         match std::env::var("PBS_PASSWORD") {
 139             Ok(p) => return Ok(p),
 140             Err(NotUnicode(_)) => bail!("PBS_PASSWORD contains bad characters"),
 141             Err(NotPresent) => {
 142                 // Try another method
 143             }
 144         }
 145
 146         // If we're on a TTY, query the user for a password
 147         if tty::stdin_isatty() {
 148             return Ok(String::from_utf8(tty::read_password("Password: ")?)?);
 149         }
 150
 151         bail!("no password input mechanism available");
 152     }
 153
 154     fn build_client() -> Client<hyper_tls::HttpsConnector<hyper::client::HttpConnector>> {
 155         let mut builder = native_tls::TlsConnector::builder();
 156         // FIXME: We need a CLI option for this!
 157         builder.danger_accept_invalid_certs(true);
 158         let tlsconnector = builder.build().unwrap();
 159         let mut httpc = hyper::client::HttpConnector::new(1);
 160         //httpc.set_nodelay(true); // not sure if this help?
 161         httpc.enforce_http(false); // we want https...
 162         let mut https = hyper_tls::HttpsConnector::from((httpc, tlsconnector));
 163         https.https_only(true); // force it!
 164         Client::builder()
 165         //.http2_initial_stream_window_size( (1 << 31) - 2)
 166         //.http2_initial_connection_window_size( (1 << 31) - 2)
 167             .build::<_, Body>(https)
 168     }
 169
 170     pub fn request(&self, mut req: Request<Body>) -> impl Future<Item=Value, Error=Error>  {
 171
 172         let login = self.auth.listen();
 173
 174         let client = self.client.clone();
 175
 176         login.and_then(move |auth| {
 177
 178             let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
 179             req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
 180             req.headers_mut().insert("CSRFPreventionToken", HeaderValue::from_str(&auth.token).unwrap());
 181
 182             let request = Self::api_request(client, req);
 183
 184             request
 185         })
 186     }
 187
 188     pub fn get(&self, path: &str, data: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 189
 190         let req = Self::request_builder(&self.server, "GET", path, data).unwrap();
 191         self.request(req)
 192     }
 193
 194     pub fn delete(&mut self, path: &str, data: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 195
 196         let req = Self::request_builder(&self.server, "DELETE", path, data).unwrap();
 197         self.request(req)
 198     }
 199
 200     pub fn post(&mut self, path: &str, data: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 201
 202         let req = Self::request_builder(&self.server, "POST", path, data).unwrap();
 203         self.request(req)
 204     }
 205
 206     pub fn download(&mut self, path: &str, mut output: Box<dyn std::io::Write + Send>) ->  impl Future<Item=(), Error=Error> {
 207
 208         let mut req = Self::request_builder(&self.server, "GET", path, None).unwrap();
 209
 210         let login = self.auth.listen();
 211
 212         let client = self.client.clone();
 213
 214         login.and_then(move |auth| {
 215
 216             let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
 217             req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
 218
 219             client.request(req)
 220                 .map_err(Error::from)
 221                 .and_then(|resp| {
 222
 223                     let _status = resp.status(); // fixme: ??
 224
 225                     resp.into_body()
 226                         .map_err(Error::from)
 227                         .for_each(move |chunk| {
 228                             output.write_all(&chunk)?;
 229                             Ok(())
 230                         })
 231
 232                 })
 233         })
 234     }
 235
 236     pub fn upload(&mut self, content_type: &str, body: Body, path: &str) -> impl Future<Item=Value, Error=Error> {
 237
 238         let path = path.trim_matches('/');
 239         let url: Uri = format!("https://{}:8007/{}", &self.server, path).parse().unwrap();
 240
 241         let req = Request::builder()
 242             .method("POST")
 243             .uri(url)
 244             .header("User-Agent", "proxmox-backup-client/1.0")
 245             .header("Content-Type", content_type)
 246             .body(body).unwrap();
 247
 248         self.request(req)
 249     }
 250
 251     pub fn start_backup(
 252         &self,
 253         datastore: &str,
 254         backup_type: &str,
 255         backup_id: &str,
 256     ) -> impl Future<Item=BackupClient, Error=Error> {
 257
 258         let path = format!("/api2/json/admin/datastore/{}/backup", datastore);
 259         let param = json!({"backup-type": backup_type, "backup-id": backup_id});
 260         let mut req = Self::request_builder(&self.server, "GET", &path, Some(param)).unwrap();
 261
 262         let login = self.auth.listen();
 263
 264         let client = self.client.clone();
 265
 266         login.and_then(move |auth| {
 267
 268             let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
 269             req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
 270             req.headers_mut().insert("UPGRADE", HeaderValue::from_str("proxmox-backup-protocol-h2").unwrap());
 271
 272             client.request(req)
 273                 .map_err(Error::from)
 274                 .and_then(|resp| {
 275
 276                     let status = resp.status();
 277                     if status != http::StatusCode::SWITCHING_PROTOCOLS {
 278                         future::Either::A(Self::api_response(resp).and_then(|_| { bail!("unknown error"); }))
 279                     } else {
 280                         future::Either::B(resp.into_body().on_upgrade().map_err(Error::from))
 281                     }
 282                 })
 283                 .and_then(|upgraded| {
 284                     h2::client::handshake(upgraded).map_err(Error::from)
 285                 })
 286                 .and_then(|(h2, connection)| {
 287                     let connection = connection
 288                         .map_err(|_| panic!("HTTP/2.0 connection failed"));
 289
 290                     // Spawn a new task to drive the connection state
 291                     hyper::rt::spawn(connection);
 292
 293                     // Wait until the `SendRequest` handle has available capacity.
 294                     h2.ready()
 295                         .map(BackupClient::new)
 296                         .map_err(Error::from)
 297                 })
 298         })
 299     }
 300
 301     fn credentials(
 302         client: Client<hyper_tls::HttpsConnector<hyper::client::HttpConnector>>,
 303         server: String,
 304         username: String,
 305         password: String,
 306     ) -> Box<Future<Item=AuthInfo, Error=Error> + Send> {
 307
 308         let server2 = server.clone();
 309
 310         let create_request = futures::future::lazy(move || {
 311             let data = json!({ "username": username, "password": password });
 312             let req = Self::request_builder(&server, "POST", "/api2/json/access/ticket", Some(data)).unwrap();
 313             Self::api_request(client, req)
 314         });
 315
 316         let login_future = create_request
 317             .and_then(move |cred| {
 318                 let auth = AuthInfo {
 319                     username: cred["data"]["username"].as_str().unwrap().to_owned(),
 320                     ticket: cred["data"]["ticket"].as_str().unwrap().to_owned(),
 321                     token: cred["data"]["CSRFPreventionToken"].as_str().unwrap().to_owned(),
 322                 };
 323
 324                 let _ = store_ticket_info(&server2, &auth.username, &auth.ticket, &auth.token);
 325
 326                 Ok(auth)
 327             });
 328
 329         Box::new(login_future)
 330     }
 331
 332     fn api_response(response: Response<Body>) -> impl Future<Item=Value, Error=Error> {
 333
 334         let status = response.status();
 335
 336         response
 337             .into_body()
 338             .concat2()
 339             .map_err(Error::from)
 340             .and_then(move |data| {
 341
 342                 let text = String::from_utf8(data.to_vec()).unwrap();
 343                 if status.is_success() {
 344                     if text.len() > 0 {
 345                         let value: Value = serde_json::from_str(&text)?;
 346                         Ok(value)
 347                     } else {
 348                         Ok(Value::Null)
 349                     }
 350                 } else {
 351                     bail!("HTTP Error {}: {}", status, text);
 352                 }
 353             })
 354     }
 355
 356     fn api_request(
 357         client: Client<hyper_tls::HttpsConnector<hyper::client::HttpConnector>>,
 358         req: Request<Body>
 359     ) -> impl Future<Item=Value, Error=Error> {
 360
 361         client.request(req)
 362             .map_err(Error::from)
 363             .and_then(Self::api_response)
 364     }
 365
 366     pub fn request_builder(server: &str, method: &str, path: &str, data: Option<Value>) -> Result<Request<Body>, Error> {
 367         let path = path.trim_matches('/');
 368         let url: Uri = format!("https://{}:8007/{}", server, path).parse()?;
 369
 370         if let Some(data) = data {
 371             if method == "POST" {
 372                 let request = Request::builder()
 373                     .method(method)
 374                     .uri(url)
 375                     .header("User-Agent", "proxmox-backup-client/1.0")
 376                     .header(hyper::header::CONTENT_TYPE, "application/json")
 377                     .body(Body::from(data.to_string()))?;
 378                 return Ok(request);
 379             } else {
 380                 let query = tools::json_object_to_query(data)?;
 381                 let url: Uri = format!("https://{}:8007/{}?{}", server, path, query).parse()?;
 382                 let request = Request::builder()
 383                     .method(method)
 384                     .uri(url)
 385                     .header("User-Agent", "proxmox-backup-client/1.0")
 386                     .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
 387                     .body(Body::empty())?;
 388                 return Ok(request);
 389             }
 390         }
 391
 392         let request = Request::builder()
 393             .method(method)
 394             .uri(url)
 395             .header("User-Agent", "proxmox-backup-client/1.0")
 396             .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
 397             .body(Body::empty())?;
 398
 399         Ok(request)
 400     }
 401 }
 402
 403 //#[derive(Clone)]
 404 pub struct BackupClient {
 405     h2: H2Client,
 406 }
 407
 408 struct ChunkInfo {
 409     digest: [u8; 32],
 410     data: bytes::BytesMut,
 411     offset: u64,
 412 }
 413
 414 impl BackupClient {
 415
 416     pub fn new(h2: h2::client::SendRequest<bytes::Bytes>) -> Self {
 417         Self { h2: H2Client::new(h2) }
 418     }
 419
 420     pub fn get(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 421         self.h2.get(path, param)
 422     }
 423
 424     pub fn put(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 425         self.h2.put(path, param)
 426     }
 427
 428     pub fn post(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 429         self.h2.post(path, param)
 430     }
 431
 432     pub fn finish(&self) -> impl Future<Item=(), Error=Error> {
 433         self.h2.clone().post("finish", None).map(|_| ())
 434     }
 435
 436     pub fn upload_dynamic_stream(
 437         &self,
 438         archive_name: &str,
 439         stream: impl Stream<Item=bytes::BytesMut, Error=Error>,
 440     ) -> impl Future<Item=(), Error=Error> {
 441
 442         let known_chunks = Arc::new(Mutex::new(HashSet::new()));
 443
 444         let mut stream_len = 0u64;
 445
 446         let stream = stream.
 447             map(move |data| {
 448                 let digest = openssl::sha::sha256(&data);
 449                 stream_len += data.len() as u64;
 450                 ChunkInfo { data, digest, offset: stream_len }
 451             });
 452
 453         let h2 = self.h2.clone();
 454         let h2_2 = self.h2.clone();
 455         let h2_3 = self.h2.clone();
 456         let h2_4 = self.h2.clone();
 457
 458         let param = json!({ "archive-name": archive_name });
 459
 460         Self::download_chunk_list(h2, "dynamic_index", archive_name, known_chunks.clone())
 461             .and_then(move |_| {
 462                 h2_2.post("dynamic_index", Some(param))
 463             })
 464             .and_then(move |res| {
 465                 let wid = res.as_u64().unwrap();
 466                 Self::upload_stream(h2_3, wid, stream, known_chunks.clone())
 467                     .and_then(move |(chunk_count, size, _speed)| {
 468                         let param = json!({
 469                             "wid": wid ,
 470                             "chunk-count": chunk_count,
 471                             "size": size,
 472                         });
 473                         h2_4.post("dynamic_close", Some(param))
 474                     })
 475                     .map(|_| ())
 476             })
 477     }
 478
 479     fn response_queue() -> (
 480         mpsc::Sender<h2::client::ResponseFuture>,
 481         sync::oneshot::Receiver<Result<(), Error>>
 482     ) {
 483         let (verify_queue_tx, verify_queue_rx) = mpsc::channel(100);
 484         let (verify_result_tx, verify_result_rx) = sync::oneshot::channel();
 485
 486         hyper::rt::spawn(
 487             verify_queue_rx
 488                 .map_err(Error::from)
 489                 .for_each(|response: h2::client::ResponseFuture| {
 490                     response
 491                         .map_err(Error::from)
 492                         .and_then(H2Client::h2api_response)
 493                         .and_then(|result| {
 494                             println!("RESPONSE: {:?}", result);
 495                             Ok(())
 496                         })
 497                         .map_err(|err| format_err!("pipelined request failed: {}", err))
 498                 })
 499                 .then(|result|
 500                       verify_result_tx.send(result)
 501                 )
 502                 .map_err(|_| { /* ignore closed channel */ })
 503         );
 504
 505         (verify_queue_tx, verify_result_rx)
 506     }
 507
 508     fn download_chunk_list(
 509         h2: H2Client,
 510         path: &str,
 511         archive_name: &str,
 512         known_chunks: Arc<Mutex<HashSet<[u8;32]>>>,
 513     ) -> impl Future<Item=(), Error=Error> {
 514
 515         let param = json!({ "archive-name": archive_name });
 516         let request = H2Client::request_builder("localhost", "GET", path, Some(param)).unwrap();
 517
 518         h2.send_request(request, None)
 519             .and_then(move |response| {
 520                 response
 521                     .map_err(Error::from)
 522                     .and_then(move |resp| {
 523                         let status = resp.status();
 524                         if !status.is_success() {
 525                             bail!("download chunk list failed with status {}", status);
 526                         }
 527
 528                         let (_head, body) = resp.into_parts();
 529
 530                         Ok(body)
 531                     })
 532                     .and_then(move |mut body| {
 533
 534                         let mut release_capacity = body.release_capacity().clone();
 535
 536                         crate::backup::DigestListDecoder::new(body.map_err(Error::from))
 537                             .for_each(move |chunk| {
 538                                 let _ = release_capacity.release_capacity(chunk.len());
 539                                 println!("GOT DOWNLOAD {}", tools::digest_to_hex(&chunk));
 540                                 known_chunks.lock().unwrap().insert(chunk);
 541                                 Ok(())
 542                             })
 543                        })
 544             })
 545     }
 546
 547     fn upload_stream(
 548         h2: H2Client,
 549         wid: u64,
 550         stream: impl Stream<Item=ChunkInfo, Error=Error>,
 551         known_chunks: Arc<Mutex<HashSet<[u8;32]>>>,
 552     ) -> impl Future<Item=(usize, usize, usize), Error=Error> {
 553
 554         let repeat = std::sync::Arc::new(AtomicUsize::new(0));
 555         let repeat2 = repeat.clone();
 556
 557         let stream_len = std::sync::Arc::new(AtomicUsize::new(0));
 558         let stream_len2 = stream_len.clone();
 559
 560         let (upload_queue, upload_result) = Self::response_queue();
 561
 562         let start_time = std::time::Instant::now();
 563
 564         stream
 565             .for_each(move |chunk_info| {
 566                 let h2 = h2.clone();
 567
 568                 repeat.fetch_add(1, Ordering::SeqCst);
 569                 stream_len.fetch_add(chunk_info.data.len(), Ordering::SeqCst);
 570
 571                 let upload_queue = upload_queue.clone();
 572
 573                 let mut known_chunks = known_chunks.lock().unwrap();
 574                 let chunk_is_known = known_chunks.contains(&chunk_info.digest);
 575
 576                 let upload_data;
 577                 let request;
 578
 579                 if chunk_is_known {
 580                     println!("append existing chunk ({} bytes)", chunk_info.data.len());
 581                     let param = json!({ "wid": wid, "digest": tools::digest_to_hex(&chunk_info.digest) });
 582                     request = H2Client::request_builder("localhost", "PUT", "dynamic_index", Some(param)).unwrap();
 583                     upload_data = None;
 584                 } else {
 585                     println!("upload new chunk {} ({} bytes)", tools::digest_to_hex(&chunk_info.digest), chunk_info.data.len());
 586                     known_chunks.insert(chunk_info.digest);
 587                     let param = json!({ "wid": wid, "size" : chunk_info.data.len() });
 588                     request = H2Client::request_builder("localhost", "POST", "dynamic_chunk", Some(param)).unwrap();
 589                     upload_data = Some(chunk_info.data.freeze());
 590                 }
 591
 592                 h2.send_request(request, upload_data)
 593                     .and_then(move |response| {
 594                         upload_queue.send(response)
 595                             .map(|_| ()).map_err(Error::from)
 596                     })
 597             })
 598             .then(move |result| {
 599                 println!("RESULT {:?}", result);
 600                 upload_result.map_err(Error::from).and_then(|upload1_result| {
 601                     Ok(upload1_result.and(result))
 602                 })
 603             })
 604             .flatten()
 605             .and_then(move |_| {
 606                 let repeat = repeat2.load(Ordering::SeqCst);
 607                 let stream_len = stream_len2.load(Ordering::SeqCst);
 608                 let speed = ((stream_len*1000000)/(1024*1024))/(start_time.elapsed().as_micros() as usize);
 609                 println!("Uploaded {} chunks in {} seconds ({} MB/s).", repeat, start_time.elapsed().as_secs(), speed);
 610                 if repeat > 0 {
 611                     println!("Average chunk size was {} bytes.", stream_len/repeat);
 612                     println!("Time per request: {} microseconds.", (start_time.elapsed().as_micros())/(repeat as u128));
 613                 }
 614                 Ok((repeat, stream_len, speed))
 615             })
 616     }
 617
 618     pub fn upload_speedtest(&self) -> impl Future<Item=usize, Error=Error> {
 619
 620         let mut data = vec![];
 621         // generate pseudo random byte sequence
 622         for i in 0..1024*1024 {
 623             for j in 0..4 {
 624                 let byte = ((i >> (j<<3))&0xff) as u8;
 625                 data.push(byte);
 626             }
 627         }
 628
 629         let item_len = data.len();
 630
 631         let repeat = std::sync::Arc::new(AtomicUsize::new(0));
 632         let repeat2 = repeat.clone();
 633
 634         let (upload_queue, upload_result) = Self::response_queue();
 635
 636         let start_time = std::time::Instant::now();
 637
 638         let h2 = self.h2.clone();
 639
 640         futures::stream::repeat(data)
 641             .take_while(move |_| {
 642                 repeat.fetch_add(1, Ordering::SeqCst);
 643                 Ok(start_time.elapsed().as_secs() < 5)
 644             })
 645             .for_each(move |data| {
 646                 let h2 = h2.clone();
 647
 648                 let upload_queue = upload_queue.clone();
 649
 650                 println!("send test data ({} bytes)", data.len());
 651                 let request = H2Client::request_builder("localhost", "POST", "speedtest", None).unwrap();
 652                 h2.send_request(request, Some(bytes::Bytes::from(data)))
 653                     .and_then(move |response| {
 654                         upload_queue.send(response)
 655                             .map(|_| ()).map_err(Error::from)
 656                     })
 657             })
 658             .then(move |result| {
 659                 println!("RESULT {:?}", result);
 660                 upload_result.map_err(Error::from).and_then(|upload1_result| {
 661                     Ok(upload1_result.and(result))
 662                 })
 663             })
 664             .flatten()
 665             .and_then(move |_| {
 666                 let repeat = repeat2.load(Ordering::SeqCst);
 667                 println!("Uploaded {} chunks in {} seconds.", repeat, start_time.elapsed().as_secs());
 668                 let speed = ((item_len*1000000*(repeat as usize))/(1024*1024))/(start_time.elapsed().as_micros() as usize);
 669                 if repeat > 0 {
 670                     println!("Time per request: {} microseconds.", (start_time.elapsed().as_micros())/(repeat as u128));
 671                 }
 672                 Ok(speed)
 673             })
 674     }
 675 }
 676
 677 #[derive(Clone)]
 678 pub struct H2Client {
 679     h2: h2::client::SendRequest<bytes::Bytes>,
 680 }
 681
 682 impl H2Client {
 683
 684     pub fn new(h2: h2::client::SendRequest<bytes::Bytes>) -> Self {
 685         Self { h2 }
 686     }
 687
 688     pub fn get(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 689         let req = Self::request_builder("localhost", "GET", path, param).unwrap();
 690         self.request(req)
 691     }
 692
 693     pub fn put(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 694         let req = Self::request_builder("localhost", "PUT", path, param).unwrap();
 695         self.request(req)
 696     }
 697
 698     pub fn post(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 699         let req = Self::request_builder("localhost", "POST", path, param).unwrap();
 700         self.request(req)
 701     }
 702
 703     pub fn upload(&self, path: &str, param: Option<Value>, data: Vec<u8>) -> impl Future<Item=Value, Error=Error> {
 704         let request = Self::request_builder("localhost", "POST", path, param).unwrap();
 705
 706
 707         self.h2.clone()
 708             .ready()
 709             .map_err(Error::from)
 710             .and_then(move |mut send_request| {
 711                 let (response, stream) = send_request.send_request(request, false).unwrap();
 712                 PipeToSendStream::new(bytes::Bytes::from(data), stream)
 713                     .and_then(|_| {
 714                         response
 715                             .map_err(Error::from)
 716                             .and_then(Self::h2api_response)
 717                     })
 718             })
 719     }
 720
 721     fn request(
 722         &self,
 723         request: Request<()>,
 724     ) -> impl Future<Item=Value, Error=Error> {
 725
 726         self.send_request(request, None)
 727             .and_then(move |response| {
 728                 response
 729                     .map_err(Error::from)
 730                     .and_then(Self::h2api_response)
 731             })
 732     }
 733
 734     fn send_request(
 735         &self,
 736         request: Request<()>,
 737         data: Option<bytes::Bytes>,
 738     ) -> impl Future<Item=h2::client::ResponseFuture, Error=Error> {
 739
 740         self.h2.clone()
 741             .ready()
 742             .map_err(Error::from)
 743             .and_then(move |mut send_request| {
 744                 if let Some(data) = data {
 745                     let (response, stream) = send_request.send_request(request, false).unwrap();
 746                     future::Either::A(PipeToSendStream::new(bytes::Bytes::from(data), stream)
 747                         .and_then(move |_| {
 748                             future::ok(response)
 749                         }))
 750                 } else {
 751                     let (response, _stream) = send_request.send_request(request, true).unwrap();
 752                     future::Either::B(future::ok(response))
 753                 }
 754             })
 755     }
 756
 757     fn h2api_response(response: Response<h2::RecvStream>) -> impl Future<Item=Value, Error=Error> {
 758
 759         let status = response.status();
 760
 761         let (_head, mut body) = response.into_parts();
 762
 763         // The `release_capacity` handle allows the caller to manage
 764         // flow control.
 765         //
 766         // Whenever data is received, the caller is responsible for
 767         // releasing capacity back to the server once it has freed
 768         // the data from memory.
 769         let mut release_capacity = body.release_capacity().clone();
 770
 771         body
 772             .map(move |chunk| {
 773                 // Let the server send more data.
 774                 let _ = release_capacity.release_capacity(chunk.len());
 775                 chunk
 776             })
 777             .concat2()
 778             .map_err(Error::from)
 779             .and_then(move |data| {
 780                 let text = String::from_utf8(data.to_vec()).unwrap();
 781                 if status.is_success() {
 782                     if text.len() > 0 {
 783                         let mut value: Value = serde_json::from_str(&text)?;
 784                         if let Some(map) = value.as_object_mut() {
 785                             if let Some(data) = map.remove("data") {
 786                                 return Ok(data);
 787                             }
 788                         }
 789                         bail!("got result without data property");
 790                     } else {
 791                         Ok(Value::Null)
 792                     }
 793                 } else {
 794                     bail!("HTTP Error {}: {}", status, text);
 795                 }
 796             })
 797     }
 798
 799     pub fn request_builder(server: &str, method: &str, path: &str, data: Option<Value>) -> Result<Request<()>, Error> {
 800         let path = path.trim_matches('/');
 801         let url: Uri = format!("https://{}:8007/{}", server, path).parse()?;
 802
 803         if let Some(data) = data {
 804             let query = tools::json_object_to_query(data)?;
 805             let url: Uri = format!("https://{}:8007/{}?{}", server, path, query).parse()?;
 806             let request = Request::builder()
 807                 .method(method)
 808                 .uri(url)
 809                 .header("User-Agent", "proxmox-backup-client/1.0")
 810                 .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
 811                 .body(())?;
 812             return Ok(request);
 813         }
 814
 815         let request = Request::builder()
 816             .method(method)
 817             .uri(url)
 818             .header("User-Agent", "proxmox-backup-client/1.0")
 819             .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
 820             .body(())?;
 821
 822         Ok(request)
 823     }
 824 }