Crude Read Support (#3)

Crude Read Support

Author: Empty2k12

Cargo.lock

@@ -10,3 +10,10 @@ futures = "0.1.27"
 tokio = "0.1.20"
 itertools = "0.8"
 failure = "0.1.5"
+serde = { version = "1.0.92", optional = true, features = ["derive"] }
+serde_json = { version = "1.0", optional = true }
+
+[features]
+use-serde = ["serde", "serde_json"]
+
+default = ["use-serde"]

Cargo.toml

@@ -0,0 +1,172 @@
+use futures::{Future, Stream};
+use reqwest::r#async::{Client, Decoder};
+
+use std::mem;
+
+use crate::error::InfluxDbError;
+use crate::query::{InfluxDbQuery, QueryType};
+
+/// Client which can read and write data from InfluxDB.
+///
+/// # Arguments
+///
+///  * `url`: The URL where InfluxDB is running (ex. `http://localhost:8086`).
+///  * `database`: The Database against which queries and writes will be run.
+///
+/// # Examples
+///
+/// ```rust
+/// use influxdb::client::InfluxDbClient;
+///
+/// let client = InfluxDbClient::new("http://localhost:8086", "test");
+///
+/// assert_eq!(client.database_name(), "test");
+/// ```
+pub struct InfluxDbClient {
+    url: String,
+    database: String,
+    // auth: Option<InfluxDbAuthentication>
+}
+
+impl InfluxDbClient {
+    /// Instantiates a new [`InfluxDbClient`]
+    ///
+    /// # Arguments
+    ///
+    ///  * `url`: The URL where InfluxDB is running (ex. `http://localhost:8086`).
+    ///  * `database`: The Database against which queries and writes will be run.
+    ///
+    /// # Examples
+    ///
+    /// ```rust
+    /// use influxdb::client::InfluxDbClient;
+    ///
+    /// let _client = InfluxDbClient::new("http://localhost:8086", "test");
+    /// ```
+    pub fn new<S>(url: S, database: S) -> Self
+    where
+        S: Into<String>,
+    {
+        InfluxDbClient {
+            url: url.into(),
+            database: database.into(),
+        }
+    }
+
+    pub fn database_name<'a>(&'a self) -> &'a str {
+        &self.database
+    }
+
+    pub fn database_url<'a>(&'a self) -> &'a str {
+        &self.url
+    }
+
+    pub fn ping(&self) -> impl Future<Item = (String, String), Error = InfluxDbError> {
+        Client::new()
+            .get(format!("{}/ping", self.url).as_str())
+            .send()
+            .map(|res| {
+                let build = res
+                    .headers()
+                    .get("X-Influxdb-Build")
+                    .unwrap()
+                    .to_str()
+                    .unwrap();
+                let version = res
+                    .headers()
+                    .get("X-Influxdb-Version")
+                    .unwrap()
+                    .to_str()
+                    .unwrap();
+
+                (String::from(build), String::from(version))
+            })
+            .map_err(|err| InfluxDbError::UnspecifiedError {
+                error: format!("{}", err),
+            })
+    }
+
+    pub fn query<Q>(&self, q: Q) -> Box<dyn Future<Item = String, Error = InfluxDbError>>
+    where
+        Q: InfluxDbQuery,
+    {
+        use futures::future;
+
+        let query_type = q.get_type();
+        let endpoint = match query_type {
+            QueryType::ReadQuery => "query",
+            QueryType::WriteQuery => "write",
+        };
+
+        let query = match q.build() {
+            Err(err) => {
+                let error = InfluxDbError::UnspecifiedError {
+                    error: format!("{}", err),
+                };
+                return Box::new(future::err::<String, InfluxDbError>(error));
+            }
+            Ok(query) => query,
+        };
+
+        let query_str = query.get();
+        let url_params = match query_type {
+            QueryType::ReadQuery => format!("&q={}", query_str),
+            QueryType::WriteQuery => String::from(""),
+        };
+
+        let client = match query_type {
+            QueryType::ReadQuery => Client::new().get(
+                format!(
+                    "{url}/{endpoint}?db={db}{url_params}",
+                    url = self.url,
+                    endpoint = endpoint,
+                    db = self.database,
+                    url_params = url_params
+                )
+                .as_str(),
+            ),
+            QueryType::WriteQuery => Client::new()
+                .post(
+                    format!(
+                        "{url}/{endpoint}?db={db}",
+                        url = self.url,
+                        endpoint = endpoint,
+                        db = self.database,
+                    )
+                    .as_str(),
+                )
+                .body(query_str),
+        };
+
+        Box::new(
+            client
+                .send()
+                .and_then(|mut res| {
+                    let body = mem::replace(res.body_mut(), Decoder::empty());
+                    body.concat2()
+                })
+                .map_err(|err| InfluxDbError::UnspecifiedError {
+                    error: format!("{}", err),
+                })
+                .and_then(|body| {
+                    if let Ok(utf8) = std::str::from_utf8(&body) {
+                        let mut s = String::new();
+                        utf8.clone_into(&mut s);
+
+                        // todo: improve error parsing without serde
+                        if s.contains("\"error\"") {
+                            return futures::future::err(InfluxDbError::UnspecifiedError {
+                                error: format!("influxdb error: \"{}\"", s),
+                            });
+                        }
+
+                        return futures::future::ok(s);
+                    }
+
+                    futures::future::err(InfluxDbError::UnspecifiedError {
+                        error: "some other error has happened here!".to_string(),
+                    })
+                }),
+        )
+    }
+}

