You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ni...@apache.org on 2014/08/13 10:17:59 UTC
[1/2] git commit: Fix CAMEL-7681 : Add Bulk Index mode to
Elasticsearch component Add BULK_INDEX method Add unit test that index 2 Map
in a row
Repository: camel
Updated Branches:
refs/heads/master 1d3682d3f -> fb7752760
Fix CAMEL-7681 : Add Bulk Index mode to Elasticsearch component
Add BULK_INDEX method
Add unit test that index 2 Map in a row
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/cea796a7
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/cea796a7
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/cea796a7
Branch: refs/heads/master
Commit: cea796a78e1e5b50c6342a46c4ca2913214c7a85
Parents: 1d3682d
Author: sebbrousse <se...@gmail.com>
Authored: Mon Aug 11 23:15:52 2014 +0200
Committer: Willem Jiang <wi...@gmail.com>
Committed: Wed Aug 13 15:52:16 2014 +0800
----------------------------------------------------------------------
.../ElasticsearchConfiguration.java | 10 ++-
.../elasticsearch/ElasticsearchProducer.java | 83 ++++++++++++++++----
.../ElasticsearchComponentTest.java | 24 +++++-
3 files changed, 96 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/cea796a7/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 e1e8aa6..1703d50 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
@@ -16,14 +16,15 @@
*/
package org.apache.camel.component.elasticsearch;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.Map;
-
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriParams;
import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeBuilder;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Map;
+
import static org.elasticsearch.node.NodeBuilder.nodeBuilder;
@UriParams
@@ -31,6 +32,7 @@ public class ElasticsearchConfiguration {
public static final String PARAM_OPERATION = "operation";
public static final String OPERATION_INDEX = "INDEX";
+ 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";
public static final String PARAM_INDEX_ID = "indexId";
http://git-wip-us.apache.org/repos/asf/camel/blob/cea796a7/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 3df10a2..2d190b0 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,13 +16,14 @@
*/
package org.apache.camel.component.elasticsearch;
-import java.util.Map;
-
import org.apache.camel.Exchange;
import org.apache.camel.ExpectedBodyTypeException;
import org.apache.camel.Message;
import org.apache.camel.impl.DefaultProducer;
import org.elasticsearch.action.ListenableActionFuture;
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
@@ -30,6 +31,10 @@ import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.xcontent.XContentBuilder;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
/**
* Represents an Elasticsearch producer.
*/
@@ -62,6 +67,8 @@ public class ElasticsearchProducer extends DefaultProducer {
getById(client, exchange);
} else if (operation.equalsIgnoreCase(ElasticsearchConfiguration.OPERATION_DELETE)) {
deleteById(client, exchange);
+ } else if (operation.equalsIgnoreCase(ElasticsearchConfiguration.OPERATION_BULK_INDEX)) {
+ addToIndexUsingBulk(client, exchange);
} else {
throw new IllegalArgumentException(ElasticsearchConfiguration.PARAM_OPERATION + " value '" + operation + "' is not supported");
}
@@ -114,7 +121,9 @@ public class ElasticsearchProducer extends DefaultProducer {
IndexRequestBuilder prepareIndex = client.prepareIndex(indexName, indexType);
- if (!setIndexRequestSource(exchange.getIn(), prepareIndex)) {
+ Object document = extractDocumentFromMessage(exchange.getIn());
+
+ if (!setIndexRequestSource(document, prepareIndex)) {
throw new ExpectedBodyTypeException(exchange, XContentBuilder.class);
}
ListenableActionFuture<IndexResponse> future = prepareIndex.execute();
@@ -122,10 +131,45 @@ public class ElasticsearchProducer extends DefaultProducer {
exchange.getIn().setBody(response.getId());
}
- @SuppressWarnings("unchecked")
- private boolean setIndexRequestSource(Message msg, IndexRequestBuilder builder) {
+ public void addToIndexUsingBulk(Client client, Exchange exchange) {
+ String indexName = exchange.getIn().getHeader(ElasticsearchConfiguration.PARAM_INDEX_NAME, String.class);
+ if (indexName == null) {
+ indexName = getEndpoint().getConfig().getIndexName();
+ }
+
+ String indexType = exchange.getIn().getHeader(ElasticsearchConfiguration.PARAM_INDEX_TYPE, String.class);
+ if (indexType == null) {
+ indexType = getEndpoint().getConfig().getIndexType();
+ }
+
+ log.debug("Preparing Bulk Request");
+ BulkRequestBuilder bulkRequest = client.prepareBulk();
+
+ List<?> body = exchange.getIn().getBody(List.class);;
+
+ for (Object document : body) {
+ IndexRequestBuilder prepareIndex = client.prepareIndex(indexName, indexType);
+ log.trace("Indexing document : {}", document);
+ if (!setIndexRequestSource(document, prepareIndex)) {
+ throw new ExpectedBodyTypeException(exchange, XContentBuilder.class);
+ }
+ bulkRequest.add(prepareIndex);
+ }
+
+ ListenableActionFuture<BulkResponse> future = bulkRequest.execute();
+ BulkResponse bulkResponse = future.actionGet();
+
+ List<String> indexedIds = new LinkedList<String>();
+ for (BulkItemResponse response : bulkResponse.getItems()) {
+ indexedIds.add(response.getId());
+ }
+ log.debug("List of successfully indexed document ids : {}", indexedIds);
+ exchange.getIn().setBody(indexedIds);
+ }
+
+
+ private Object extractDocumentFromMessage(Message msg) {
Object body = null;
- boolean converted = false;
// order is important
Class<?>[] types = new Class[] {XContentBuilder.class, Map.class, byte[].class, String.class};
@@ -135,16 +179,25 @@ public class ElasticsearchProducer extends DefaultProducer {
body = msg.getBody(type);
}
- if (body != null) {
+ return body;
+
+ }
+
+
+ @SuppressWarnings("unchecked")
+ private boolean setIndexRequestSource(Object document, IndexRequestBuilder builder) {
+ boolean converted = false;
+
+ if (document != null) {
converted = true;
- if (body instanceof byte[]) {
- builder.setSource((byte[])body);
- } else if (body instanceof Map) {
- builder.setSource((Map<String, Object>) body);
- } else if (body instanceof String) {
- builder.setSource((String)body);
- } else if (body instanceof XContentBuilder) {
- builder.setSource((XContentBuilder)body);
+ if (document instanceof byte[]) {
+ builder.setSource((byte[])document);
+ } else if (document instanceof Map) {
+ builder.setSource((Map<String, Object>) document);
+ } else if (document instanceof String) {
+ builder.setSource((String)document);
+ } else if (document instanceof XContentBuilder) {
+ builder.setSource((XContentBuilder)document);
} else {
converted = false;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/cea796a7/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 cab6451..d1e9945 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
@@ -16,8 +16,6 @@
*/
package org.apache.camel.component.elasticsearch;
-import java.util.HashMap;
-import java.util.Map;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.elasticsearch.action.delete.DeleteResponse;
@@ -26,6 +24,11 @@ import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
public class ElasticsearchComponentTest extends CamelTestSupport {
@Override
@@ -50,6 +53,22 @@ public class ElasticsearchComponentTest extends CamelTestSupport {
}
@Test
+ public void testBulkIndex() throws Exception {
+ List<Map<String, String>> documents = new ArrayList<Map<String, String>>();
+ Map<String, String> document1 = new HashMap<String, String>();
+ document1.put("content1", "test1");
+ Map<String, String> document2 = new HashMap<String, String>();
+ document2.put("content2", "test2");
+
+ documents.add(document1);
+ documents.add(document2);
+
+ List indexIds = template.requestBody("direct:bulk_index", documents, List.class);
+ assertNotNull("indexIds should be set", indexIds);
+ assertCollectionSize("Indexed documents should match the size of documents", indexIds, documents.size());
+ }
+
+ @Test
public void testGet() throws Exception {
//first, INDEX a value
Map<String, String> map = new HashMap<String, String>();
@@ -189,6 +208,7 @@ public class ElasticsearchComponentTest extends CamelTestSupport {
from("direct:index").to("elasticsearch://local?operation=INDEX&indexName=twitter&indexType=tweet");
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: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");
}
[2/2] git commit: CAMEL-7681 Fixed the CS errors
Posted by ni...@apache.org.
CAMEL-7681 Fixed the CS errors
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/fb775276
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/fb775276
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/fb775276
Branch: refs/heads/master
Commit: fb77527608ce0c3f862865183e8ec29c2f8075a4
Parents: cea796a
Author: Willem Jiang <wi...@gmail.com>
Authored: Wed Aug 13 16:17:31 2014 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Wed Aug 13 16:17:31 2014 +0800
----------------------------------------------------------------------
.../elasticsearch/ElasticsearchConfiguration.java | 9 +++++----
.../component/elasticsearch/ElasticsearchProducer.java | 10 +++++-----
.../elasticsearch/ElasticsearchComponentTest.java | 10 +++++-----
3 files changed, 15 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/fb775276/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 1703d50..0c483bd 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
@@ -16,17 +16,18 @@
*/
package org.apache.camel.component.elasticsearch;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Map;
+
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriParams;
import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeBuilder;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.Map;
-
import static org.elasticsearch.node.NodeBuilder.nodeBuilder;
+
@UriParams
public class ElasticsearchConfiguration {
http://git-wip-us.apache.org/repos/asf/camel/blob/fb775276/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 2d190b0..60b9c47 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,6 +16,10 @@
*/
package org.apache.camel.component.elasticsearch;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
import org.apache.camel.Exchange;
import org.apache.camel.ExpectedBodyTypeException;
import org.apache.camel.Message;
@@ -31,10 +35,6 @@ import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.xcontent.XContentBuilder;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-
/**
* Represents an Elasticsearch producer.
*/
@@ -145,7 +145,7 @@ public class ElasticsearchProducer extends DefaultProducer {
log.debug("Preparing Bulk Request");
BulkRequestBuilder bulkRequest = client.prepareBulk();
- List<?> body = exchange.getIn().getBody(List.class);;
+ List<?> body = exchange.getIn().getBody(List.class);
for (Object document : body) {
IndexRequestBuilder prepareIndex = client.prepareIndex(indexName, indexType);
http://git-wip-us.apache.org/repos/asf/camel/blob/fb775276/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 d1e9945..7e96c93 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
@@ -16,6 +16,11 @@
*/
package org.apache.camel.component.elasticsearch;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.elasticsearch.action.delete.DeleteResponse;
@@ -24,11 +29,6 @@ import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
public class ElasticsearchComponentTest extends CamelTestSupport {
@Override