longshoreman 0.1.0

asynchronous Docker client in pure Rust
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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252

use super::{BodyType, HttpClient};
use crate::{Error, Result};
use futures_util::{
    future::TryFutureExt,
    stream::{Stream, StreamExt, TryStreamExt},
};
use hyper::{
    body::{to_bytes, Bytes},
    Body, Method, Request, StatusCode,
};
use serde::{Deserialize, Serialize};
use std::convert::TryFrom;

/// A builder API for constructing and sending an HTTP request to the Docker
/// host.
///
/// ## Builder Methods
/// This object has a number of 'builder' methods for building the request.
/// These are modelled after `Reqwest`'s API.
///
/// ## Finalisers
/// A number of 'finaliser' methods are provided for turning the received
/// responses into different response types, depending on needs of the endpoint.
/// These include unary responses, streams, and decoding into JSON objects.
///
/// Implementations of specific endpoints may make use of these finalisers, and
/// need not reimplement this logic.
pub(crate) struct RequestBuilder<'a> {
    http_client: &'a HttpClient,
    uri_base: String,
    query: Option<String>,
    body: Option<BodyType>,
    builder: http::request::Builder,
}

impl<'a> RequestBuilder<'a> {
    pub fn new(http_client: &'a HttpClient, endpoint: impl AsRef<str>) -> Self {
        let uri_base = http_client.uri(endpoint.as_ref());
        let query = None;
        let body = None;
        let mut builder = hyper::Request::builder();
        builder = builder.method(Method::GET);

        debug_assert!(builder.method_ref().is_some());

        Self {
            http_client,
            uri_base,
            query,
            body,
            builder,
        }
    }

    // Builder methods

    /// Set the HTTP method
    pub fn method(mut self, method: Method) -> Self {
        self.builder = self.builder.method(method);
        self
    }

    /// Add a custom header to the request
    /*     pub fn header(mut self, key: impl IntoHeaderName, value: &str) -> Self {
        let value = http::header::HeaderValue::from_str(value).unwrap();
        self.builder.headers_mut().unwrap().append(key, value);
        self
    } */

    /// Add a form-encoded query to the request.
    ///
    /// Any type that implements [`serde::Deserialize`] can be used
    pub fn query<T: Serialize>(mut self, query: T) -> Self {
        let query_string = serde_urlencoded::ser::to_string(query).unwrap();

        self.query = Some(query_string);
        self
    }

    /// Add a JSON body to the request.
    ///
    /// Any type that implements [`serde::Deserialize`] can be used. This method
    /// will automatically set the content type header
    pub fn json_body<T: Serialize>(mut self, body: T) -> Self {
        let data = serde_json::to_vec(&body).unwrap();
        self.body = Some(BodyType::json(data));
        self
    }

    /// Add a tar archive as the body of the request
    ///
    /// This method will automatically set the content type header
    pub fn tar_body(mut self, data: Vec<u8>) -> Self {
        self.body = Some(BodyType::tar(data));
        self
    }

    // Finalisers

    /// Build the request
    fn into_request(self) -> Result<(Request<Body>, &'a HttpClient)> {
        let uri = if let Some(query_string) = &self.query {
            format!("{}?{}", self.uri_base, query_string)
        } else {
            self.uri_base
        };

        let mut builder = self.builder;
        builder = builder.uri(uri);

        let request = match self.body {
            Some(body_type) => {
                let mime = http::HeaderValue::try_from(body_type.mime())?;
                let data = body_type.into_data();
                let body = Body::from(data);

                builder.headers_mut().unwrap().append("content-type", mime);
                builder.body(body)?
            }
            None => builder.body(Body::empty())?,
        };

        Ok((request, self.http_client))
    }

    /// Send the request and return a [`hyper::Result`]
    pub async fn into_response(self) -> Result<hyper::Response<Body>> {
        let (request, client) = self.into_request()?;
        Ok(client.send_request(request).await?)
    }

