docker-client-async 0.1.0

/*
 * Copyright 2020 Damian Peckett <damian@pecke.tt>.
 * Copyright 2013-2018 Docker, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

//! Docker container api implementation.

use crate::error::*;
use crate::types::client::*;
use crate::types::container::{Container, ContainerConfig, UpdateConfig, WaitCondition};
use crate::types::filters;
use crate::types::stats::Stats;
use crate::{
    get_docker_os, get_filters_query, read_response_body, read_response_body_raw, version,
    AsyncHttpBodyReader, DockerEngineClient,
};
use bytes::Bytes;
use futures::ready;
use futures::task::{Context, Poll};
use futures::Stream;
use hyper::client::connect::Connect;
use hyper::{Body, Method, Request, Response};
use percent_encoding::{utf8_percent_encode, NON_ALPHANUMERIC};
use snafu::ensure;
use snafu::OptionExt;
use snafu::ResultExt;
use std::collections::HashMap;
use std::time::{Duration, SystemTime};
use tokio::io::AsyncBufReadExt;
use tokio::io::BufReader;
use tokio::io::Lines;
use tokio::macros::support::Pin;
use tokio::time::timeout;

impl<C: Connect + Clone + Send + Sync + 'static> DockerEngineClient<C> {
    /// container_list returns the list of containers in the docker host.
    pub async fn container_list(
        &self,
        options: Option<ContainerListOptions>,
    ) -> Result<Vec<Container>, Error> {
        let mut query_params: HashMap<String, String> = HashMap::new();
        if let Some(options) = options {
            if options.size {
                query_params.insert("size".into(), "1".into());
            }
            if options.all {
                query_params.insert("all".into(), "1".into());
            }
            if let Some(limit) = options.limit {
                query_params.insert("limit".into(), limit.to_string());
            }
            if let Some(filters) = options.filters.as_ref() {
                query_params.insert(
                    "filters".into(),
                    serde_json::to_string(&filters.fields).context(JsonSerializationError {})?,
                );
            }
        }
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::GET)
            .uri(self.request_uri("/containers/json", query_params)?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );
        let response_body = read_response_body(response, self.timeout).await?;
        Ok(serde_json::from_str(&response_body).context(JsonDeserializationError {})?)
    }

    /// container_create creates a new container based in the given configuration.
    /// It can be associated with a name, but it's not mandatory.
    pub async fn container_create(
        &self,
        mut config: ContainerConfig,
        container_name: Option<String>,
    ) -> Result<ContainerCreateCreatedBody, Error> {
        // When using API 1.24 and under, the client is responsible for removing the container
        if version::less_than(&self.version, "1.25") {
            if let Some(mut host_config) = config.host_config.take() {
                host_config.auto_remove = Some(false);
                config.host_config = Some(host_config);
            }
        }

        let mut query_params: HashMap<String, String> = HashMap::new();
        if let Some(container_name) = &container_name {
            query_params.insert("name".into(), container_name.clone());
        }
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::POST)
            .uri(self.request_uri("/containers/create", query_params)?)
            .header("Content-Type", "application/json")
            .header("Accept", "application/json")
            .body(Body::from(
                serde_json::to_string(&config).context(JsonSerializationError {})?,
            ))
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );
        let response_body = read_response_body(response, self.timeout).await?;
        Ok(serde_json::from_str(&response_body).context(JsonDeserializationError {})?)
    }

    /// container_inspect returns the container information.
    pub async fn container_inspect(&self, container_id: &str) -> Result<Container, Error> {
        let request = Request::builder()
            .method(Method::GET)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/json",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                None,
            )?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );
        let response_body = read_response_body(response, self.timeout).await?;
        Ok(serde_json::from_str(&response_body).context(JsonDeserializationError {})?)
    }

    /// container_top shows process information from within a container.
    pub async fn container_top(
        &self,
        container_id: &str,
        arguments: Option<Vec<String>>,
    ) -> Result<ContainerTopOKBody, Error> {
        let mut query_params: HashMap<String, String> = HashMap::new();
        if let Some(arguments) = arguments {
            query_params.insert("ps_args".into(), arguments.join(" "));
        }
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::GET)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/top",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                query_params,
            )?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );
        let response_body = read_response_body(response, self.timeout).await?;
        Ok(serde_json::from_str(&response_body).context(JsonDeserializationError {})?)
    }

    /// container_logs returns the logs generated by a container.
    /// Currently only supports TTY containers.
    pub async fn container_logs(
        &self,
        container_id: &str,
        options: Option<ContainerLogsOptions>,
    ) -> Result<LogStream, Error> {
        let mut query_params: HashMap<String, String> = HashMap::new();
        if let Some(options) = options {
            if options.show_stdout {
                query_params.insert("stdout".into(), "1".into());
            }
            if options.show_stderr {
                query_params.insert("stderr".into(), "1".into());
            }
            if let Some(since) = options.since {
                let since_epoch = since.duration_since(SystemTime::UNIX_EPOCH).unwrap();
                query_params.insert("since".into(), format!("{}", since_epoch.as_secs_f64()));
            }
            if let Some(until) = options.until {
                let until_epoch = until.duration_since(SystemTime::UNIX_EPOCH).unwrap();
                query_params.insert("until".into(), format!("{}", until_epoch.as_secs_f64()));
            }
            if options.timestamps {
                query_params.insert("timestamps".into(), "1".into());
            }
            if options.details {
                query_params.insert("details".into(), "1".into());
            }
            if options.follow {
                query_params.insert("follow".into(), "1".into());
            }
            if let Some(tail) = options.tail {
                query_params.insert("tail".into(), tail);
            }
        }
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::GET)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/logs",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                query_params,
            )?)
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;

        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        Ok(LogStream::new(response))
    }

    /// container_diff shows differences in a container filesystem since it was started.
    pub async fn container_diff(
        &self,
        container_id: &str,
    ) -> Result<Option<Vec<ContainerChangeResponseItem>>, Error> {
        let request = Request::builder()
            .method(Method::GET)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/changes",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                None,
            )?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );
        let response_body = read_response_body(response, self.timeout).await?;
        Ok(serde_json::from_str(&response_body).context(JsonDeserializationError {})?)
    }

    /// container_export retrieves the raw contents of a container.
    pub async fn container_export(&self, container_id: &str) -> Result<Vec<u8>, Error> {
        let request = Request::builder()
            .method(Method::GET)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/export",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                None,
            )?)
            .header("Accept", "application/x-tar")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;
        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        let response_body_raw = read_response_body_raw(response, Duration::from_secs(300)).await?;
        Ok(response_body_raw)
    }

    /// container_stats returns near realtime stats for a given container.
    pub async fn container_stats(
        &self,
        container_id: &str,
        stream: bool,
    ) -> Result<ContainerStats, Error> {
        let mut query_params: HashMap<String, String> = HashMap::new();
        query_params.insert("stream".into(), "0".into());
        if stream {
            query_params.insert("stream".into(), "1".into());
        }
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::GET)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/stats",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                query_params,
            )?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;
        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        let os_type = response
            .headers()
            .get("Server")
            .map(|v| get_docker_os(v.to_str().unwrap()))
            .unwrap_or_else(String::new);

        Ok(ContainerStats {
            os_type,
            body: StatsStream::new(response),
        })
    }

    /// container_resize changes the size of the tty for a container.
    pub async fn container_resize(
        &self,
        container_id: &str,
        options: ResizeOptions,
    ) -> Result<(), Error> {
        let mut query_params: HashMap<String, String> = HashMap::new();
        if let Some(height) = options.height {
            query_params.insert("h".into(), height.to_string());
        }
        if let Some(width) = options.width {
            query_params.insert("w".into(), width.to_string());
        }
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::POST)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/resize",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                query_params,
            )?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        Ok(())
    }

    /// container_start sends a request to the docker daemon to start a container.
    pub async fn container_start(
        &self,
        container_id: &str,
        options: Option<ContainerStartOptions>,
    ) -> Result<(), Error> {
        let mut query_params: HashMap<String, String> = HashMap::new();
        if let Some(options) = options {
            if let Some(checkpoint_id) = options.checkpoint_id {
                query_params.insert("checkpoint".into(), checkpoint_id);
            }
            if let Some(checkpoint_dir) = options.checkpoint_dir {
                query_params.insert("checkpoint-dir".into(), checkpoint_dir);
            }
        }
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::POST)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/start",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                query_params,
            )?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        Ok(())
    }

    /// container_stop stops a container. In case the container fails to stop gracefully within a
    /// time frame specified by the timeout argument, it is forcefully terminated (killed).
    pub async fn container_stop(&self, container_id: &str, timeout: Duration) -> Result<(), Error> {
        let mut query_params: HashMap<String, String> = HashMap::new();
        query_params.insert("t".into(), timeout.as_secs().to_string());
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::POST)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/stop",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                query_params,
            )?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = tokio::time::timeout(timeout + self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        Ok(())
    }

    /// container_restart stops and starts a container again. It makes the daemon to wait for the
    /// container to be up again for a specific amount of time, given the timeout.
    pub async fn container_restart(
        &self,
        container_id: &str,
        timeout: Duration,
    ) -> Result<(), Error> {
        let mut query_params: HashMap<String, String> = HashMap::new();
        query_params.insert("t".into(), timeout.as_secs().to_string());
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::POST)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/restart",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                query_params,
            )?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = tokio::time::timeout(timeout + self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        Ok(())
    }

    /// container_kill terminates the container process but does not remove the container from the
    /// docker host.
    pub async fn container_kill(
        &self,
        container_id: &str,
        signal: Option<String>,
    ) -> Result<(), Error> {
        let mut query_params: HashMap<String, String> = HashMap::new();
        if let Some(signal) = signal {
            query_params.insert("signal".into(), signal);
        }
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::POST)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/kill",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                query_params,
            )?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        Ok(())
    }

    /// container_update updates resources of a container.
    pub async fn container_update(
        &self,
        container_id: &str,
        update_config: UpdateConfig,
    ) -> Result<ContainerUpdateOKBody, Error> {
        let request = Request::builder()
            .method(Method::POST)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/update",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                None,
            )?)
            .header("Content-Type", "application/json")
            .header("Accept", "application/json")
            .body(Body::from(
                serde_json::to_string(&update_config).context(JsonSerializationError {})?,
            ))
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );
        let response_body = read_response_body(response, self.timeout).await?;
        Ok(serde_json::from_str(&response_body).context(JsonDeserializationError {})?)
    }

    /// container_rename changes the name of a given container.
    pub async fn container_rename(
        &self,
        container_id: &str,
        new_container_name: &str,
    ) -> Result<(), Error> {
        let mut query_params: HashMap<String, String> = HashMap::new();
        query_params.insert("name".into(), new_container_name.to_string());
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::POST)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/rename",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                query_params,
            )?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        Ok(())
    }

    /// container_pause pauses the main process of a given container without terminating it.
    pub async fn container_pause(&self, container_id: &str) -> Result<(), Error> {
        let request = Request::builder()
            .method(Method::POST)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/pause",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                None,
            )?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        Ok(())
    }

    /// container_unpause resumes the process execution within a container.
    pub async fn container_unpause(&self, container_id: &str) -> Result<(), Error> {
        let request = Request::builder()
            .method(Method::POST)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/unpause",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                None,
            )?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        Ok(())
    }

    /// container_wait waits until the specified container is in a certain state
    /// indicated by the given condition, either "not-running" (default),
    /// "next-exit", or "removed".
    pub async fn container_wait(
        &self,
        container_id: &str,
        condition: WaitCondition,
        timeout: Duration,
    ) -> Result<ContainerWaitOKBody, Error> {
        let mut query_params: HashMap<String, String> = HashMap::new();
        query_params.insert("condition".into(), condition.to_string());
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::POST)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/wait",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                query_params,
            )?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = tokio::time::timeout(timeout + self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        let response_body = read_response_body(response, timeout).await?;
        Ok(serde_json::from_str(&response_body).context(JsonDeserializationError {})?)
    }

    /// container_remove kills and removes a container from the docker host.
    pub async fn container_remove(
        &self,
        container_id: &str,
        options: Option<ContainerRemoveOptions>,
    ) -> Result<(), Error> {
        let mut query_params: HashMap<String, String> = HashMap::new();
        if let Some(options) = options {
            if options.remove_volumes {
                query_params.insert("v".into(), "1".into());
            }
            if options.remove_links {
                query_params.insert("link".into(), "1".into());
            }
            if options.force {
                query_params.insert("force".into(), "1".into());
            }
        }
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::DELETE)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                query_params,
            )?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        Ok(())
    }

    /// container_stat_path returns Stat information about a path inside the container filesystem.
    pub async fn container_stat_path(
        &self,
        container_id: &str,
        path: &str,
    ) -> Result<ContainerPathStat, Error> {
        let mut query_params: HashMap<String, String> = HashMap::new();
        query_params.insert("path".into(), path.to_string());
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::HEAD)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/archive",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                query_params,
            )?)
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        response
            .headers()
            .get("X-Docker-Container-Path-Stat")
            .map(|v| get_container_path_stat_from_header(v.to_str().unwrap()))
            .context(MalformedResponseError {})?
    }

    /// copy_from_container gets the content from the container and returns it.
    /// for a TAR archive to manipulate it in the host.
    pub async fn copy_from_container(
        &self,
        container_id: &str,
        src_path: &str,
    ) -> Result<(ContainerPathStat, Vec<u8>), Error> {
        let mut query_params: HashMap<String, String> = HashMap::new();
        query_params.insert("path".into(), src_path.to_string());
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::GET)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/archive",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                query_params,
            )?)
            .header("Accept", "application/x-tar")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        let container_path_stat = response
            .headers()
            .get("X-Docker-Container-Path-Stat")
            .map(|v| get_container_path_stat_from_header(v.to_str().unwrap()))
            .context(MalformedResponseError {})??;

        let response_body_raw = read_response_body_raw(response, Duration::from_secs(300)).await?;
        Ok((container_path_stat, response_body_raw))
    }

    /// copy_to_container copies content into the container filesystem.
    /// Note that `content` must be a stream for a TAR archive
    pub async fn copy_to_container<S, O, E>(
        &self,
        container_id: &str,
        dst_path: &str,
        content: S,
        options: Option<CopyToContainerOptions>,
    ) -> Result<(), Error>
    where
        S: Stream<Item = Result<O, E>> + Send + Sync + 'static,
        O: Into<Bytes> + 'static,
        E: Into<Box<dyn std::error::Error + Send + Sync>> + 'static,
    {
        let mut query_params: HashMap<String, String> = HashMap::new();
        query_params.insert("path".into(), dst_path.to_string());
        if let Some(options) = options {
            if options.allow_overwrite_dir_with_file {
                query_params.insert("noOverwriteDirNonDir".into(), "true".into());
            }
            if options.copy_uid_gid {
                query_params.insert("copyUIDGID".into(), "true".into());
            }
        }
        let query_params = if !query_params.is_empty() {
            Some(query_params)
        } else {
            None
        };

        let request = Request::builder()
            .method(Method::PUT)
            .uri(self.request_uri(
                &format!(
                    "/containers/{}/archive",
                    utf8_percent_encode(container_id, NON_ALPHANUMERIC).to_string()
                ),
                query_params,
            )?)
            .header("Content-Type", "application/x-tar")
            .body(Body::wrap_stream(content))
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );

        Ok(())
    }

    /// containers_prune requests the daemon to delete unused data.
    pub async fn containers_prune(
        &self,
        prune_filters: Option<filters::Args>,
    ) -> Result<ContainersPruneReport, Error> {
        let query_params = if let Some(prune_filters) = prune_filters {
            Some(get_filters_query(prune_filters)?)
        } else {
            None
        };
        let request = Request::builder()
            .method(Method::POST)
            .uri(self.request_uri("/containers/prune", query_params)?)
            .header("Accept", "application/json")
            .body(Body::empty())
            .context(HttpClientRequestBuilderError {})?;

        let client = self.client.as_ref().unwrap();
        let response = timeout(self.timeout, client.request(request))
            .await
            .context(HttpClientTimeoutError {})?
            .context(HttpClientError {})?;
        ensure!(
            response.status().is_success(),
            HttpClientResponseError {
                status: response.status().as_u16()
            }
        );
        let response_body = read_response_body(response, self.timeout).await?;
        Ok(serde_json::from_str(&response_body).context(JsonDeserializationError {})?)
    }
}

fn get_container_path_stat_from_header(header: &str) -> Result<ContainerPathStat, Error> {
    let contents = base64::decode(header).context(B64DecodingError)?;
    Ok(serde_json::from_str(&String::from_utf8_lossy(&contents))
        .context(JsonDeserializationError {})?)
}

/// A stream of container logs.
pub struct LogStream {
    inner: Pin<Box<Lines<BufReader<AsyncHttpBodyReader>>>>,
}

impl LogStream {
    pub fn new(response: Response<Body>) -> Self {
        Self {
            inner: Box::pin(BufReader::new(AsyncHttpBodyReader::new(response)).lines()),
        }
    }
}

impl Stream for LogStream {
    type Item = Result<String, Error>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        if let Some(line) = ready!(self.inner.as_mut().poll_next(cx)) {
            match line {
                Ok(line) => Poll::Ready(Some(Ok(line))),
                Err(err) => Poll::Ready(Some(Err(err).context(IoError {}))),
            }
        } else {
            Poll::Ready(None)
        }
    }
}

/// A stream of container statistics.
pub struct StatsStream {
    inner: Pin<Box<Lines<BufReader<AsyncHttpBodyReader>>>>,
}

impl StatsStream {
    pub fn new(response: Response<Body>) -> Self {
        Self {
            inner: Box::pin(BufReader::new(AsyncHttpBodyReader::new(response)).lines()),
        }
    }
}

impl Stream for StatsStream {
    type Item = Result<Stats, Error>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        if let Some(line) = ready!(self.inner.as_mut().poll_next(cx)) {
            match line {
                Ok(line) => Poll::Ready(Some(
                    serde_json::from_str(&line).context(JsonDeserializationError),
                )),
                Err(err) => Poll::Ready(Some(Err(err).context(IoError {}))),
            }
        } else {
            Poll::Ready(None)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::container::*;
    use crate::{opts, LocalDockerEngineClient};
    use futures::StreamExt;
    use maplit::hashmap;
    use std::collections::HashSet;
    use std::io::Read;
    use tar::Archive;

    #[tokio::test]
    async fn test_container_api() {
        let docker_client =
            LocalDockerEngineClient::new_client_with_opts(Some(vec![Box::new(opts::from_env)]))
                .unwrap();

        // Create a new ephemeral container.
        let create_response = docker_client
            .container_create(
                ContainerConfigBuilder::default()
                    .image(Some("busybox".into()))
                    .tty(Some(true))
                    .cmd(Some(vec![
                        "/bin/sh".into(),
                        "-c".into(),
                        "while true; do echo 'hello test_container_api'; sleep 1; done".into(),
                    ]))
                    .build()
                    .unwrap(),
                Some("test_container_api".into()),
            )
            .await
            .unwrap();

        // Start the ephemeral container.
        docker_client
            .container_start(&create_response.id, None)
            .await
            .unwrap();

        // Resize the console.
        docker_client
            .container_resize(
                &create_response.id,
                ResizeOptionsBuilder::default()
                    .width(Some(100))
                    .height(Some(20))
                    .build()
                    .unwrap(),
            )
            .await
            .unwrap();

        // Rename the container.
        docker_client
            .container_rename(&create_response.id, "test_container_api_renamed")
            .await
            .unwrap();

        // List the ephemeral container.
        let containers = docker_client
            .container_list(Some(
                ContainerListOptionsBuilder::default()
                    .filters(
                        filters::ArgsBuilder::default()
                            .fields(hashmap! {
                                "id".into() => vec![create_response.id.clone()],
                            })
                            .build()
                            .unwrap(),
                    )
                    .build()
                    .unwrap(),
            ))
            .await
            .unwrap();

        assert_eq!(containers.len(), 1);
        assert_eq!(
            &containers[0].names.as_ref().unwrap()[0],
            "/test_container_api_renamed"
        );
        assert_eq!(&containers[0].image, "busybox");

        // Set a memory limit on the container.
        docker_client
            .container_update(
                &create_response.id,
                UpdateConfigBuilder::default()
                    .memory(Some(10_000_000))
                    .build()
                    .unwrap(),
            )
            .await
            .unwrap();

        // Get detailed container information.
        let container = docker_client
            .container_inspect(&create_response.id)
            .await
            .unwrap();

        assert_eq!(container.host_config.unwrap().memory.unwrap(), 10_000_000);

        // Get container statistics.
        let mut stats = docker_client
            .container_stats(&create_response.id, false)
            .await
            .unwrap();

        let stats = stats.body.next().await.unwrap().unwrap();

        // Reporting a reasonable quantity of memory?
        let memory_stats = stats.memory_stats.unwrap();
        assert!(memory_stats.usage.unwrap() > 500_000);
        assert!(memory_stats.usage.unwrap() < 5_000_000);

        // Get the first log line from the container.
        let mut logs = docker_client
            .container_logs(
                &create_response.id,
                Some(
                    ContainerLogsOptionsBuilder::default()
                        .show_stdout(true)
                        .follow(true)
                        .build()
                        .unwrap(),
                ),
            )
            .await
            .unwrap();

        let log_line = logs.next().await.unwrap().unwrap();
        assert_eq!(&log_line, "hello test_container_api");

        // Get the list of processes inside the ephemeral container.
        let top_response = docker_client
            .container_top(&create_response.id, None)
            .await
            .unwrap();

        let cmd_index = top_response
            .titles
            .iter()
            .position(|title| title.eq("CMD"))
            .unwrap();
        let commands = top_response
            .processes
            .iter()
            .map(|process| process[cmd_index].clone())
            .collect::<Vec<_>>();
        let shell_command = commands
            .iter()
            .find(|command| command.starts_with("/bin/sh"));

        assert!(shell_command.is_some());

        // Stop the ephemeral container.
        docker_client
            .container_stop(&create_response.id, Duration::from_secs(0))
            .await
            .unwrap();

        // Cleanup the ephemeral container.
        docker_client
            .container_remove(
                &create_response.id,
                Some(
                    ContainerRemoveOptionsBuilder::default()
                        .force(true)
                        .build()
                        .unwrap(),
                ),
            )
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_container_lifecycle_api() {
        let docker_client =
            LocalDockerEngineClient::new_client_with_opts(Some(vec![Box::new(opts::from_env)]))
                .unwrap();

        // Create a new ephemeral container.
        let create_response = docker_client
            .container_create(
                ContainerConfigBuilder::default()
                    .image(Some("busybox".into()))
                    .tty(Some(true))
                    .cmd(Some(vec![
                        "/bin/sh".into(),
                        "-c".into(),
                        "for i in 1 2 3; do echo \"$i\"; sleep 1; done".into(),
                    ]))
                    .build()
                    .unwrap(),
                Some("test_container_lifecycle_api".into()),
            )
            .await
            .unwrap();

        // Start the ephemeral container.
        docker_client
            .container_start(&create_response.id, None)
            .await
            .unwrap();

        // Get the first log line from the container.
        let mut logs = docker_client
            .container_logs(
                &create_response.id,
                Some(
                    ContainerLogsOptionsBuilder::default()
                        .show_stdout(true)
                        .follow(true)
                        .build()
                        .unwrap(),
                ),
            )
            .await
            .unwrap();

        let log_line = logs.next().await.unwrap().unwrap();
        let last_log_time = SystemTime::now();
        assert_eq!(&log_line, "1");

        // Restart the ephemeral container.
        docker_client
            .container_restart(&create_response.id, Duration::from_secs(0))
            .await
            .unwrap();

        // Get the first log line from the container.
        let mut logs = docker_client
            .container_logs(
                &create_response.id,
                Some(
                    ContainerLogsOptionsBuilder::default()
                        .show_stdout(true)
                        .follow(true)
                        .since(Some(last_log_time))
                        .build()
                        .unwrap(),
                ),
            )
            .await
            .unwrap();

        let log_line = logs.next().await.unwrap().unwrap();
        assert_eq!(&log_line, "1");

        // Pause the ephemeral container.
        docker_client
            .container_pause(&create_response.id)
            .await
            .unwrap();

        // Should timeout as no logs are written while paused.
        let log_line = timeout(Duration::from_secs(2), logs.next()).await;
        assert!(log_line.is_err());

        // Unpause the ephemeral container.
        docker_client
            .container_unpause(&create_response.id)
            .await
            .unwrap();

        let log_line = logs.next().await.unwrap().unwrap();
        assert_eq!(&log_line, "2");

        // Ephemeral container should exit successfully within one second.
        let exit_status = docker_client
            .container_wait(
                &create_response.id,
                WaitCondition::NextExit,
                Duration::from_secs(5),
            )
            .await
            .unwrap();

        assert_eq!(exit_status.status_code, 0);

        // Cleanup the ephemeral container.
        docker_client
            .container_remove(
                &create_response.id,
                Some(
                    ContainerRemoveOptionsBuilder::default()
                        .force(true)
                        .build()
                        .unwrap(),
                ),
            )
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_container_filesystem_api() {
        let docker_client =
            LocalDockerEngineClient::new_client_with_opts(Some(vec![Box::new(opts::from_env)]))
                .unwrap();

        // Create a new ephemeral container.
        let create_response = docker_client
            .container_create(
                ContainerConfigBuilder::default()
                    .image(Some("busybox".into()))
                    .cmd(Some(vec!["sleep".into(), "60".into()]))
                    .build()
                    .unwrap(),
                Some("test_container_filesystem_api".into()),
            )
            .await
            .unwrap();

        let chunks: Vec<Result<_, std::io::Error>> =
            vec![std::fs::read_to_string("src/test_fixture/hello_world.tar")];
        let content = futures::stream::iter(chunks);

        // Copy a simple hello world text file into the ephemeral container.
        docker_client
            .copy_to_container(&create_response.id, "/", content, None)
            .await
            .unwrap();

        // Confirm the hello text file exists in the ephemeral container.
        let stat_path = docker_client
            .container_stat_path(&create_response.id, "/hello.txt")
            .await
            .unwrap();

        assert_eq!(stat_path.size, 12);

        // Copy the hello text file back out of the ephemeral container.
        let (stat_path, hello_tar) = docker_client
            .copy_from_container(&create_response.id, "/hello.txt")
            .await
            .unwrap();

        assert_eq!(stat_path.size, 12);

        let mut hello_tar = Archive::new(hello_tar.as_slice());
        let mut hello_txt_entry = hello_tar.entries().unwrap().next().unwrap().unwrap();

        // Verify the echoed hello text file contains the expected content.
        let mut s = String::new();
        hello_txt_entry.read_to_string(&mut s).unwrap();
        assert_eq!(&s, "Hello World\n");

        let changes = docker_client
            .container_diff(&create_response.id)
            .await
            .unwrap()
            .unwrap();

        assert_eq!(&changes[0].path, "/hello.txt");

        // Export the contents of the ephemeral container.
        let container_tar = docker_client
            .container_export(&create_response.id)
            .await
            .unwrap();

        // Get the complete list of files from the ephemeral containers bundle.
        let mut container_tar = Archive::new(container_tar.as_slice());
        let container_files = container_tar
            .entries()
            .unwrap()
            .map(|entry| String::from(entry.unwrap().path().unwrap().to_string_lossy()))
            .collect::<HashSet<_>>();

        // Ensure the exported bundle contained our hello world file.
        assert!(container_files.contains("hello.txt"));

        // Cleanup the ephemeral container.
        docker_client
            .container_remove(
                &create_response.id,
                Some(
                    ContainerRemoveOptionsBuilder::default()
                        .force(true)
                        .build()
                        .unwrap(),
                ),
            )
            .await
            .unwrap();
    }
}