5 use hyper
::client
::Client
;
6 use xdg
::BaseDirectories
;
8 use std
::collections
::HashSet
;
9 use std
::sync
::{Arc, Mutex}
;
12 use http
::{Request, Response}
;
13 use http
::header
::HeaderValue
;
16 use futures
::stream
::Stream
;
17 use std
::sync
::atomic
::{AtomicUsize, Ordering}
;
18 use tokio
::sync
::mpsc
;
20 use serde_json
::{json, Value}
;
21 use url
::percent_encoding
::{percent_encode, DEFAULT_ENCODE_SET}
;
23 use crate::tools
::{self, BroadcastFuture, tty}
;
24 use crate::tools
::futures
::{cancellable, Canceller}
;
25 use super::pipe_to_stream
::*;
26 use super::merge_known_chunks
::*;
38 /// HTTP(S) API client
39 pub struct HttpClient
{
40 client
: Client
<hyper_tls
::HttpsConnector
<hyper
::client
::HttpConnector
>>,
42 auth
: BroadcastFuture
<AuthInfo
>,
45 fn store_ticket_info(server
: &str, username
: &str, ticket
: &str, token
: &str) -> Result
<(), Error
> {
47 let base
= BaseDirectories
::with_prefix("proxmox-backup")?
;
49 // usually /run/user/<uid>/...
50 let path
= base
.place_runtime_file("tickets")?
;
52 let mode
= nix
::sys
::stat
::Mode
::from_bits_truncate(0o0600);
54 let mut data
= tools
::file_get_json(&path
, Some(json
!({}
)))?
;
56 let now
= Utc
::now().timestamp();
58 data
[server
][username
] = json
!({ "timestamp": now, "ticket": ticket, "token": token}
);
60 let mut new_data
= json
!({}
);
62 let ticket_lifetime
= tools
::ticket
::TICKET_LIFETIME
- 60;
64 let empty
= serde_json
::map
::Map
::new();
65 for (server
, info
) in data
.as_object().unwrap_or(&empty
) {
66 for (_user
, uinfo
) in info
.as_object().unwrap_or(&empty
) {
67 if let Some(timestamp
) = uinfo
["timestamp"].as_i64() {
68 let age
= now
- timestamp
;
69 if age
< ticket_lifetime
{
70 new_data
[server
][username
] = uinfo
.clone();
76 tools
::file_set_contents(path
, new_data
.to_string().as_bytes(), Some(mode
))?
;
81 fn load_ticket_info(server
: &str, username
: &str) -> Option
<(String
, String
)> {
82 let base
= match BaseDirectories
::with_prefix("proxmox-backup") {
87 // usually /run/user/<uid>/...
88 let path
= match base
.place_runtime_file("tickets") {
93 let data
= match tools
::file_get_json(&path
, None
) {
98 let now
= Utc
::now().timestamp();
100 let ticket_lifetime
= tools
::ticket
::TICKET_LIFETIME
- 60;
102 if let Some(uinfo
) = data
[server
][username
].as_object() {
103 if let Some(timestamp
) = uinfo
["timestamp"].as_i64() {
104 let age
= now
- timestamp
;
105 if age
< ticket_lifetime
{
106 let ticket
= match uinfo
["ticket"].as_str() {
110 let token
= match uinfo
["token"].as_str() {
114 return Some((ticket
.to_owned(), token
.to_owned()));
124 pub fn new(server
: &str, username
: &str) -> Result
<Self, Error
> {
125 let client
= Self::build_client();
127 let password
= if let Some((ticket
, _token
)) = load_ticket_info(server
, username
) {
130 Self::get_password(&username
)?
133 let login
= Self::credentials(client
.clone(), server
.to_owned(), username
.to_owned(), password
);
137 server
: String
::from(server
),
138 auth
: BroadcastFuture
::new(login
),
142 fn get_password(_username
: &str) -> Result
<String
, Error
> {
143 use std
::env
::VarError
::*;
144 match std
::env
::var("PBS_PASSWORD") {
145 Ok(p
) => return Ok(p
),
146 Err(NotUnicode(_
)) => bail
!("PBS_PASSWORD contains bad characters"),
148 // Try another method
152 // If we're on a TTY, query the user for a password
153 if tty
::stdin_isatty() {
154 return Ok(String
::from_utf8(tty
::read_password("Password: ")?
)?
);
157 bail
!("no password input mechanism available");
160 fn build_client() -> Client
<hyper_tls
::HttpsConnector
<hyper
::client
::HttpConnector
>> {
161 let mut builder
= native_tls
::TlsConnector
::builder();
162 // FIXME: We need a CLI option for this!
163 builder
.danger_accept_invalid_certs(true);
164 let tlsconnector
= builder
.build().unwrap();
165 let mut httpc
= hyper
::client
::HttpConnector
::new(1);
166 //httpc.set_nodelay(true); // not sure if this help?
167 httpc
.enforce_http(false); // we want https...
168 let mut https
= hyper_tls
::HttpsConnector
::from((httpc
, tlsconnector
));
169 https
.https_only(true); // force it!
171 //.http2_initial_stream_window_size( (1 << 31) - 2)
172 //.http2_initial_connection_window_size( (1 << 31) - 2)
173 .build
::<_
, Body
>(https
)
176 pub fn request(&self, mut req
: Request
<Body
>) -> impl Future
<Item
=Value
, Error
=Error
> {
178 let login
= self.auth
.listen();
180 let client
= self.client
.clone();
182 login
.and_then(move |auth
| {
184 let enc_ticket
= format
!("PBSAuthCookie={}", percent_encode(auth
.ticket
.as_bytes(), DEFAULT_ENCODE_SET
));
185 req
.headers_mut().insert("Cookie", HeaderValue
::from_str(&enc_ticket
).unwrap());
186 req
.headers_mut().insert("CSRFPreventionToken", HeaderValue
::from_str(&auth
.token
).unwrap());
188 let request
= Self::api_request(client
, req
);
194 pub fn get(&self, path
: &str, data
: Option
<Value
>) -> impl Future
<Item
=Value
, Error
=Error
> {
196 let req
= Self::request_builder(&self.server
, "GET", path
, data
).unwrap();
200 pub fn delete(&mut self, path
: &str, data
: Option
<Value
>) -> impl Future
<Item
=Value
, Error
=Error
> {
202 let req
= Self::request_builder(&self.server
, "DELETE", path
, data
).unwrap();
206 pub fn post(&mut self, path
: &str, data
: Option
<Value
>) -> impl Future
<Item
=Value
, Error
=Error
> {
208 let req
= Self::request_builder(&self.server
, "POST", path
, data
).unwrap();
212 pub fn download
<W
: Write
>(&mut self, path
: &str, output
: W
) -> impl Future
<Item
=W
, Error
=Error
> {
214 let mut req
= Self::request_builder(&self.server
, "GET", path
, None
).unwrap();
216 let login
= self.auth
.listen();
218 let client
= self.client
.clone();
220 login
.and_then(move |auth
| {
222 let enc_ticket
= format
!("PBSAuthCookie={}", percent_encode(auth
.ticket
.as_bytes(), DEFAULT_ENCODE_SET
));
223 req
.headers_mut().insert("Cookie", HeaderValue
::from_str(&enc_ticket
).unwrap());
226 .map_err(Error
::from
)
228 let status
= resp
.status();
229 if !status
.is_success() {
231 HttpClient
::api_response(resp
)
232 .and_then(|_
| { bail!("unknown error"); }
)
237 .map_err(Error
::from
)
238 .fold(output
, move |mut acc
, chunk
| {
239 acc
.write_all(&chunk
)?
;
248 pub fn upload(&mut self, content_type
: &str, body
: Body
, path
: &str) -> impl Future
<Item
=Value
, Error
=Error
> {
250 let path
= path
.trim_matches('
/'
);
251 let url
: Uri
= format
!("https://{}:8007/{}", &self.server
, path
).parse().unwrap();
253 let req
= Request
::builder()
256 .header("User-Agent", "proxmox-backup-client/1.0")
257 .header("Content-Type", content_type
)
258 .body(body
).unwrap();
269 ) -> impl Future
<Item
=BackupClient
, Error
=Error
> {
271 let param
= json
!({"backup-type": backup_type, "backup-id": backup_id, "store": datastore, "debug": debug}
);
272 let mut req
= Self::request_builder(&self.server
, "GET", "/api2/json/backup", Some(param
)).unwrap();
274 let login
= self.auth
.listen();
276 let client
= self.client
.clone();
278 login
.and_then(move |auth
| {
280 let enc_ticket
= format
!("PBSAuthCookie={}", percent_encode(auth
.ticket
.as_bytes(), DEFAULT_ENCODE_SET
));
281 req
.headers_mut().insert("Cookie", HeaderValue
::from_str(&enc_ticket
).unwrap());
282 req
.headers_mut().insert("UPGRADE", HeaderValue
::from_str(PROXMOX_BACKUP_PROTOCOL_ID_V1
!()).unwrap());
285 .map_err(Error
::from
)
288 let status
= resp
.status();
289 if status
!= http
::StatusCode
::SWITCHING_PROTOCOLS
{
290 future
::Either
::A(Self::api_response(resp
).and_then(|_
| { bail!("unknown error"); }
))
292 future
::Either
::B(resp
.into_body().on_upgrade().map_err(Error
::from
))
295 .and_then(|upgraded
| {
296 h2
::client
::handshake(upgraded
).map_err(Error
::from
)
298 .and_then(|(h2
, connection
)| {
299 let connection
= connection
300 .map_err(|_
| panic
!("HTTP/2.0 connection failed"));
302 let (connection
, canceller
) = cancellable(connection
)?
;
303 // A cancellable future returns an Option which is None when cancelled and
304 // Some when it finished instead, since we don't care about the return type we
305 // need to map it away:
306 let connection
= connection
.map(|_
| ());
308 // Spawn a new task to drive the connection state
309 hyper
::rt
::spawn(connection
);
311 // Wait until the `SendRequest` handle has available capacity.
313 .map(move |c
| BackupClient
::new(c
, canceller
))
314 .map_err(Error
::from
))
321 client
: Client
<hyper_tls
::HttpsConnector
<hyper
::client
::HttpConnector
>>,
325 ) -> Box
<dyn Future
<Item
=AuthInfo
, Error
=Error
> + Send
> {
327 let server2
= server
.clone();
329 let create_request
= futures
::future
::lazy(move || {
330 let data
= json
!({ "username": username, "password": password }
);
331 let req
= Self::request_builder(&server
, "POST", "/api2/json/access/ticket", Some(data
)).unwrap();
332 Self::api_request(client
, req
)
335 let login_future
= create_request
336 .and_then(move |cred
| {
337 let auth
= AuthInfo
{
338 username
: cred
["data"]["username"].as_str().unwrap().to_owned(),
339 ticket
: cred
["data"]["ticket"].as_str().unwrap().to_owned(),
340 token
: cred
["data"]["CSRFPreventionToken"].as_str().unwrap().to_owned(),
343 let _
= store_ticket_info(&server2
, &auth
.username
, &auth
.ticket
, &auth
.token
);
348 Box
::new(login_future
)
351 fn api_response(response
: Response
<Body
>) -> impl Future
<Item
=Value
, Error
=Error
> {
353 let status
= response
.status();
358 .map_err(Error
::from
)
359 .and_then(move |data
| {
361 let text
= String
::from_utf8(data
.to_vec()).unwrap();
362 if status
.is_success() {
364 let value
: Value
= serde_json
::from_str(&text
)?
;
370 bail
!("HTTP Error {}: {}", status
, text
);
376 client
: Client
<hyper_tls
::HttpsConnector
<hyper
::client
::HttpConnector
>>,
378 ) -> impl Future
<Item
=Value
, Error
=Error
> {
381 .map_err(Error
::from
)
382 .and_then(Self::api_response
)
385 pub fn request_builder(server
: &str, method
: &str, path
: &str, data
: Option
<Value
>) -> Result
<Request
<Body
>, Error
> {
386 let path
= path
.trim_matches('
/'
);
387 let url
: Uri
= format
!("https://{}:8007/{}", server
, path
).parse()?
;
389 if let Some(data
) = data
{
390 if method
== "POST" {
391 let request
= Request
::builder()
394 .header("User-Agent", "proxmox-backup-client/1.0")
395 .header(hyper
::header
::CONTENT_TYPE
, "application/json")
396 .body(Body
::from(data
.to_string()))?
;
399 let query
= tools
::json_object_to_query(data
)?
;
400 let url
: Uri
= format
!("https://{}:8007/{}?{}", server
, path
, query
).parse()?
;
401 let request
= Request
::builder()
404 .header("User-Agent", "proxmox-backup-client/1.0")
405 .header(hyper
::header
::CONTENT_TYPE
, "application/x-www-form-urlencoded")
406 .body(Body
::empty())?
;
411 let request
= Request
::builder()
414 .header("User-Agent", "proxmox-backup-client/1.0")
415 .header(hyper
::header
::CONTENT_TYPE
, "application/x-www-form-urlencoded")
416 .body(Body
::empty())?
;
423 pub struct BackupClient
{
425 canceller
: Option
<Canceller
>,
431 pub fn new(h2
: h2
::client
::SendRequest
<bytes
::Bytes
>, canceller
: Canceller
) -> Self {
433 h2
: H2Client
::new(h2
),
434 canceller
: Some(canceller
),
438 pub fn get(&self, path
: &str, param
: Option
<Value
>) -> impl Future
<Item
=Value
, Error
=Error
> {
439 self.h2
.get(path
, param
)
442 pub fn put(&self, path
: &str, param
: Option
<Value
>) -> impl Future
<Item
=Value
, Error
=Error
> {
443 self.h2
.put(path
, param
)
446 pub fn post(&self, path
: &str, param
: Option
<Value
>) -> impl Future
<Item
=Value
, Error
=Error
> {
447 self.h2
.post(path
, param
)
450 pub fn finish(mut self) -> impl Future
<Item
=(), Error
=Error
> {
451 let canceler
= self.canceller
.take().unwrap();
452 self.h2
.clone().post("finish", None
).map(move |_
| {
458 pub fn force_close(mut self) {
459 self.canceller
.take().unwrap().cancel();
462 pub fn upload_blob_from_data(
466 crypt_config
: Option
<Arc
<CryptConfig
>>,
468 ) -> impl Future
<Item
=(), Error
=Error
> {
470 let h2
= self.h2
.clone();
471 let file_name
= file_name
.to_owned();
473 futures
::future
::ok(())
475 let blob
= if let Some(ref crypt_config
) = crypt_config
{
476 DataBlob
::encode(&data
, Some(crypt_config
), compress
)?
478 DataBlob
::encode(&data
, None
, compress
)?
481 let raw_data
= blob
.into_inner();
484 .and_then(move |raw_data
| {
485 let param
= json
!({"encoded-size": raw_data.len(), "file-name": file_name }
);
486 h2
.upload("blob", Some(param
), raw_data
)
491 pub fn upload_blob_from_file
<P
: AsRef
<std
::path
::Path
>>(
495 crypt_config
: Option
<Arc
<CryptConfig
>>,
497 ) -> impl Future
<Item
=(), Error
=Error
> {
499 let h2
= self.h2
.clone();
500 let file_name
= file_name
.to_owned();
501 let src_path
= src_path
.as_ref().to_owned();
503 let task
= tokio
::fs
::File
::open(src_path
.clone())
504 .map_err(move |err
| format_err
!("unable to open file {:?} - {}", src_path
, err
))
505 .and_then(move |file
| {
506 let contents
= vec
![];
507 tokio
::io
::read_to_end(file
, contents
)
508 .map_err(Error
::from
)
509 .and_then(move |(_
, contents
)| {
510 let blob
= if let Some(ref crypt_config
) = crypt_config
{
511 DataBlob
::encode(&contents
, Some(crypt_config
), compress
)?
513 DataBlob
::encode(&contents
, None
, compress
)?
515 let raw_data
= blob
.into_inner();
518 .and_then(move |raw_data
| {
519 let param
= json
!({"encoded-size": raw_data.len(), "file-name": file_name }
);
520 h2
.upload("blob", Some(param
), raw_data
)
528 pub fn upload_stream(
531 stream
: impl Stream
<Item
=bytes
::BytesMut
, Error
=Error
>,
533 fixed_size
: Option
<u64>,
534 crypt_config
: Option
<Arc
<CryptConfig
>>,
535 ) -> impl Future
<Item
=(), Error
=Error
> {
537 let known_chunks
= Arc
::new(Mutex
::new(HashSet
::new()));
539 let h2
= self.h2
.clone();
540 let h2_2
= self.h2
.clone();
541 let h2_3
= self.h2
.clone();
542 let h2_4
= self.h2
.clone();
544 let mut param
= json
!({ "archive-name": archive_name }
);
545 if let Some(size
) = fixed_size
{
546 param
["size"] = size
.into();
549 let index_path
= format
!("{}_index", prefix
);
550 let close_path
= format
!("{}_close", prefix
);
552 let prefix
= prefix
.to_owned();
554 Self::download_chunk_list(h2
, &index_path
, archive_name
, known_chunks
.clone())
556 h2_2
.post(&index_path
, Some(param
))
558 .and_then(move |res
| {
559 let wid
= res
.as_u64().unwrap();
560 Self::upload_chunk_info_stream(h2_3
, wid
, stream
, &prefix
, known_chunks
.clone(), crypt_config
)
561 .and_then(move |(chunk_count
, size
, _speed
)| {
564 "chunk-count": chunk_count
,
567 h2_4
.post(&close_path
, Some(param
))
573 fn response_queue() -> (
574 mpsc
::Sender
<h2
::client
::ResponseFuture
>,
575 sync
::oneshot
::Receiver
<Result
<(), Error
>>
577 let (verify_queue_tx
, verify_queue_rx
) = mpsc
::channel(100);
578 let (verify_result_tx
, verify_result_rx
) = sync
::oneshot
::channel();
582 .map_err(Error
::from
)
583 .for_each(|response
: h2
::client
::ResponseFuture
| {
585 .map_err(Error
::from
)
586 .and_then(H2Client
::h2api_response
)
588 println
!("RESPONSE: {:?}", result
);
591 .map_err(|err
| format_err
!("pipelined request failed: {}", err
))
594 verify_result_tx
.send(result
)
596 .map_err(|_
| { /* ignore closed channel */ }
)
599 (verify_queue_tx
, verify_result_rx
)
602 fn append_chunk_queue(h2
: H2Client
, wid
: u64, path
: String
) -> (
603 mpsc
::Sender
<(MergedChunkInfo
, Option
<h2
::client
::ResponseFuture
>)>,
604 sync
::oneshot
::Receiver
<Result
<(), Error
>>
606 let (verify_queue_tx
, verify_queue_rx
) = mpsc
::channel(64);
607 let (verify_result_tx
, verify_result_rx
) = sync
::oneshot
::channel();
609 let h2_2
= h2
.clone();
613 .map_err(Error
::from
)
614 .and_then(move |(merged_chunk_info
, response
): (MergedChunkInfo
, Option
<h2
::client
::ResponseFuture
>)| {
615 match (response
, merged_chunk_info
) {
616 (Some(response
), MergedChunkInfo
::Known(list
)) => {
619 .map_err(Error
::from
)
620 .and_then(H2Client
::h2api_response
)
621 .and_then(move |_result
| {
622 Ok(MergedChunkInfo
::Known(list
))
626 (None
, MergedChunkInfo
::Known(list
)) => {
627 future
::Either
::B(future
::ok(MergedChunkInfo
::Known(list
)))
632 .merge_known_chunks()
633 .and_then(move |merged_chunk_info
| {
634 match merged_chunk_info
{
635 MergedChunkInfo
::Known(chunk_list
) => {
636 let mut digest_list
= vec
![];
637 let mut offset_list
= vec
![];
638 for (offset
, digest
) in chunk_list
{
639 //println!("append chunk {} (offset {})", proxmox::tools::digest_to_hex(&digest), offset);
640 digest_list
.push(proxmox
::tools
::digest_to_hex(&digest
));
641 offset_list
.push(offset
);
643 println
!("append chunks list len ({})", digest_list
.len());
644 let param
= json
!({ "wid": wid, "digest-list": digest_list, "offset-list": offset_list }
);
645 let mut request
= H2Client
::request_builder("localhost", "PUT", &path
, None
).unwrap();
646 request
.headers_mut().insert(hyper
::header
::CONTENT_TYPE
, HeaderValue
::from_static("application/json"));
647 let param_data
= bytes
::Bytes
::from(param
.to_string().as_bytes());
648 let upload_data
= Some(param_data
);
649 h2_2
.send_request(request
, upload_data
)
650 .and_then(move |response
| {
652 .map_err(Error
::from
)
653 .and_then(H2Client
::h2api_response
)
654 .and_then(|_
| Ok(()))
656 .map_err(|err
| format_err
!("pipelined request failed: {}", err
))
661 .for_each(|_
| Ok(()))
663 verify_result_tx
.send(result
)
665 .map_err(|_
| { /* ignore closed channel */ }
)
668 (verify_queue_tx
, verify_result_rx
)
671 fn download_chunk_list(
675 known_chunks
: Arc
<Mutex
<HashSet
<[u8;32]>>>,
676 ) -> impl Future
<Item
=(), Error
=Error
> {
678 let param
= json
!({ "archive-name": archive_name }
);
679 let request
= H2Client
::request_builder("localhost", "GET", path
, Some(param
)).unwrap();
681 h2
.send_request(request
, None
)
682 .and_then(move |response
| {
684 .map_err(Error
::from
)
685 .and_then(move |resp
| {
686 let status
= resp
.status();
688 if !status
.is_success() {
689 future
::Either
::A(H2Client
::h2api_response(resp
).and_then(|_
| { bail!("unknown error"); }
))
691 future
::Either
::B(future
::ok(resp
.into_body()))
694 .and_then(move |mut body
| {
696 let mut release_capacity
= body
.release_capacity().clone();
698 DigestListDecoder
::new(body
.map_err(Error
::from
))
699 .for_each(move |chunk
| {
700 let _
= release_capacity
.release_capacity(chunk
.len());
701 println
!("GOT DOWNLOAD {}", proxmox
::tools
::digest_to_hex(&chunk
));
702 known_chunks
.lock().unwrap().insert(chunk
);
709 fn upload_chunk_info_stream(
712 stream
: impl Stream
<Item
=bytes
::BytesMut
, Error
=Error
>,
714 known_chunks
: Arc
<Mutex
<HashSet
<[u8;32]>>>,
715 crypt_config
: Option
<Arc
<CryptConfig
>>,
716 ) -> impl Future
<Item
=(usize, usize, usize), Error
=Error
> {
718 let repeat
= std
::sync
::Arc
::new(AtomicUsize
::new(0));
719 let repeat2
= repeat
.clone();
721 let stream_len
= std
::sync
::Arc
::new(AtomicUsize
::new(0));
722 let stream_len2
= stream_len
.clone();
724 let append_chunk_path
= format
!("{}_index", prefix
);
725 let upload_chunk_path
= format
!("{}_chunk", prefix
);
727 let (upload_queue
, upload_result
) = Self::append_chunk_queue(h2
.clone(), wid
, append_chunk_path
.to_owned());
729 let start_time
= std
::time
::Instant
::now();
732 .and_then(move |data
| {
734 let chunk_len
= data
.len();
736 repeat
.fetch_add(1, Ordering
::SeqCst
);
737 let offset
= stream_len
.fetch_add(chunk_len
, Ordering
::SeqCst
) as u64;
739 let mut chunk_builder
= DataChunkBuilder
::new(data
.as_ref())
742 if let Some(ref crypt_config
) = crypt_config
{
743 chunk_builder
= chunk_builder
.crypt_config(crypt_config
);
746 let mut known_chunks
= known_chunks
.lock().unwrap();
747 let digest
= chunk_builder
.digest();
748 let chunk_is_known
= known_chunks
.contains(digest
);
750 Ok(MergedChunkInfo
::Known(vec
![(offset
, *digest
)]))
752 known_chunks
.insert(*digest
);
753 let chunk
= chunk_builder
.build()?
;
754 Ok(MergedChunkInfo
::New(ChunkInfo { chunk, chunk_len: chunk_len as u64, offset }
))
757 .merge_known_chunks()
758 .for_each(move |merged_chunk_info
| {
760 if let MergedChunkInfo
::New(chunk_info
) = merged_chunk_info
{
761 let offset
= chunk_info
.offset
;
762 let digest
= *chunk_info
.chunk
.digest();
763 let digest_str
= proxmox
::tools
::digest_to_hex(&digest
);
764 let upload_queue
= upload_queue
.clone();
766 println
!("upload new chunk {} ({} bytes, offset {})", digest_str
,
767 chunk_info
.chunk_len
, offset
);
769 let chunk_data
= chunk_info
.chunk
.raw_data();
772 "digest": digest_str
,
773 "size": chunk_info
.chunk_len
,
774 "encoded-size": chunk_data
.len(),
777 let request
= H2Client
::request_builder("localhost", "POST", &upload_chunk_path
, Some(param
)).unwrap();
778 let upload_data
= Some(bytes
::Bytes
::from(chunk_data
));
780 let new_info
= MergedChunkInfo
::Known(vec
![(offset
, digest
)]);
783 h2
.send_request(request
, upload_data
)
784 .and_then(move |response
| {
785 upload_queue
.clone().send((new_info
, Some(response
)))
786 .map(|_
| ()).map_err(Error
::from
)
792 upload_queue
.clone().send((merged_chunk_info
, None
))
793 .map(|_
| ()).map_err(Error
::from
)
797 .then(move |result
| {
798 println
!("RESULT {:?}", result
);
799 upload_result
.map_err(Error
::from
).and_then(|upload1_result
| {
800 Ok(upload1_result
.and(result
))
805 let repeat
= repeat2
.load(Ordering
::SeqCst
);
806 let stream_len
= stream_len2
.load(Ordering
::SeqCst
);
807 let speed
= ((stream_len
*1000000)/(1024*1024))/(start_time
.elapsed().as_micros() as usize);
808 println
!("Uploaded {} chunks in {} seconds ({} MB/s).", repeat
, start_time
.elapsed().as_secs(), speed
);
810 println
!("Average chunk size was {} bytes.", stream_len
/repeat
);
811 println
!("Time per request: {} microseconds.", (start_time
.elapsed().as_micros())/(repeat
as u128
));
813 Ok((repeat
, stream_len
, speed
))
817 pub fn upload_speedtest(&self) -> impl Future
<Item
=usize, Error
=Error
> {
819 let mut data
= vec
![];
820 // generate pseudo random byte sequence
821 for i
in 0..1024*1024 {
823 let byte
= ((i
>> (j
<<3))&0xff) as u8;
828 let item_len
= data
.len();
830 let repeat
= std
::sync
::Arc
::new(AtomicUsize
::new(0));
831 let repeat2
= repeat
.clone();
833 let (upload_queue
, upload_result
) = Self::response_queue();
835 let start_time
= std
::time
::Instant
::now();
837 let h2
= self.h2
.clone();
839 futures
::stream
::repeat(data
)
840 .take_while(move |_
| {
841 repeat
.fetch_add(1, Ordering
::SeqCst
);
842 Ok(start_time
.elapsed().as_secs() < 5)
844 .for_each(move |data
| {
847 let upload_queue
= upload_queue
.clone();
849 println
!("send test data ({} bytes)", data
.len());
850 let request
= H2Client
::request_builder("localhost", "POST", "speedtest", None
).unwrap();
851 h2
.send_request(request
, Some(bytes
::Bytes
::from(data
)))
852 .and_then(move |response
| {
853 upload_queue
.send(response
)
854 .map(|_
| ()).map_err(Error
::from
)
857 .then(move |result
| {
858 println
!("RESULT {:?}", result
);
859 upload_result
.map_err(Error
::from
).and_then(|upload1_result
| {
860 Ok(upload1_result
.and(result
))
865 let repeat
= repeat2
.load(Ordering
::SeqCst
);
866 println
!("Uploaded {} chunks in {} seconds.", repeat
, start_time
.elapsed().as_secs());
867 let speed
= ((item_len
*1000000*(repeat
as usize))/(1024*1024))/(start_time
.elapsed().as_micros() as usize);
869 println
!("Time per request: {} microseconds.", (start_time
.elapsed().as_micros())/(repeat
as u128
));
877 pub struct H2Client
{
878 h2
: h2
::client
::SendRequest
<bytes
::Bytes
>,
883 pub fn new(h2
: h2
::client
::SendRequest
<bytes
::Bytes
>) -> Self {
887 pub fn get(&self, path
: &str, param
: Option
<Value
>) -> impl Future
<Item
=Value
, Error
=Error
> {
888 let req
= Self::request_builder("localhost", "GET", path
, param
).unwrap();
892 pub fn put(&self, path
: &str, param
: Option
<Value
>) -> impl Future
<Item
=Value
, Error
=Error
> {
893 let req
= Self::request_builder("localhost", "PUT", path
, param
).unwrap();
897 pub fn post(&self, path
: &str, param
: Option
<Value
>) -> impl Future
<Item
=Value
, Error
=Error
> {
898 let req
= Self::request_builder("localhost", "POST", path
, param
).unwrap();
902 pub fn upload(&self, path
: &str, param
: Option
<Value
>, data
: Vec
<u8>) -> impl Future
<Item
=Value
, Error
=Error
> {
903 let request
= Self::request_builder("localhost", "POST", path
, param
).unwrap();
908 .map_err(Error
::from
)
909 .and_then(move |mut send_request
| {
910 let (response
, stream
) = send_request
.send_request(request
, false).unwrap();
911 PipeToSendStream
::new(bytes
::Bytes
::from(data
), stream
)
914 .map_err(Error
::from
)
915 .and_then(Self::h2api_response
)
922 request
: Request
<()>,
923 ) -> impl Future
<Item
=Value
, Error
=Error
> {
925 self.send_request(request
, None
)
926 .and_then(move |response
| {
928 .map_err(Error
::from
)
929 .and_then(Self::h2api_response
)
935 request
: Request
<()>,
936 data
: Option
<bytes
::Bytes
>,
937 ) -> impl Future
<Item
=h2
::client
::ResponseFuture
, Error
=Error
> {
941 .map_err(Error
::from
)
942 .and_then(move |mut send_request
| {
943 if let Some(data
) = data
{
944 let (response
, stream
) = send_request
.send_request(request
, false).unwrap();
945 future
::Either
::A(PipeToSendStream
::new(data
, stream
)
950 let (response
, _stream
) = send_request
.send_request(request
, true).unwrap();
951 future
::Either
::B(future
::ok(response
))
956 fn h2api_response(response
: Response
<h2
::RecvStream
>) -> impl Future
<Item
=Value
, Error
=Error
> {
958 let status
= response
.status();
960 let (_head
, mut body
) = response
.into_parts();
962 // The `release_capacity` handle allows the caller to manage
965 // Whenever data is received, the caller is responsible for
966 // releasing capacity back to the server once it has freed
967 // the data from memory.
968 let mut release_capacity
= body
.release_capacity().clone();
972 // Let the server send more data.
973 let _
= release_capacity
.release_capacity(chunk
.len());
977 .map_err(Error
::from
)
978 .and_then(move |data
| {
979 let text
= String
::from_utf8(data
.to_vec()).unwrap();
980 if status
.is_success() {
982 let mut value
: Value
= serde_json
::from_str(&text
)?
;
983 if let Some(map
) = value
.as_object_mut() {
984 if let Some(data
) = map
.remove("data") {
988 bail
!("got result without data property");
993 bail
!("HTTP Error {}: {}", status
, text
);
998 // Note: We always encode parameters with the url
999 pub fn request_builder(server
: &str, method
: &str, path
: &str, data
: Option
<Value
>) -> Result
<Request
<()>, Error
> {
1000 let path
= path
.trim_matches('
/'
);
1002 if let Some(data
) = data
{
1003 let query
= tools
::json_object_to_query(data
)?
;
1004 // We detected problem with hyper around 6000 characters - seo we try to keep on the safe side
1005 if query
.len() > 4096 { bail!("h2 query data too large ({} bytes
) - please encode data inside body
", query.len()); }
1006 let url: Uri = format!("https
://{}:8007/{}?{}", server, path, query).parse()?;
1007 let request
= Request
::builder()
1010 .header("User-Agent", "proxmox-backup-client/1.0")
1011 .header(hyper
::header
::CONTENT_TYPE
, "application/x-www-form-urlencoded")
1015 let url
: Uri
= format
!("https://{}:8007/{}", server
, path
).parse()?
;
1016 let request
= Request
::builder()
1019 .header("User-Agent", "proxmox-backup-client/1.0")
1020 .header(hyper
::header
::CONTENT_TYPE
, "application/x-www-form-urlencoded")