ocpi/

client.rs

use crate::{
    context::ExtendedContext,
    error::ServerError,
    types::{self, CommandResult},
    Context, Result, Session,
};
use http::HeaderMap;
use reqwest::Method;

#[derive(serde::Deserialize)]
struct Reply<T> {
    pub status_code: u32,

    pub data: Option<T>,

    #[serde(rename = "status_message")]
    pub message: Option<String>,

    #[allow(dead_code)]
    pub timestamp: types::DateTime,
}

/// Implements the different OCPI calls as a client.
pub struct Client {
    http: reqwest::Client,
}

impl Clone for Client {
    fn clone(&self) -> Self {
        Self {
            http: self.http.clone(),
        }
    }
}

impl Default for Client {
    fn default() -> Self {
        const CARGO_PKG_VERSION: &str = env!("CARGO_PKG_VERSION");

        let mut def_headers = HeaderMap::new();
        def_headers.append(
            "user-agent",
            format!("ocpi-rs {}", CARGO_PKG_VERSION)
                .parse()
                .expect("Invalid CARGO_PKG_VERSION"),
        );

        Self {
            http: reqwest::Client::builder()
                .default_headers(def_headers)
                .build()
                .expect("Building default OCPI client"),
        }
    }
}

impl Client {
    pub fn new(http: reqwest::Client) -> Self {
        Self { http }
    }

    fn req(
        &self,
        ctx: ExtendedContext<'_>,
        method: reqwest::Method,
        url: impl Into<String>,
    ) -> ReqBuilder {
        let url = url.into();
        let b = self.http.request(method, &url).set_ocpi_ctx(ctx);
        ReqBuilder { url, b }
    }

    /// Given a version number and the URL of the
    /// root OCPI versions endpoint and the client.
    /// The function first calls the versions URL to confirm
    /// that the required version exists.
    /// And then retrieves the available endpoints for that version.
    pub async fn get_endpoints_for_version(
        &self,
        ctx: ExtendedContext<'_>,
        versions_url: types::Url,
        desired_version: types::VersionNumber,
    ) -> Result<types::VersionDetails> {
        let versions = self
            .req(ctx, Method::GET, versions_url.clone())
            .send::<Vec<types::Version>>()
            .await?;

        // Try to find the matching version.
        let version = versions
            .into_iter()
            .find(|v| v.version == desired_version)
            .ok_or(ServerError::IncompatibleEndpoints)?;

        let version_details = self
            .req(ctx, Method::GET, version.url.clone())
            .send::<types::VersionDetails>()
            .await?;

        Ok(version_details)
    }

    pub async fn post_response(
        &self,
        ctx: ExtendedContext<'_>,
        url: &url::Url,
        command_result: CommandResult,
    ) -> Result<()> {
        self.req(ctx, Method::GET, url.to_string())
            .body(&command_result)
            .send()
            .await
    }

    pub async fn put_session(&self, ctx: &Context, url: &url::Url, session: Session) -> Result<()> {
        let cc = session.country_code.as_str();
        let pid = session.party_id.as_str();
        let id = session.id.as_str();

        let url = format!("{url}?country_code={cc}&party_id={pid}&session_id={id}");

        self.req(ctx.as_extended(), Method::PUT, url)
            .body(&session)
            .send()
            .await
    }
}

struct ReqBuilder {
    // Storing url here for better errors.
    url: String,
    b: reqwest::RequestBuilder,
}

impl ReqBuilder {
    fn body<T: serde::Serialize + ?Sized>(mut self, b: &T) -> Self {
        self.b = self.b.json(b);
        self
    }

    async fn send<T>(self) -> Result<T>
    where
        T: serde::de::DeserializeOwned,
    {
        let url = self.url;

        let rep = self
            .b
            .send()
            .await
            .map_err(|err| ServerError::unusable_api(err.to_string()))?;

        let status = rep.status();

        if !status.is_success() {
            return Err(ServerError::unusable_api(format!(
                "Non 2xx-reply: {} from server",
                status.as_u16()
            )))?;
        }

        let body = rep.json::<Reply<T>>().await.map_err(|err| {
            ServerError::unusable_api(format!(
                "Error parsing result from `{}` as json: {}",
                url, err
            ))
        })?;

        let code = body.status_code / 100;
        // 1000 is generic success.
        // 19 is custom success range.
        if !(code == 10 || code == 19) {
            return Err(ServerError::unusable_api(format!(
                "Non non success status_code `{} ({})` reply from `{}`. With message: `{}`",
                body.status_code,
                code,
                url,
                body.message.as_deref().unwrap_or("")
            )))?;
        }

        match body.data {
            Some(body) => Ok(body),
            None => Err(ServerError::unusable_api(format!(
                "Received unexpected empty body from `{}` with message: `{}`",
                url,
                body.message.as_deref().unwrap_or("")
            )))?,
        }
    }
}

trait SetOcpiCtx {
    fn set_ocpi_ctx(self, ctx: ExtendedContext<'_>) -> Self;
}

impl SetOcpiCtx for reqwest::RequestBuilder {
    fn set_ocpi_ctx(self, ctx: ExtendedContext<'_>) -> Self {
        let b64 = base64::encode(ctx.credentials_token.as_str());
        self.header("Authorization", format!("Token {}", b64))
            .header("X-Request-Id", ctx.request_id)
            .header("X-Correlation-Id", ctx.correlation_id)
    }
}

#[cfg(test)]
mod tests {

    use super::*;
    use crate::context::test_ctx;
    use serde_json::json;
    use wiremock::{matchers, Mock, MockServer, ResponseTemplate};

    #[tokio::test]
    #[rustfmt::skip::macros(json)]
    async fn test_endpoints_for_version() {
        let cli = Client::default();
        let mock = MockServer::start().await;

        Mock::given(matchers::method("GET"))
            .and(matchers::path("/versions"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!(
		{
		    "status_code": 1000,
		    "status_message": "Success",
		    "timestamp": "2015-06-30T21:59:59Z",
		    "data": 
		    [
			{
			    "version": "2.1.1",
			    "url": "http://www.server.com/ocpi/2.1.1/"
			},
			{
			    "version": "2.2",
			    "url": format!("{}/2.2", mock.uri())
			}
		    ]
		}
            )))
            .mount(&mock)
            .await;

        Mock::given(matchers::method("GET"))
            .and(matchers::path("/2.2"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!(
		{
		    "status_code": 1000,
		    "status_message": "Success",
		    "timestamp": "2015-06-30T21:59:59Z",
		    "data": {
			"version": "2.2",
			"endpoints": [
			    {
				"identifier": "credentials",
				"role": "SENDER",
				"url": format!("{}/2.2/credentials", mock.uri())
			    }
			]
		    }
		}
	    )))
            .mount(&mock)
            .await;

        let versions_url = format!("{}/versions", mock.uri())
            .parse::<types::Url>()
            .expect("Versions url");

        let details = cli
            .get_endpoints_for_version(
                test_ctx().extend(&"imatoken".parse().unwrap()),
                versions_url.clone(),
                types::VersionNumber::V2_2,
            )
            .await
            .expect(&format!("Making request to {}", versions_url));

        assert_eq!(details.version, types::VersionNumber::V2_2);
        assert_eq!(details.endpoints.len(), 1);
        assert_eq!(
            details.endpoints[0].identifier,
            types::ModuleId::Credentials
        );
        assert_eq!(details.endpoints[0].role, types::InterfaceRole::Sender);
    }
}