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