You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2015/02/10 06:56:22 UTC
[2/2] camel git commit: CAMEL-8299: Restored old behavior and added
BULK as operation, to return the BulkResponse.
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 <da...@apache.org>
Authored: Tue Feb 10 06:56:41 2015 +0100
Committer: Claus Ibsen <da...@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");
}