src/client/mod.rs

@@ -0,0 +1,18 @@
+#[derive(Debug, Fail)]
+/// Errors that might happen in the crate
+pub enum InfluxDbError {
+    #[fail(display = "query must contain at least one field")]
+    /// Error happens when query has zero fields
+    InvalidQueryError,
+
+    #[fail(
+        display = "an error happened: \"{}\". this case should be handled better, please file an issue.",
+        error
+    )]
+    /// todo: Error which is a placeholder for more meaningful errors. This should be refactored away.
+    UnspecifiedError { error: String },
+
+    #[fail(display = "InfluxDB encountered the following error: {}", error)]
+    /// Error which has happened inside InfluxDB
+    DatabaseError { error: String },
+}

src/error.rs

@@ -0,0 +1,126 @@
+use crate::client::InfluxDbClient;
+
+use serde::de::DeserializeOwned;
+
+use futures::{Future, Stream};
+use reqwest::r#async::{Client, Decoder};
+
+use serde_json;
+use serde::Deserialize;
+use std::mem;
+
+use crate::error::InfluxDbError;
+use crate::query::{InfluxDbQuery, QueryType};
+
+#[derive(Deserialize)]
+#[doc(hidden)]
+struct _DatabaseError {
+    error: String,
+}
+
+#[derive(Deserialize, Debug)]
+#[doc(hidden)]
+pub struct DatabaseQueryResult<T> {
+    pub results: Vec<InfluxDbReturn<T>>,
+}
+
+#[derive(Deserialize, Debug)]
+#[doc(hidden)]
+pub struct InfluxDbReturn<T> {
+    pub series: Option<Vec<InfluxDbSeries<T>>>,
+}
+
+#[derive(Deserialize, Debug)]
+#[doc(hidden)]
+pub struct InfluxDbSeries<T> {
+    pub name: String,
+    pub values: Vec<T>,
+}
+
+impl InfluxDbClient {
+    pub fn json_query<T: 'static, Q>(&self, q: Q) -> Box<dyn Future<Item = Option<Vec<InfluxDbSeries<T>>>, Error = InfluxDbError>>
+    where
+        Q: InfluxDbQuery,
+        T: DeserializeOwned,
+    {
+        use futures::future;
+
+        let query_type = q.get_type();
+        let endpoint = match query_type {
+            QueryType::ReadQuery => "query",
+            QueryType::WriteQuery => "write",
+        };
+
+        let query = match q.build() {
+            Err(err) => {
+                let error = InfluxDbError::UnspecifiedError {
+                    error: format!("{}", err),
+                };
+                return Box::new(future::err::<Option<Vec<InfluxDbSeries<T>>>, InfluxDbError>(error));
+            }
+            Ok(query) => query,
+        };
+
+        let query_str = query.get();
+        let url_params = match query_type {
+            QueryType::ReadQuery => format!("&q={}", query_str),
+            QueryType::WriteQuery => String::from(""),
+        };
+
+        let client = match query_type {
+            QueryType::ReadQuery => Client::new().get(
+                format!(
+                    "{url}/{endpoint}?db={db}{url_params}",
+                    url = self.database_url(),
+                    endpoint = endpoint,
+                    db = self.database_name(),
+                    url_params = url_params
+                )
+                .as_str(),
+            ),
+            QueryType::WriteQuery => Client::new()
+                .post(
+                    format!(
+                        "{url}/{endpoint}?db={db}",
+                        url = self.database_url(),
+                        endpoint = endpoint,
+                        db = self.database_name(),
+                    )
+                    .as_str(),
+                )
+                .body(query_str),
+        };
+
+        Box::new(
+            client
+                .send()
+                .and_then(|mut res| {
+                    let body = mem::replace(res.body_mut(), Decoder::empty());
+                    body.concat2()
+                })
+                .map_err(|err| InfluxDbError::UnspecifiedError {
+                    error: format!("{}", err)
+                })
+                .and_then(|body| {
+                    // Try parsing InfluxDBs { "error": "error message here" }
+                    if let Ok(error) = serde_json::from_slice::<_DatabaseError>(&body) {
+                        return futures::future::err(InfluxDbError::DatabaseError {
+                            error: error.error.to_string()
+                        })
+                    } else {
+                        // Json has another structure, let's try actually parsing it to the type we're deserializing 
+                        let from_slice = serde_json::from_slice::<DatabaseQueryResult<T>>(&body);
+
+                        let mut deserialized = match from_slice {
+                            Ok(deserialized) => deserialized,
+                            Err(err) => return futures::future::err(InfluxDbError::UnspecifiedError {
+                                error: format!("serde error: {}", err)
+                            })
+                        };
+
+                        return futures::future::result(Ok(deserialized.results.remove(0).series));
+                    }
+                })
+        )
+    }
+}