camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [2/2] camel git commit: CAMEL-8299: Restored old behavior and added BULK as operation, to return the BulkResponse.
Date Tue, 10 Feb 2015 05:56:22 GMT
CAMEL-8299: Restored old behavior and added BULK as operation, to return the BulkResponse.


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/4641a45c
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/4641a45c
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/4641a45c

Branch: refs/heads/master
Commit: 4641a45c227c169452e42dbd369537760dc48712
Parents: e4d991e
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Tue Feb 10 06:56:41 2015 +0100
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Tue Feb 10 06:56:41 2015 +0100

----------------------------------------------------------------------
 .../ElasticsearchConfiguration.java             |  3 ++-
 .../elasticsearch/ElasticsearchProducer.java    | 21 ++++++++++++++++++--
 .../ElasticsearchComponentTest.java             | 21 +++++++++++++++++++-
 3 files changed, 41 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/4641a45c/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchConfiguration.java
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchConfiguration.java
index 22f9264..90a6361 100644
--- a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchConfiguration.java
+++ b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchConfiguration.java
@@ -34,6 +34,7 @@ public class ElasticsearchConfiguration {
 
     public static final String PARAM_OPERATION = "operation";
     public static final String OPERATION_INDEX = "INDEX";
+    public static final String OPERATION_BULK = "BULK";
     public static final String OPERATION_BULK_INDEX = "BULK_INDEX";
     public static final String OPERATION_GET_BY_ID = "GET_BY_ID";
     public static final String OPERATION_DELETE = "DELETE";
@@ -62,7 +63,7 @@ public class ElasticsearchConfiguration {
     private boolean local;
     @UriParam
     private Boolean data;
-    @UriParam
+    @UriParam(enums = "INDEX,BULK,BULK_INDEX,GET_BY_ID,DELETE")
     private String operation;
     @UriParam
     private String ip;

http://git-wip-us.apache.org/repos/asf/camel/blob/4641a45c/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchProducer.java
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchProducer.java
index 75ecd19..c60a1df 100644
--- a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchProducer.java
+++ b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchProducer.java
@@ -16,9 +16,14 @@
  */
 package org.apache.camel.component.elasticsearch;
 
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+
 import org.apache.camel.Exchange;
 import org.apache.camel.Message;
 import org.apache.camel.impl.DefaultProducer;
+import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.get.GetRequest;
@@ -56,7 +61,12 @@ public class ElasticsearchProducer extends DefaultProducer {
         } else if (request instanceof GetRequest) {
             return ElasticsearchConfiguration.OPERATION_GET_BY_ID;
         } else if (request instanceof BulkRequest) {
-            return ElasticsearchConfiguration.OPERATION_BULK_INDEX;
+            // do we want bulk or bulk_index?
+            if ("BULK_INDEX".equals(getEndpoint().getConfig().getOperation())) {
+                return ElasticsearchConfiguration.OPERATION_BULK_INDEX;
+            } else {
+                return ElasticsearchConfiguration.OPERATION_BULK;
+            }
         } else if (request instanceof DeleteRequest) {
             return ElasticsearchConfiguration.OPERATION_DELETE;
         }
@@ -108,9 +118,16 @@ public class ElasticsearchProducer extends DefaultProducer {
         } else if (ElasticsearchConfiguration.OPERATION_GET_BY_ID.equals(operation)) {
             GetRequest getRequest = message.getBody(GetRequest.class);
             message.setBody(client.get(getRequest));
-        } else if (ElasticsearchConfiguration.OPERATION_BULK_INDEX.equals(operation)) {
+        } else if (ElasticsearchConfiguration.OPERATION_BULK.equals(operation)) {
             BulkRequest bulkRequest = message.getBody(BulkRequest.class);
             message.setBody(client.bulk(bulkRequest).actionGet());
+        } else if (ElasticsearchConfiguration.OPERATION_BULK_INDEX.equals(operation)) {
+            BulkRequest bulkRequest = message.getBody(BulkRequest.class);
+            List<String> indexedIds = new ArrayList<String>();
+            for (BulkItemResponse response : client.bulk(bulkRequest).actionGet().getItems())
{
+                indexedIds.add(response.getId());
+            }
+            message.setBody(indexedIds);
         } else if (ElasticsearchConfiguration.OPERATION_DELETE.equals(operation)) {
             DeleteRequest deleteRequest = message.getBody(DeleteRequest.class);
             message.setBody(client.delete(deleteRequest).actionGet());

http://git-wip-us.apache.org/repos/asf/camel/blob/4641a45c/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchComponentTest.java
----------------------------------------------------------------------
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchComponentTest.java
b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchComponentTest.java
index 62ab6a7..88b7299 100644
--- a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchComponentTest.java
+++ b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchComponentTest.java
@@ -24,6 +24,7 @@ import java.util.Map;
 import org.apache.camel.builder.RouteBuilder;
 import org.apache.camel.test.junit4.CamelTestSupport;
 import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.delete.DeleteResponse;
 import org.elasticsearch.action.get.GetRequest;
@@ -268,7 +269,8 @@ public class ElasticsearchComponentTest extends CamelTestSupport {
     }
 
     @Test
-    public void bulkRequestBody() throws Exception {
+    @SuppressWarnings("unchecked")
+    public void bulkIndexRequestBody() throws Exception {
         // given
         BulkRequest request = new BulkRequest();
         request.add(new IndexRequest("foo", "bar", "baz")
@@ -284,6 +286,22 @@ public class ElasticsearchComponentTest extends CamelTestSupport {
         assertThat(indexedDocumentIds, hasItem("baz"));
     }
 
+    @Test
+    public void bulkRequestBody() throws Exception {
+        // given
+        BulkRequest request = new BulkRequest();
+        request.add(new IndexRequest("foo", "bar", "baz")
+                .source("{\"content\": \"hello\"}"));
+
+        // when
+        BulkResponse response = template.requestBody(
+                "direct:bulk", request, BulkResponse.class);
+
+        // then
+        assertThat(response, notNullValue());
+        assertEquals("baz", response.getItems()[0].getId());
+    }
+
     @Override
     protected RouteBuilder createRouteBuilder() throws Exception {
         return new RouteBuilder() {
@@ -294,6 +312,7 @@ public class ElasticsearchComponentTest extends CamelTestSupport {
                 from("direct:get").to("elasticsearch://local?operation=GET_BY_ID&indexName=twitter&indexType=tweet");
                 from("direct:delete").to("elasticsearch://local?operation=DELETE&indexName=twitter&indexType=tweet");
                 from("direct:bulk_index").to("elasticsearch://local?operation=BULK_INDEX&indexName=twitter&indexType=tweet");
+                from("direct:bulk").to("elasticsearch://local?operation=BULK&indexName=twitter&indexType=tweet");
                 //from("direct:indexWithIp").to("elasticsearch://elasticsearch?operation=INDEX&indexName=twitter&indexType=tweet&ip=localhost");
                 //from("direct:indexWithIpAndPort").to("elasticsearch://elasticsearch?operation=INDEX&indexName=twitter&indexType=tweet&ip=localhost&port=9300");
             }


Mime
View raw message