    /// Send the request and return the body of the response.
    ///
    /// This method will check the status code of the response convert it into
    /// an error, as required.
    async fn into_body(self) -> Result<Body> {
        let response = self.into_response().await?;
        let status = response.status();

        match status {
            // Success case: pass on the response
            StatusCode::OK
            | StatusCode::CREATED
            | StatusCode::SWITCHING_PROTOCOLS
            | StatusCode::NO_CONTENT => Ok(response.into_body()),
            // Error case: parse the text
            _ => {
                let bytes = to_bytes(response.into_body()).await?;

                Err(Error::Fault {
                    code: status,
                    message: get_error_message(&bytes).unwrap_or_else(|_| {
                        status
                            .canonical_reason()
                            .unwrap_or_else(|| "unknown error code")
                            .to_owned()
                    }),
                })
            }
        }
    }

    /// Upgrade the HTTP connection into a duplex stream
    /*     pub async fn upgrade(mut self) -> Result<impl AsyncRead + AsyncWrite> {
        self = self.header(hyper::header::CONNECTION, "Upgrade");
        self = self.header(hyper::header::UPGRADE, "tcp");

        let hyper_response = self.into_response().await?;

        match hyper_response.status() {
            StatusCode::SWITCHING_PROTOCOLS => Ok(hyper_response.into_body().on_upgrade().await?),
            _ => Err(Error::ConnectionNotUpgraded),
        }
    } */

    /// Send the request and concatenate the response into a Bytes object
    async fn into_bytes(self) -> Result<Bytes> {
        let body = self.into_body().await?;
        Ok(to_bytes(body).await?)
    }

    /*     /// Send the request, and concatenate the response into a string
    pub async fn into_string(self) -> Result<String> {
        let bytes = self.into_bytes().await?;
        Ok(String::from_utf8(bytes.to_vec())?)
    } */

    /// Send the request and deserialize the JSON body response into an object
    pub async fn into_json<T>(self) -> Result<T>
    where
        for<'de> T: Deserialize<'de>,
    {
        let bytes = self.into_bytes().await?;
        Ok(serde_json::from_slice(&bytes)?)
    }

    /// Send the request, and return the response body as a stream of bytes
    pub fn into_stream(self) -> impl Stream<Item = Result<Bytes>> + 'a {
        async fn unfold(mut body: Body) -> Option<(Result<Bytes>, Body)> {
            let bytes_result = body.next().await?.map_err(Error::from);

            Some((bytes_result, body))
        }

        async move {
            let body = self.into_body().await?;
            Ok(futures_util::stream::unfold(body, unfold))
        }
        .try_flatten_stream()
    }

    /// Send the request, and deserialize the returned stream of JSON into a
    /// stream of objects.
    pub fn into_stream_json<T>(self) -> impl Stream<Item = Result<T>> + 'a
    where
        for<'de> T: Deserialize<'de>,
    {
        let byte_stream = self.into_stream();

        byte_stream.and_then(|bytes| async move { Ok(serde_json::from_slice(&bytes)?) })
    }

    /*     /// Send the request, and deserialize the returned stream using the given
    /// codec
    pub fn decode<T, C, I, E>(self, codec: C) -> impl Stream<Item = Result<T>> + 'a
    where
        for<'de> T: Deserialize<'de>,
        C: Decoder<Item = I, Error = E> + 'a,
        I: AsRef<[u8]> + 'a,
        Error: From<E>,
        E: From<std::io::Error> + 'a,
    {
        let stream = self
            .into_stream()
            .map_err(|e| io::Error::new(io::ErrorKind::Other, e));

        let reader = stream_reader(stream);
        FramedRead::new(reader, codec)
            .map_err(Error::from)
            .and_then(|bytes| async move { Ok(serde_json::from_slice(bytes.as_ref())?) })
    } */
}

fn get_error_message(bytes: impl AsRef<[u8]>) -> Result<String> {
    #[derive(Serialize, Deserialize)]
    struct ErrorResponse {
        message: String,
    }

    let error_response: ErrorResponse = serde_json::from_slice(bytes.as_ref())?;
    Ok(error_response.message)
}