fix invalid bulk index format

Author: KodrAus

src/elastic/src/client/requests/bulk/operation.rs

@@ -189,15 +189,15 @@ where
         }
     }
 
-    pub fn index(self, doc: TDocument) -> BulkOperation<Doc<TDocument>> {
+    pub fn index(self, doc: TDocument) -> BulkOperation<TDocument> {
         BulkOperation {
             action: Action::Index,
             header: BulkHeader {
                 index: Some(Index::from(doc.index().into_owned())),
                 ty: Some(Type::from(doc.ty().into_owned())),
                 id: doc.partial_id().map(|id| Id::from(id.into_owned())),
             },
-            inner: Some(Doc::value(doc)),
+            inner: Some(doc),
         }
     }
 
@@ -256,15 +256,15 @@ where
         .script_fluent(builder)
     }
 
-    pub fn create(self, doc: TDocument) -> BulkOperation<Doc<TDocument>> {
+    pub fn create(self, doc: TDocument) -> BulkOperation<TDocument> {
         BulkOperation {
             action: Action::Create,
             header: BulkHeader {
                 index: Some(Index::from(doc.index().into_owned())),
                 ty: Some(Type::from(doc.ty().into_owned())),
                 id: doc.partial_id().map(|id| Id::from(id.into_owned())),
             },
-            inner: Some(Doc::value(doc)),
+            inner: Some(doc),
         }
     }
 

tests/run/Cargo.toml

@@ -17,4 +17,5 @@ tokio-threadpool = "~0.1"
 tokio-timer = "~0.2"
 term-painter = "~0.2"
 clap = "~2"
+log = "~0.4"
 env_logger = "~0.6"

tests/run/src/bulk/index_get.rs

@@ -0,0 +1,58 @@
+use elastic::error::Error;
+use elastic::prelude::*;
+use futures::Future;
+use run_tests::IntegrationTest;
+
+#[derive(Debug, Clone, Copy)]
+pub struct IndexGet;
+
+#[derive(Debug, PartialEq, Serialize, Deserialize, ElasticType)]
+#[elastic(ty = "document", index = "bulk_index")]
+pub struct Doc {
+    #[elastic(id)]
+    id: String,
+    title: String,
+    timestamp: Date<DefaultDateMapping>,
+}
+
+impl IntegrationTest for IndexGet {
+    type Response = GetResponse<Doc>;
+
+    fn kind() -> &'static str {
+        "bulk"
+    }
+    fn name() -> &'static str {
+        "index"
+    }
+
+    // Ensure the index doesn't exist
+    fn prepare(&self, client: AsyncClient) -> Box<Future<Item = (), Error = Error>> {
+        let delete_res = client.index(Doc::static_index()).delete().send().map(|_| ());
+
+        Box::new(delete_res)
+    }
+
+    // Index some bulk documents
+    fn request(&self, client: AsyncClient) -> Box<Future<Item = Self::Response, Error = Error>> {
+        let ops = (0..10).into_iter().map(|i| bulk().index(Doc {
+            id: i.to_string(),
+            title: "A document title".to_owned(),
+            timestamp: Date::build(2017, 03, 24, 13, 44, 0, 0),
+        }));
+
+        let bulk_res = client
+            .bulk()
+            .extend(ops)
+            .params_fluent(|p| p.url_param("refresh", true))
+            .send();
+
+        let get_res = client.document().get("4").send();
+
+        Box::new(bulk_res.and_then(|_| get_res))
+    }
+
+    // Ensure the response contains the expected document
+    fn assert_ok(&self, res: &Self::Response) -> bool {
+        res.document().map(|doc| &*doc.id) == Some("4")
+    }
+}

tests/run/src/bulk/mod.rs

@@ -4,6 +4,7 @@ use run_tests::{
 };
 
 mod delete;
+mod index_get;
 mod index_create;
 mod stream;
 mod stream_tiny_size_limit;
@@ -13,6 +14,7 @@ mod stream_tiny_timeout;
 pub fn tests() -> Vec<Test> {
     vec![
         Box::new(|client| test(client, delete::Delete)),
+        Box::new(|client| test(client, index_get::IndexGet)),
         Box::new(|client| test(client, index_create::IndexCreate)),
         Box::new(|client| test(client, stream::BulkStream)),
         Box::new(|client| test(client, stream_tiny_size_limit::BulkStreamTinySize)),

tests/run/src/document/mod.rs

@@ -7,6 +7,7 @@ mod compile_test;
 
 mod delete;
 mod simple_index_get;
+mod simple_mapping;
 mod update_no_index;
 mod update_with_doc;
 mod update_with_inline_script;
@@ -15,6 +16,7 @@ mod update_with_script;
 pub fn tests() -> Vec<Test> {
     vec![
         Box::new(|client| test(client, simple_index_get::SimpleIndexGet)),
+        Box::new(|client| test(client, simple_mapping::SimpleMapping)),
         Box::new(|client| test(client, update_with_doc::UpdateWithDoc)),
         Box::new(|client| test(client, update_with_script::UpdateWithScript)),
         Box::new(|client| test(client, update_with_inline_script::UpdateWithInlineScript)),

tests/run/src/document/simple_mapping.rs

@@ -0,0 +1,85 @@
+use elastic::error::Error;
+use elastic::prelude::*;
+use futures::Future;
+use run_tests::IntegrationTest;
+
+use serde_json::Value;
+
+#[derive(Debug, Clone, Copy)]
+pub struct SimpleMapping;
+
+#[derive(Debug, PartialEq, Serialize, Deserialize, ElasticType)]
+#[elastic(ty = "document", index = "simple_mapping")]
+pub struct Doc {
+    #[elastic(id)]
+    id: String,
+    timestamp: Date<DefaultDateMapping>,
+}
+
+impl IntegrationTest for SimpleMapping {
+    type Response = Value;
+
+    fn kind() -> &'static str {
+        "document"
+    }
+    fn name() -> &'static str {
+        "simple mapping"
+    }
+
+    // Ensure the index doesn't exist
+    fn prepare(&self, client: AsyncClient) -> Box<Future<Item = (), Error = Error>> {
+        let delete_res = client
+            .index(Doc::static_index())
+            .delete()
+            .send()
+            .map(|_| ());
+
+        Box::new(delete_res)
+    }
+
+    // Put the document mapping, then get it back from Elasticsearch
+    fn request(&self, client: AsyncClient) -> Box<Future<Item = Self::Response, Error = Error>> {
+        let create_index = client.index(Doc::static_index()).create().send().map(|_| ());
+
+        let put_mapping = client
+            .document::<Doc>()
+            .put_mapping()
+            .send();
+
+        let get_mapping = client
+            .request(IndicesGetMappingRequest::for_index_ty(Doc::static_index(), Doc::static_ty()))
+            .send()
+            .and_then(|res| res.into_response::<Value>());
+
+        Box::new(create_index.and_then(|_| put_mapping).and_then(|_| get_mapping))
+    }
+
+    // Ensure the response contains the expected document
+    fn assert_ok(&self, res: &Self::Response) -> bool {
+        let expected = json!({
+            "simple_mapping": {
+                "mappings": {
+                    "document": {
+                        "properties": {
+                            "id": {
+                                "type": "text",
+                                "fields": {
+                                    "keyword": {
+                                        "type": "keyword",
+                                        "ignore_above": 256
+                                    }
+                                }
+                            },
+                            "timestamp": {
+                                "type": "date",
+                                "format": "basic_date_time"
+                            }
+                        }
+                    }
+                }
+            }
+        });
+
+        res.to_string() == expected.to_string()
+    }
+}