Updates

Author: Empty2k12

Cargo.toml

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

src/client/mod.rs

@@ -86,7 +86,7 @@ impl InfluxDbClient {
             })
     }
 
-    pub fn query<Q>(self, q: Q) -> Box<dyn Future<Item = String, Error = InfluxDbError>>
+    pub fn query<Q>(&self, q: Q) -> Box<dyn Future<Item = String, Error = InfluxDbError>>
     where
         Q: InfluxDbQuery,
     {

src/integrations/serde_integration.rs

@@ -18,14 +18,27 @@ struct _DatabaseError {
     error: String,
 }
 
-pub trait InfluxDbSerdeORM {
-    fn json_query<T: 'static, Q>(self, q: Q) -> Box<dyn Future<Item = T, Error = InfluxDbError>> where
-        Q: InfluxDbQuery,
-        T: DeserializeOwned;
+#[derive(Deserialize, Debug)]
+#[doc(hidden)]
+pub struct DatabaseQueryResult<T> {
+    pub results: Vec<InfluxDbReturn<T>>,
 }
 
-impl InfluxDbSerdeORM for InfluxDbClient {
-    fn json_query<T: 'static, Q>(self, q: Q) -> Box<dyn Future<Item = T, Error = InfluxDbError>>
+#[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<T>>, Error = InfluxDbError>>
     where
         Q: InfluxDbQuery,
         T: DeserializeOwned,
@@ -43,7 +56,7 @@ impl InfluxDbSerdeORM for InfluxDbClient {
                 let error = InfluxDbError::UnspecifiedError {
                     error: format!("{}", err),
                 };
-                return Box::new(future::err::<T, InfluxDbError>(error));
+                return Box::new(future::err::<Option<Vec<T>>, InfluxDbError>(error));
             }
             Ok(query) => query,
         };
@@ -89,18 +102,28 @@ impl InfluxDbSerdeORM for InfluxDbClient {
                     error: format!("{}", err)
                 })
                 .and_then(|body| {
+                    println!("{:?}", &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 if let Ok(t_result) = serde_json::from_slice::<T>(&body) {
-                        // Json has another structure, let's try actually parsing it to the type we're deserializing to
-                        return futures::future::result(Ok(t_result));
                     } else {
-                        return futures::future::err(InfluxDbError::UnspecifiedError {
-                            error: "something wen't wrong during deserializsation of the database response. this might be a bug!".to_string()
-                        })
+                        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)
+                            })
+                        };
+
+                        // Json has another structure, let's try actually parsing it to the type we're deserializing to
+                        let t_result = match deserialized.results.remove(0).series {
+                            Some(series) => Ok(Some(series.into_iter().flat_map(|x| { x.values }).collect::<Vec<T>>())),
+                            None => Ok(None)
+                        };
+                        return futures::future::result(t_result);
                     }
                 })
         )

src/lib.rs

@@ -8,8 +8,9 @@ pub mod client;
 pub mod error;
 pub mod query;
 
+#[cfg(feature = "use-serde")]
 pub mod integrations {
-    #[cfg(feature = "serde-orm")]
+    #[cfg(feature = "use-serde")]
     pub mod serde_integration;
 }
 

tests/integration_tests.rs

@@ -47,7 +47,26 @@ fn test_read() {
     let client = create_client();
     let query = InfluxDbQuery::raw_read_query("SELECT * FROM weather");
     let result = get_runtime().block_on(client.query(query));
-    println!("{:?}", result);
     assert!(result.is_ok(), "Should be no error");
-    println!("{}", result.unwrap());
-}
+}
+
+#[test]
+#[cfg(feature = "use-serde")]
+/// INTEGRATION TEST
+///
+/// This test case tests whether the InfluxDB server can be connected to and gathers info about it
+fn test_json() {
+    use serde::Deserialize;
+
+    #[derive(Deserialize, Debug)]
+    struct Weather {
+        time: String,
+        temperature: i32,
+    }
+
+    let client = create_client();
+    let query = InfluxDbQuery::raw_read_query("SELECT * FROM weather");
+    let result = get_runtime().block_on(client.json_query::<Weather, _>(query));
+
+    assert!(result.is_ok(), "We could read from the DB");
+}