]> git.proxmox.com Git - proxmox-backup.git/blame - src/client/http_client.rs
tree-wide: use the new vec/io tools modules
[proxmox-backup.git] / src / client / http_client.rs
CommitLineData
597641fd
DM
1use failure::*;
2
3use http::Uri;
4use hyper::Body;
5use hyper::client::Client;
ba3a60b2
DM
6use xdg::BaseDirectories;
7use chrono::Utc;
553610b4
DM
8use std::collections::HashSet;
9use std::sync::{Arc, Mutex};
597641fd 10
b57cb264 11use http::{Request, Response};
5a2df000
DM
12use http::header::HeaderValue;
13
82ab7230 14use futures::*;
1fdb4c6f 15use futures::stream::Stream;
82ab7230
DM
16use std::sync::atomic::{AtomicUsize, Ordering};
17use tokio::sync::mpsc;
1fdb4c6f 18
ba3a60b2 19use serde_json::{json, Value};
0dffe3f9 20use url::percent_encoding::{percent_encode, DEFAULT_ENCODE_SET};
1fdb4c6f 21
5a2df000 22use crate::tools::{self, BroadcastFuture, tty};
e3dbd41b 23use super::pipe_to_stream::*;
5a2df000
DM
24
25#[derive(Clone)]
26struct AuthInfo {
27 username: String,
28 ticket: String,
29 token: String,
30}
56458d97 31
151c6ce2 32/// HTTP(S) API client
597641fd 33pub struct HttpClient {
5a2df000 34 client: Client<hyper_tls::HttpsConnector<hyper::client::HttpConnector>>,
597641fd 35 server: String,
5a2df000 36 auth: BroadcastFuture<AuthInfo>,
597641fd
DM
37}
38
ba3a60b2
DM
39fn 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
49cf9f3d 48 let mut data = tools::file_get_json(&path, Some(json!({})))?;
ba3a60b2
DM
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
75fn 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
49cf9f3d
DM
87 let data = match tools::file_get_json(&path, None) {
88 Ok(v) => v,
89 _ => return None,
90 };
ba3a60b2
DM
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,
21ea0158 107 };
ba3a60b2
DM
108 return Some((ticket.to_owned(), token.to_owned()));
109 }
110 }
111 }
112
113 None
114}
115
597641fd
DM
116impl HttpClient {
117
45cdce06 118 pub fn new(server: &str, username: &str) -> Result<Self, Error> {
5a2df000 119 let client = Self::build_client();
5a2df000 120
45cdce06
DM
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 {
5a2df000 130 client,
597641fd 131 server: String::from(server),
5a2df000 132 auth: BroadcastFuture::new(login),
45cdce06 133 })
597641fd
DM
134 }
135
5a2df000 136 fn get_password(_username: &str) -> Result<String, Error> {
56458d97
WB
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
5a2df000 154 fn build_client() -> Client<hyper_tls::HttpsConnector<hyper::client::HttpConnector>> {
4a3f6517
WB
155 let mut builder = native_tls::TlsConnector::builder();
156 // FIXME: We need a CLI option for this!
157 builder.danger_accept_invalid_certs(true);
5a2df000 158 let tlsconnector = builder.build().unwrap();
4a3f6517 159 let mut httpc = hyper::client::HttpConnector::new(1);
9c9ad941 160 //httpc.set_nodelay(true); // not sure if this help?
4a3f6517
WB
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!
adec8ea2
DM
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)
a6b75513
DM
168 }
169
5a2df000 170 pub fn request(&self, mut req: Request<Body>) -> impl Future<Item=Value, Error=Error> {
597641fd 171
5a2df000 172 let login = self.auth.listen();
597641fd 173
5a2df000 174 let client = self.client.clone();
597641fd 175
5a2df000 176 login.and_then(move |auth| {
597641fd 177
5a2df000
DM
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());
597641fd 181
5a2df000 182 let request = Self::api_request(client, req);
597641fd 183
5a2df000
DM
184 request
185 })
1fdb4c6f
DM
186 }
187
9e391bb7 188 pub fn get(&self, path: &str, data: Option<Value>) -> impl Future<Item=Value, Error=Error> {
a6b75513 189
9e391bb7 190 let req = Self::request_builder(&self.server, "GET", path, data).unwrap();
5a2df000 191 self.request(req)
a6b75513
DM
192 }
193
9e391bb7 194 pub fn delete(&mut self, path: &str, data: Option<Value>) -> impl Future<Item=Value, Error=Error> {
a6b75513 195
9e391bb7 196 let req = Self::request_builder(&self.server, "DELETE", path, data).unwrap();
5a2df000 197 self.request(req)
a6b75513
DM
198 }
199
5a2df000 200 pub fn post(&mut self, path: &str, data: Option<Value>) -> impl Future<Item=Value, Error=Error> {
024f11bb 201
5a2df000
DM
202 let req = Self::request_builder(&self.server, "POST", path, data).unwrap();
203 self.request(req)
024f11bb
DM
204 }
205
5a2df000 206 pub fn download(&mut self, path: &str, mut output: Box<dyn std::io::Write + Send>) -> impl Future<Item=(), Error=Error> {
024f11bb 207
5a2df000 208 let mut req = Self::request_builder(&self.server, "GET", path, None).unwrap();
024f11bb 209
5a2df000 210 let login = self.auth.listen();
024f11bb 211
5a2df000 212 let client = self.client.clone();
1fdb4c6f 213
5a2df000 214 login.and_then(move |auth| {
81da38c1 215
5a2df000
DM
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());
6f62c924 218
5a2df000
DM
219 client.request(req)
220 .map_err(Error::from)
221 .and_then(|resp| {
6f62c924 222
5a2df000 223 let _status = resp.status(); // fixme: ??
6f62c924 224
5a2df000
DM
225 resp.into_body()
226 .map_err(Error::from)
227 .for_each(move |chunk| {
228 output.write_all(&chunk)?;
229 Ok(())
230 })
6f62c924 231
5a2df000
DM
232 })
233 })
6f62c924
DM
234 }
235
5a2df000 236 pub fn upload(&mut self, content_type: &str, body: Body, path: &str) -> impl Future<Item=Value, Error=Error> {
81da38c1
DM
237
238 let path = path.trim_matches('/');
5a2df000 239 let url: Uri = format!("https://{}:8007/{}", &self.server, path).parse().unwrap();
81da38c1 240
5a2df000 241 let req = Request::builder()
81da38c1
DM
242 .method("POST")
243 .uri(url)
244 .header("User-Agent", "proxmox-backup-client/1.0")
5a2df000
DM
245 .header("Content-Type", content_type)
246 .body(body).unwrap();
81da38c1 247
5a2df000 248 self.request(req)
1fdb4c6f
DM
249 }
250
6ab34afa
DM
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> {
cf639a47 257
6ab34afa
DM
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();
cf639a47
DM
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 {
9af37c8f
DM
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))
cf639a47 281 }
cf639a47 282 })
cf639a47 283 .and_then(|upgraded| {
cf639a47
DM
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.
850ac6d0 294 h2.ready()
6ab34afa 295 .map(BackupClient::new)
850ac6d0 296 .map_err(Error::from)
cf639a47
DM
297 })
298 })
299 }
300
5a2df000
DM
301 fn credentials(
302 client: Client<hyper_tls::HttpsConnector<hyper::client::HttpConnector>>,
45cdce06
DM
303 server: String,
304 username: String,
305 password: String,
5a2df000 306 ) -> Box<Future<Item=AuthInfo, Error=Error> + Send> {
0ffbccce 307
45cdce06 308 let server2 = server.clone();
0ffbccce 309
5a2df000 310 let create_request = futures::future::lazy(move || {
45cdce06 311 let data = json!({ "username": username, "password": password });
5a2df000 312 let req = Self::request_builder(&server, "POST", "/api2/json/access/ticket", Some(data)).unwrap();
45cdce06 313 Self::api_request(client, req)
5a2df000 314 });
0dffe3f9 315
5a2df000
DM
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 };
0dffe3f9 323
5a2df000 324 let _ = store_ticket_info(&server2, &auth.username, &auth.ticket, &auth.token);
0dffe3f9 325
5a2df000
DM
326 Ok(auth)
327 });
0dffe3f9 328
5a2df000 329 Box::new(login_future)
ba3a60b2
DM
330 }
331
d2c48afc
DM
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
5a2df000
DM
356 fn api_request(
357 client: Client<hyper_tls::HttpsConnector<hyper::client::HttpConnector>>,
358 req: Request<Body>
359 ) -> impl Future<Item=Value, Error=Error> {
ba3a60b2 360
5a2df000
DM
361 client.request(req)
362 .map_err(Error::from)
d2c48afc 363 .and_then(Self::api_response)
0dffe3f9
DM
364 }
365
5a2df000 366 pub fn request_builder(server: &str, method: &str, path: &str, data: Option<Value>) -> Result<Request<Body>, Error> {
591f570b 367 let path = path.trim_matches('/');
5a2df000
DM
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 {
9e391bb7
DM
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);
5a2df000 389 }
5a2df000 390 }
0dffe3f9 391
1fdb4c6f 392 let request = Request::builder()
5a2df000 393 .method(method)
1fdb4c6f
DM
394 .uri(url)
395 .header("User-Agent", "proxmox-backup-client/1.0")
5a2df000
DM
396 .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
397 .body(Body::empty())?;
1fdb4c6f 398
5a2df000 399 Ok(request)
597641fd
DM
400 }
401}
b57cb264 402
6ab34afa
DM
403//#[derive(Clone)]
404pub struct BackupClient {
9af37c8f 405 h2: H2Client,
b57cb264
DM
406}
407
6ab34afa 408impl BackupClient {
b57cb264
DM
409
410 pub fn new(h2: h2::client::SendRequest<bytes::Bytes>) -> Self {
9af37c8f 411 Self { h2: H2Client::new(h2) }
b57cb264
DM
412 }
413
414 pub fn get(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
9af37c8f 415 self.h2.get(path, param)
b57cb264
DM
416 }
417
82ab7230 418 pub fn put(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
9af37c8f 419 self.h2.put(path, param)
82ab7230
DM
420 }
421
b57cb264 422 pub fn post(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
9af37c8f 423 self.h2.post(path, param)
97f22ce5
DM
424 }
425
d6f204ed
DM
426 pub fn finish(&self) -> impl Future<Item=(), Error=Error> {
427 self.h2.clone().post("finish", None).map(|_| ())
428 }
429
430 pub fn upload_dynamic_stream(
431 &self,
432 archive_name: &str,
433 stream: impl Stream<Item=bytes::BytesMut, Error=Error>,
434 ) -> impl Future<Item=(), Error=Error> {
435
436 let known_chunks = Arc::new(Mutex::new(HashSet::new()));
437
438 let h2 = self.h2.clone();
439 let h2_2 = self.h2.clone();
440 let h2_3 = self.h2.clone();
441 let h2_4 = self.h2.clone();
442
443 let param = json!({ "archive-name": archive_name });
444
445 Self::download_chunk_list(h2, "dynamic_index", archive_name, known_chunks.clone())
446 .and_then(move |_| {
447 h2_2.post("dynamic_index", Some(param))
448 })
449 .and_then(move |res| {
450 let wid = res.as_u64().unwrap();
451 Self::upload_stream(h2_3, wid, stream, known_chunks.clone())
8bea85b4
DM
452 .and_then(move |(chunk_count, size, _speed)| {
453 let param = json!({
454 "wid": wid ,
455 "chunk-count": chunk_count,
456 "size": size,
457 });
458 h2_4.post("dynamic_close", Some(param))
d6f204ed
DM
459 })
460 .map(|_| ())
461 })
462 }
463
6ab34afa 464 fn response_queue() -> (
82ab7230
DM
465 mpsc::Sender<h2::client::ResponseFuture>,
466 sync::oneshot::Receiver<Result<(), Error>>
467 ) {
468 let (verify_queue_tx, verify_queue_rx) = mpsc::channel(100);
469 let (verify_result_tx, verify_result_rx) = sync::oneshot::channel();
adec8ea2 470
82ab7230
DM
471 hyper::rt::spawn(
472 verify_queue_rx
473 .map_err(Error::from)
474 .for_each(|response: h2::client::ResponseFuture| {
475 response
476 .map_err(Error::from)
9af37c8f 477 .and_then(H2Client::h2api_response)
82ab7230
DM
478 .and_then(|result| {
479 println!("RESPONSE: {:?}", result);
480 Ok(())
481 })
482 .map_err(|err| format_err!("pipelined request failed: {}", err))
483 })
484 .then(|result|
485 verify_result_tx.send(result)
486 )
487 .map_err(|_| { /* ignore closed channel */ })
488 );
adec8ea2 489
82ab7230
DM
490 (verify_queue_tx, verify_result_rx)
491 }
492
6ab34afa 493 fn download_chunk_list(
9af37c8f 494 h2: H2Client,
553610b4
DM
495 path: &str,
496 archive_name: &str,
497 known_chunks: Arc<Mutex<HashSet<[u8;32]>>>,
498 ) -> impl Future<Item=(), Error=Error> {
499
500 let param = json!({ "archive-name": archive_name });
9af37c8f 501 let request = H2Client::request_builder("localhost", "GET", path, Some(param)).unwrap();
553610b4 502
9af37c8f 503 h2.send_request(request, None)
553610b4
DM
504 .and_then(move |response| {
505 response
506 .map_err(Error::from)
507 .and_then(move |resp| {
508 let status = resp.status();
509 if !status.is_success() {
510 bail!("download chunk list failed with status {}", status);
511 }
512
513 let (_head, body) = resp.into_parts();
514
515 Ok(body)
516 })
517 .and_then(move |mut body| {
518
519 let mut release_capacity = body.release_capacity().clone();
520
521 crate::backup::DigestListDecoder::new(body.map_err(Error::from))
522 .for_each(move |chunk| {
523 let _ = release_capacity.release_capacity(chunk.len());
524 println!("GOT DOWNLOAD {}", tools::digest_to_hex(&chunk));
525 known_chunks.lock().unwrap().insert(chunk);
526 Ok(())
527 })
528 })
529 })
530 }
531
6ab34afa 532 fn upload_stream(
9af37c8f 533 h2: H2Client,
82ab7230 534 wid: u64,
553610b4
DM
535 stream: impl Stream<Item=bytes::BytesMut, Error=Error>,
536 known_chunks: Arc<Mutex<HashSet<[u8;32]>>>,
8bea85b4 537 ) -> impl Future<Item=(usize, usize, usize), Error=Error> {
adec8ea2 538
82ab7230
DM
539 let repeat = std::sync::Arc::new(AtomicUsize::new(0));
540 let repeat2 = repeat.clone();
adec8ea2 541
82ab7230
DM
542 let stream_len = std::sync::Arc::new(AtomicUsize::new(0));
543 let stream_len2 = stream_len.clone();
9c9ad941 544
6ab34afa 545 let (upload_queue, upload_result) = Self::response_queue();
9c9ad941 546
82ab7230 547 let start_time = std::time::Instant::now();
9c9ad941 548
82ab7230
DM
549 stream
550 .for_each(move |data| {
6ab34afa
DM
551 let h2 = h2.clone();
552
82ab7230
DM
553 repeat.fetch_add(1, Ordering::SeqCst);
554 stream_len.fetch_add(data.len(), Ordering::SeqCst);
9c9ad941 555
82ab7230 556 let upload_queue = upload_queue.clone();
adec8ea2 557
82ab7230
DM
558 let digest = openssl::sha::sha256(&data);
559
553610b4
DM
560 let mut known_chunks = known_chunks.lock().unwrap();
561 let chunk_is_known = known_chunks.contains(&digest);
82ab7230
DM
562
563 let upload_data;
564 let request;
565
82ab7230 566 if chunk_is_known {
553610b4 567 println!("append existing chunk ({} bytes)", data.len());
82ab7230 568 let param = json!({ "wid": wid, "digest": tools::digest_to_hex(&digest) });
9af37c8f 569 request = H2Client::request_builder("localhost", "PUT", "dynamic_index", Some(param)).unwrap();
82ab7230
DM
570 upload_data = None;
571 } else {
553610b4
DM
572 println!("upload new chunk {} ({} bytes)", tools::digest_to_hex(&digest), data.len());
573 known_chunks.insert(digest);
82ab7230 574 let param = json!({ "wid": wid, "size" : data.len() });
9af37c8f 575 request = H2Client::request_builder("localhost", "POST", "dynamic_chunk", Some(param)).unwrap();
82ab7230
DM
576 upload_data = Some(bytes::Bytes::from(data));
577 }
578
9af37c8f 579 h2.send_request(request, upload_data)
82ab7230
DM
580 .and_then(move |response| {
581 upload_queue.send(response)
582 .map(|_| ()).map_err(Error::from)
9c9ad941 583 })
82ab7230
DM
584 })
585 .then(move |result| {
586 println!("RESULT {:?}", result);
587 upload_result.map_err(Error::from).and_then(|upload1_result| {
588 Ok(upload1_result.and(result))
589 })
590 })
591 .flatten()
592 .and_then(move |_| {
593 let repeat = repeat2.load(Ordering::SeqCst);
594 let stream_len = stream_len2.load(Ordering::SeqCst);
595 let speed = ((stream_len*1000000)/(1024*1024))/(start_time.elapsed().as_micros() as usize);
596 println!("Uploaded {} chunks in {} seconds ({} MB/s).", repeat, start_time.elapsed().as_secs(), speed);
597 if repeat > 0 {
598 println!("Average chunk size was {} bytes.", stream_len/repeat);
599 println!("Time per request: {} microseconds.", (start_time.elapsed().as_micros())/(repeat as u128));
600 }
8bea85b4 601 Ok((repeat, stream_len, speed))
82ab7230
DM
602 })
603 }
604
605 pub fn upload_speedtest(&self) -> impl Future<Item=usize, Error=Error> {
606
607 let mut data = vec![];
608 // generate pseudo random byte sequence
609 for i in 0..1024*1024 {
610 for j in 0..4 {
611 let byte = ((i >> (j<<3))&0xff) as u8;
612 data.push(byte);
613 }
614 }
615
616 let item_len = data.len();
617
618 let repeat = std::sync::Arc::new(AtomicUsize::new(0));
619 let repeat2 = repeat.clone();
620
6ab34afa 621 let (upload_queue, upload_result) = Self::response_queue();
82ab7230
DM
622
623 let start_time = std::time::Instant::now();
624
6ab34afa 625 let h2 = self.h2.clone();
82ab7230
DM
626
627 futures::stream::repeat(data)
628 .take_while(move |_| {
629 repeat.fetch_add(1, Ordering::SeqCst);
630 Ok(start_time.elapsed().as_secs() < 5)
631 })
632 .for_each(move |data| {
6ab34afa 633 let h2 = h2.clone();
82ab7230
DM
634
635 let upload_queue = upload_queue.clone();
636
637 println!("send test data ({} bytes)", data.len());
9af37c8f
DM
638 let request = H2Client::request_builder("localhost", "POST", "speedtest", None).unwrap();
639 h2.send_request(request, Some(bytes::Bytes::from(data)))
82ab7230
DM
640 .and_then(move |response| {
641 upload_queue.send(response)
642 .map(|_| ()).map_err(Error::from)
adec8ea2
DM
643 })
644 })
82ab7230
DM
645 .then(move |result| {
646 println!("RESULT {:?}", result);
647 upload_result.map_err(Error::from).and_then(|upload1_result| {
648 Ok(upload1_result.and(result))
649 })
650 })
651 .flatten()
652 .and_then(move |_| {
653 let repeat = repeat2.load(Ordering::SeqCst);
654 println!("Uploaded {} chunks in {} seconds.", repeat, start_time.elapsed().as_secs());
655 let speed = ((item_len*1000000*(repeat as usize))/(1024*1024))/(start_time.elapsed().as_micros() as usize);
656 if repeat > 0 {
657 println!("Time per request: {} microseconds.", (start_time.elapsed().as_micros())/(repeat as u128));
658 }
659 Ok(speed)
660 })
adec8ea2 661 }
9af37c8f
DM
662}
663
664#[derive(Clone)]
665pub struct H2Client {
666 h2: h2::client::SendRequest<bytes::Bytes>,
667}
668
669impl H2Client {
670
671 pub fn new(h2: h2::client::SendRequest<bytes::Bytes>) -> Self {
672 Self { h2 }
673 }
674
675 pub fn get(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
676 let req = Self::request_builder("localhost", "GET", path, param).unwrap();
677 self.request(req)
678 }
679
680 pub fn put(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
681 let req = Self::request_builder("localhost", "PUT", path, param).unwrap();
682 self.request(req)
683 }
684
685 pub fn post(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
686 let req = Self::request_builder("localhost", "POST", path, param).unwrap();
687 self.request(req)
688 }
689
690 pub fn upload(&self, path: &str, param: Option<Value>, data: Vec<u8>) -> impl Future<Item=Value, Error=Error> {
691 let request = Self::request_builder("localhost", "POST", path, param).unwrap();
692
693
694 self.h2.clone()
695 .ready()
696 .map_err(Error::from)
697 .and_then(move |mut send_request| {
698 let (response, stream) = send_request.send_request(request, false).unwrap();
699 PipeToSendStream::new(bytes::Bytes::from(data), stream)
700 .and_then(|_| {
701 response
702 .map_err(Error::from)
703 .and_then(Self::h2api_response)
704 })
705 })
706 }
adec8ea2 707
b57cb264 708 fn request(
9af37c8f 709 &self,
b57cb264
DM
710 request: Request<()>,
711 ) -> impl Future<Item=Value, Error=Error> {
712
9af37c8f 713 self.send_request(request, None)
82ab7230
DM
714 .and_then(move |response| {
715 response
716 .map_err(Error::from)
717 .and_then(Self::h2api_response)
718 })
719 }
720
721 fn send_request(
9af37c8f 722 &self,
82ab7230
DM
723 request: Request<()>,
724 data: Option<bytes::Bytes>,
725 ) -> impl Future<Item=h2::client::ResponseFuture, Error=Error> {
726
9af37c8f 727 self.h2.clone()
10130cf4
DM
728 .ready()
729 .map_err(Error::from)
730 .and_then(move |mut send_request| {
82ab7230
DM
731 if let Some(data) = data {
732 let (response, stream) = send_request.send_request(request, false).unwrap();
733 future::Either::A(PipeToSendStream::new(bytes::Bytes::from(data), stream)
734 .and_then(move |_| {
735 future::ok(response)
736 }))
737 } else {
738 let (response, _stream) = send_request.send_request(request, true).unwrap();
739 future::Either::B(future::ok(response))
740 }
b57cb264
DM
741 })
742 }
743
744 fn h2api_response(response: Response<h2::RecvStream>) -> impl Future<Item=Value, Error=Error> {
745
746 let status = response.status();
747
748 let (_head, mut body) = response.into_parts();
749
750 // The `release_capacity` handle allows the caller to manage
751 // flow control.
752 //
753 // Whenever data is received, the caller is responsible for
754 // releasing capacity back to the server once it has freed
755 // the data from memory.
756 let mut release_capacity = body.release_capacity().clone();
757
758 body
759 .map(move |chunk| {
b57cb264
DM
760 // Let the server send more data.
761 let _ = release_capacity.release_capacity(chunk.len());
762 chunk
763 })
764 .concat2()
765 .map_err(Error::from)
766 .and_then(move |data| {
b57cb264
DM
767 let text = String::from_utf8(data.to_vec()).unwrap();
768 if status.is_success() {
769 if text.len() > 0 {
770 let mut value: Value = serde_json::from_str(&text)?;
771 if let Some(map) = value.as_object_mut() {
772 if let Some(data) = map.remove("data") {
773 return Ok(data);
774 }
775 }
776 bail!("got result without data property");
777 } else {
778 Ok(Value::Null)
779 }
780 } else {
781 bail!("HTTP Error {}: {}", status, text);
782 }
783 })
784 }
785
786 pub fn request_builder(server: &str, method: &str, path: &str, data: Option<Value>) -> Result<Request<()>, Error> {
787 let path = path.trim_matches('/');
788 let url: Uri = format!("https://{}:8007/{}", server, path).parse()?;
789
790 if let Some(data) = data {
791 let query = tools::json_object_to_query(data)?;
792 let url: Uri = format!("https://{}:8007/{}?{}", server, path, query).parse()?;
793 let request = Request::builder()
794 .method(method)
795 .uri(url)
796 .header("User-Agent", "proxmox-backup-client/1.0")
797 .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
798 .body(())?;
799 return Ok(request);
800 }
801
802 let request = Request::builder()
803 .method(method)
804 .uri(url)
805 .header("User-Agent", "proxmox-backup-client/1.0")
806 .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
807 .body(())?;
808
809 Ok(request)
810 }
811}