rusoto_credential 0.12.0

AWS credential tooling
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

use std::io::{Error as IoError};
use std::io::ErrorKind::InvalidData;
use std::mem;
use std::time::{Duration, Instant};

use futures::{Async, Future, Poll, Stream};
use futures::stream::Concat2;
use hyper::{Client as HyperClient, Body, Request, Uri};
use hyper::client::{ResponseFuture as HyperResponseFuture, HttpConnector};
use tokio_timer::Deadline;

use super::CredentialsError;

/// A future that will resolve to an `HttpResponse`.
pub struct HttpClientFuture(ClientFutureInner);

enum RequestFuture {
    Waiting(HyperResponseFuture),
    Buffering(Concat2<Body>),
    Swapping
}

impl Future for RequestFuture {
    type Item = String;
    type Error = CredentialsError;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        match mem::replace(self, RequestFuture::Swapping) {
            RequestFuture::Waiting(mut hyper_future) => {
                match hyper_future.poll()? {
                    Async::NotReady => {
                        *self = RequestFuture::Waiting(hyper_future);
                        Ok(Async::NotReady)
                    },
                    Async::Ready(res) => {
                        if !res.status().is_success() {
                            Err(CredentialsError {
                                message: format!("Invalid Response Code: {}", res.status())
                            })
                        } else {
                            *self = RequestFuture::Buffering(res.into_body().concat2());
                            self.poll()
                        }
                    }
                }
            },
            RequestFuture::Buffering(mut concat_future) => {
                match concat_future.poll()? {
                    Async::NotReady => {
                        *self = RequestFuture::Buffering(concat_future);
                        Ok(Async::NotReady)
                    },
                    Async::Ready(body) => {
                        let string_body = String::from_utf8(body.to_vec())
                            .map_err(|_| IoError::new(InvalidData, "Non UTF-8 Data returned"))?;
                        Ok(Async::Ready(string_body))
                    }
                }
            },
            RequestFuture::Swapping => unreachable!()
        }
    }
}

enum ClientFutureInner {
    Request(Deadline<RequestFuture>),
    Error(String)
}

impl Future for HttpClientFuture {
    type Item = String;
    type Error = CredentialsError;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        match self.0 {
            ClientFutureInner::Error(ref message) =>
                Err(CredentialsError { message: message.clone() }),
            ClientFutureInner::Request(ref mut deadline_future) => {
                match deadline_future.poll() {
                    Err(deadline_error) => {
                        Err(CredentialsError {
                            message: deadline_error.to_string()
                        })
                    }
                    Ok(Async::NotReady) =>
                        Ok(Async::NotReady),
                    Ok(Async::Ready(body)) =>
                        Ok(Async::Ready(body)),
                }
            }
        }
    }
}

/// Http client for use in a credentials provider.
#[derive(Debug, Clone)]
pub struct HttpClient {
    inner: HyperClient<HttpConnector>
}

impl HttpClient {
    /// Create an http client.
    pub fn new() -> HttpClient {
        HttpClient {
            inner: HyperClient::new(),
        }
    }

    pub fn get(&self, uri: Uri, timeout: Duration) -> HttpClientFuture {
        match Request::get(uri).body(Body::empty()) {
            Ok(request) => self.request(request, timeout),
            Err(err) => HttpClientFuture(ClientFutureInner::Error(err.to_string())),
        }
    }

    pub fn request(&self, req: Request<Body>, timeout: Duration) -> HttpClientFuture {
        let future = RequestFuture::Waiting(self.inner.request(req));
        let deadline = Instant::now() + timeout;
        let inner = ClientFutureInner::Request(Deadline::new(future, deadline));
        HttpClientFuture(inner)
    }
}