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