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/11/25 09:47:55 UTC
[5/6] camel git commit: Allow to use existing elasticsearch client.
Speedup unit test to run much much faster as embedded reused ES node.
Allow to use existing elasticsearch client. Speedup unit test to run much much faster as embedded reused ES node.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/2ccd02f1
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/2ccd02f1
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/2ccd02f1
Branch: refs/heads/master
Commit: 2ccd02f1121e2cbb27fcf1408594bd96610c9086
Parents: 7f0190a
Author: Claus Ibsen <da...@apache.org>
Authored: Wed Nov 25 09:46:21 2015 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed Nov 25 09:46:21 2015 +0100
----------------------------------------------------------------------
components/camel-elasticsearch/pom.xml | 114 +++--
.../elasticsearch/ElasticsearchComponent.java | 17 +-
.../elasticsearch/ElasticsearchEndpoint.java | 79 ++--
.../elasticsearch/ElasticsearchBaseTest.java | 108 +++++
.../elasticsearch/ElasticsearchBulkTest.java | 95 ++++
.../ElasticsearchComponentTest.java | 468 -------------------
.../ElasticsearchGetSearchDeleteUpdateTest.java | 224 +++++++++
.../elasticsearch/ElasticsearchIndexTest.java | 145 ++++++
.../elasticsearch/SpringElasticsearchTest.java | 80 ----
.../src/test/resources/log4j.properties | 2 +-
.../SpringElasticsearchTest-context.xml | 32 --
11 files changed, 688 insertions(+), 676 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/2ccd02f1/components/camel-elasticsearch/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-elasticsearch/pom.xml b/components/camel-elasticsearch/pom.xml
index f426aaf..ef92eb9 100644
--- a/components/camel-elasticsearch/pom.xml
+++ b/components/camel-elasticsearch/pom.xml
@@ -15,72 +15,64 @@
See the License for the specific language governing permissions and
limitations under the License.
-->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
- <modelVersion>4.0.0</modelVersion>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.apache.camel</groupId>
- <artifactId>components</artifactId>
- <version>2.17-SNAPSHOT</version>
- </parent>
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components</artifactId>
+ <version>2.17-SNAPSHOT</version>
+ </parent>
- <artifactId>camel-elasticsearch</artifactId>
- <packaging>bundle</packaging>
- <name>Camel :: ElasticSearch</name>
- <description>Camel Elasticsearch support</description>
+ <artifactId>camel-elasticsearch</artifactId>
+ <packaging>bundle</packaging>
+ <name>Camel :: ElasticSearch</name>
+ <description>Camel Elasticsearch support</description>
- <properties>
- <camel.osgi.export.pkg>org.apache.camel.component.elasticsearch.*;${camel.osgi.version}</camel.osgi.export.pkg>
- <camel.osgi.export.service>org.apache.camel.spi.ComponentResolver;component=elasticsearch</camel.osgi.export.service>
- </properties>
+ <properties>
+ <camel.osgi.export.pkg>org.apache.camel.component.elasticsearch.*;${camel.osgi.version}</camel.osgi.export.pkg>
+ <camel.osgi.export.service>org.apache.camel.spi.ComponentResolver;component=elasticsearch</camel.osgi.export.service>
+ </properties>
- <dependencies>
- <dependency>
- <groupId>org.apache.camel</groupId>
- <artifactId>camel-core</artifactId>
- </dependency>
- <dependency>
- <groupId>org.elasticsearch</groupId>
- <artifactId>elasticsearch</artifactId>
- <version>${elasticsearch-version}</version>
- </dependency>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.elasticsearch</groupId>
+ <artifactId>elasticsearch</artifactId>
+ <version>${elasticsearch-version}</version>
+ </dependency>
- <!-- for testing -->
- <dependency>
- <groupId>org.apache.camel</groupId>
- <artifactId>camel-test</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.camel</groupId>
- <artifactId>camel-test-spring</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.springframework</groupId>
- <artifactId>spring-test</artifactId>
- <scope>test</scope>
- </dependency>
+ <!-- for testing -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test</artifactId>
+ <scope>test</scope>
+ </dependency>
- <!-- logging -->
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- <scope>test</scope>
- </dependency>
- </dependencies>
+ <!-- logging -->
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <systemPropertyVariables>
+ <es.path.data>target/data</es.path.data>
+ </systemPropertyVariables>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-surefire-plugin</artifactId>
- <configuration>
- <systemPropertyVariables>
- <es.path.data>target/data</es.path.data>
- </systemPropertyVariables>
- </configuration>
- </plugin>
- </plugins>
- </build>
</project>
http://git-wip-us.apache.org/repos/asf/camel/blob/2ccd02f1/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchComponent.java b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchComponent.java
index cb3b197..f12b414 100644
--- a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchComponent.java
+++ b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchComponent.java
@@ -24,6 +24,7 @@ import java.util.Map;
import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.impl.UriEndpointComponent;
+import org.elasticsearch.client.Client;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
/**
@@ -31,6 +32,8 @@ import org.elasticsearch.common.transport.InetSocketTransportAddress;
*/
public class ElasticsearchComponent extends UriEndpointComponent {
+ private Client client;
+
public ElasticsearchComponent() {
super(ElasticsearchEndpoint.class);
}
@@ -60,8 +63,7 @@ public class ElasticsearchComponent extends UriEndpointComponent {
config.setTransportAddressesList(parseTransportAddresses(config.getTransportAddresses(), config));
- Endpoint endpoint = new ElasticsearchEndpoint(uri, this, config);
-
+ Endpoint endpoint = new ElasticsearchEndpoint(uri, this, config, client);
return endpoint;
}
@@ -84,4 +86,15 @@ public class ElasticsearchComponent extends UriEndpointComponent {
}
return addressesTrAd;
}
+
+ public Client getClient() {
+ return client;
+ }
+
+ /**
+ * To use an existing configured Elasticsearch client, instead of creating a client per endpoint.
+ */
+ public void setClient(Client client) {
+ this.client = client;
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/2ccd02f1/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchEndpoint.java b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchEndpoint.java
index 07ff1e0..e6f0099 100644
--- a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchEndpoint.java
+++ b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchEndpoint.java
@@ -37,6 +37,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.elasticsearch.node.NodeBuilder.nodeBuilder;
+
/**
* Represents an Elasticsearch endpoint.
*/
@@ -47,12 +48,15 @@ public class ElasticsearchEndpoint extends DefaultEndpoint {
private Node node;
private Client client;
+ private volatile boolean closeClient;
@UriParam
private ElasticsearchConfiguration configuration;
- public ElasticsearchEndpoint(String uri, ElasticsearchComponent component, ElasticsearchConfiguration config) throws Exception {
+ public ElasticsearchEndpoint(String uri, ElasticsearchComponent component, ElasticsearchConfiguration config, Client client) throws Exception {
super(uri, component);
this.configuration = config;
+ this.client = client;
+ this.closeClient = client == null;
}
public Producer createProducer() throws Exception {
@@ -71,31 +75,37 @@ public class ElasticsearchEndpoint extends DefaultEndpoint {
@SuppressWarnings("unchecked")
protected void doStart() throws Exception {
super.doStart();
- if (configuration.isLocal()) {
- LOG.info("Starting local ElasticSearch server");
- } else {
- LOG.info("Joining ElasticSearch cluster " + configuration.getClusterName());
- }
- if (configuration.getIp() != null) {
- this.client = new TransportClient(getSettings())
- .addTransportAddress(new InetSocketTransportAddress(configuration.getIp(), configuration.getPort()));
-
- } else if (configuration.getTransportAddressesList() != null
- && !configuration.getTransportAddressesList().isEmpty()) {
- List<TransportAddress> addresses = new ArrayList(configuration.getTransportAddressesList().size());
- for (TransportAddress address : configuration.getTransportAddressesList()) {
- addresses.add(address);
+
+ if (client == null) {
+ if (configuration.isLocal()) {
+ LOG.info("Starting local ElasticSearch server");
+ } else {
+ LOG.info("Joining ElasticSearch cluster " + configuration.getClusterName());
}
- this.client = new TransportClient(getSettings())
- .addTransportAddresses(addresses.toArray(new TransportAddress[addresses.size()]));
- } else {
- NodeBuilder builder = nodeBuilder().local(configuration.isLocal()).data(configuration.getData());
- if (!configuration.isLocal() && configuration.getClusterName() != null) {
- builder.clusterName(configuration.getClusterName());
+ if (configuration.getIp() != null) {
+ this.client = new TransportClient(getSettings())
+ .addTransportAddress(new InetSocketTransportAddress(configuration.getIp(), configuration.getPort()));
+
+ } else if (configuration.getTransportAddressesList() != null
+ && !configuration.getTransportAddressesList().isEmpty()) {
+ List<TransportAddress> addresses = new ArrayList(configuration.getTransportAddressesList().size());
+ for (TransportAddress address : configuration.getTransportAddressesList()) {
+ addresses.add(address);
+ }
+ this.client = new TransportClient(getSettings())
+ .addTransportAddresses(addresses.toArray(new TransportAddress[addresses.size()]));
+ } else {
+ NodeBuilder builder = nodeBuilder().local(configuration.isLocal()).data(configuration.getData());
+ builder.getSettings().classLoader(Settings.class.getClassLoader());
+ if (configuration.isLocal()) {
+ builder.getSettings().put("http.enabled", false);
+ }
+ if (!configuration.isLocal() && configuration.getClusterName() != null) {
+ builder.clusterName(configuration.getClusterName());
+ }
+ node = builder.node();
+ client = node.client();
}
- builder.getSettings().classLoader(Settings.class.getClassLoader());
- node = builder.node();
- client = node.client();
}
}
@@ -110,19 +120,24 @@ public class ElasticsearchEndpoint extends DefaultEndpoint {
.put("client.transport.ignore_cluster_name", false)
.put("node.client", true)
.put("client.transport.sniff", true)
+ .put("http.enabled", false)
.build();
}
@Override
protected void doStop() throws Exception {
- if (configuration.isLocal()) {
- LOG.info("Stopping local ElasticSearch server");
- } else {
- LOG.info("Leaving ElasticSearch cluster " + configuration.getClusterName());
- }
- client.close();
- if (node != null) {
- node.close();
+ if (closeClient) {
+ if (configuration.isLocal()) {
+ LOG.info("Stopping local ElasticSearch server");
+ } else {
+ LOG.info("Leaving ElasticSearch cluster " + configuration.getClusterName());
+ }
+ client.close();
+ if (node != null) {
+ node.close();
+ }
+ client = null;
+ node = null;
}
super.doStop();
}
http://git-wip-us.apache.org/repos/asf/camel/blob/2ccd02f1/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchBaseTest.java
----------------------------------------------------------------------
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchBaseTest.java b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchBaseTest.java
new file mode 100644
index 0000000..0d6208c
--- /dev/null
+++ b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchBaseTest.java
@@ -0,0 +1,108 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.elasticsearch;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.node.Node;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import static org.elasticsearch.node.NodeBuilder.nodeBuilder;
+
+public class ElasticsearchBaseTest extends CamelTestSupport {
+
+ private static Node node;
+ private static Client client;
+
+ @BeforeClass
+ public static void cleanupOnce() {
+ deleteDirectory("target/data");
+
+ // create an embedded node to resue
+ node = nodeBuilder().local(true)
+ .settings(ImmutableSettings.settingsBuilder().put("http.enabled", false).put("path.data", "target/data"))
+ .node();
+
+ client = node.client();
+ }
+
+ @AfterClass
+ public static void teardownOnce() {
+ if (client != null) {
+ client.close();
+ }
+ if (node != null) {
+ node.close();
+ }
+ }
+
+ @Override
+ public boolean isCreateCamelContextPerClass() {
+ // let's speed up the tests using the same context
+ return true;
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+
+ // reuse existing client
+ ElasticsearchComponent es = context.getComponent("elasticsearch", ElasticsearchComponent.class);
+ es.setClient(client);
+
+ return context;
+ }
+
+ /**
+ * As we don't delete the {@code target/data} folder for <b>each</b> test
+ * below (otherwise they would run much slower), we need to make sure
+ * there's no side effect of the same used data through creating unique
+ * indexes.
+ */
+ Map<String, String> createIndexedData(String... additionalPrefixes) {
+ String prefix = createPrefix();
+
+ // take over any potential prefixes we may have been asked for
+ if (additionalPrefixes.length > 0) {
+ StringBuilder sb = new StringBuilder(prefix);
+ for (String additionalPrefix : additionalPrefixes) {
+ sb.append(additionalPrefix).append("-");
+ }
+ prefix = sb.toString();
+ }
+
+ String key = prefix + "key";
+ String value = prefix + "value";
+ log.info("Creating indexed data using the key/value pair {} => {}", key, value);
+
+ Map<String, String> map = new HashMap<String, String>();
+ map.put(key, value);
+ return map;
+ }
+
+ String createPrefix() {
+ // make use of the test method name to avoid collision
+ return getTestMethodName().toLowerCase() + "-";
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/2ccd02f1/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchBulkTest.java
----------------------------------------------------------------------
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchBulkTest.java b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchBulkTest.java
new file mode 100644
index 0000000..5aee873
--- /dev/null
+++ b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchBulkTest.java
@@ -0,0 +1,95 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.elasticsearch;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.index.IndexRequest;
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.notNullValue;
+
+public class ElasticsearchBulkTest extends ElasticsearchBaseTest {
+
+ @Test
+ public void testBulkIndex() throws Exception {
+ List<Map<String, String>> documents = new ArrayList<Map<String, String>>();
+ Map<String, String> document1 = createIndexedData("1");
+ Map<String, String> document2 = createIndexedData("2");
+
+ 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 bulkIndexRequestBody() throws Exception {
+ String prefix = createPrefix();
+
+ // given
+ BulkRequest request = new BulkRequest();
+ request.add(new IndexRequest(prefix + "foo", prefix + "bar", prefix + "baz")
+ .source("{\"" + prefix + "content\": \"" + prefix + "hello\"}"));
+
+ // when
+ @SuppressWarnings("unchecked")
+ List<String> indexedDocumentIds = template.requestBody("direct:bulk_index", request, List.class);
+
+ // then
+ assertThat(indexedDocumentIds, notNullValue());
+ assertThat(indexedDocumentIds.size(), equalTo(1));
+ assertThat(indexedDocumentIds, hasItem(prefix + "baz"));
+ }
+
+ @Test
+ public void bulkRequestBody() throws Exception {
+ String prefix = createPrefix();
+
+ // given
+ BulkRequest request = new BulkRequest();
+ request.add(new IndexRequest(prefix + "foo", prefix + "bar", prefix + "baz")
+ .source("{\"" + prefix + "content\": \"" + prefix + "hello\"}"));
+
+ // when
+ BulkResponse response = template.requestBody("direct:bulk", request, BulkResponse.class);
+
+ // then
+ assertThat(response, notNullValue());
+ assertEquals(prefix + "baz", response.getItems()[0].getId());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ 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");
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/2ccd02f1/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
deleted file mode 100644
index 66459a0..0000000
--- a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchComponentTest.java
+++ /dev/null
@@ -1,468 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-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.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;
-import org.elasticsearch.action.get.GetResponse;
-import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.action.update.UpdateRequest;
-import org.junit.BeforeClass;
-import org.junit.Ignore;
-import org.junit.Test;
-
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.CoreMatchers.notNullValue;
-
-
-public class ElasticsearchComponentTest extends CamelTestSupport {
-
- @BeforeClass
- public static void cleanupOnce() {
- deleteDirectory("target/data");
- }
-
- @Override
- public boolean isCreateCamelContextPerClass() {
- // let's speed up the tests using the same context
- return true;
- }
-
- /**
- * As we don't delete the {@code target/data} folder for <b>each</b> test
- * below (otherwise they would run much slower), we need to make sure
- * there's no side effect of the same used data through creating unique
- * indexes.
- */
- private Map<String, String> createIndexedData(String... additionalPrefixes) {
- String prefix = createPrefix();
-
- // take over any potential prefixes we may have been asked for
- if (additionalPrefixes.length > 0) {
- StringBuilder sb = new StringBuilder(prefix);
- for (String additionalPrefix : additionalPrefixes) {
- sb.append(additionalPrefix).append("-");
- }
- prefix = sb.toString();
- }
-
- String key = prefix + "key";
- String value = prefix + "value";
- log.info("Creating indexed data using the key/value pair {} => {}", key, value);
-
- Map<String, String> map = new HashMap<String, String>();
- map.put(key, value);
- return map;
- }
-
- private String createPrefix() {
- // make use of the test method name to avoid collision
- return getTestMethodName().toLowerCase() + "-";
- }
-
- @Test
- public void testIndex() throws Exception {
- Map<String, String> map = createIndexedData();
- String indexId = template.requestBody("direct:index", map, String.class);
- assertNotNull("indexId should be set", indexId);
- }
-
- @Test
- public void testUpdate() throws Exception {
- Map<String, String> map = createIndexedData();
- String indexId = template.requestBody("direct:index", map, String.class);
- assertNotNull("indexId should be set", indexId);
-
- Map<String, String> newMap = new HashMap<>();
- newMap.put(createPrefix() + "key2", createPrefix() + "value2");
- Map<String, Object> headers = new HashMap<>();
- headers.put(ElasticsearchConstants.PARAM_INDEX_ID, indexId);
- indexId = template.requestBodyAndHeaders("direct:update", newMap, headers, String.class);
- assertNotNull("indexId should be set", indexId);
- }
-
- @Test
- public void testIndexWithReplication() throws Exception {
- Map<String, String> map = createIndexedData();
- String indexId = template.requestBody("direct:indexWithReplication", map, String.class);
- assertNotNull("indexId should be set", indexId);
- }
-
- @Test
- public void testIndexWithWriteConsistency() throws Exception {
- Map<String, String> map = createIndexedData();
- String indexId = template.requestBody("direct:indexWithWriteConsistency", map, String.class);
- assertNotNull("indexId should be set", indexId);
- }
-
- @Test
- public void testBulkIndex() throws Exception {
- List<Map<String, String>> documents = new ArrayList<Map<String, String>>();
- Map<String, String> document1 = createIndexedData("1");
- Map<String, String> document2 = createIndexedData("2");
-
- 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 = createIndexedData();
- sendBody("direct:index", map);
- String indexId = template.requestBody("direct:index", map, String.class);
- assertNotNull("indexId should be set", indexId);
-
- //now, verify GET succeeded
- GetResponse response = template.requestBody("direct:get", indexId, GetResponse.class);
- assertNotNull("response should not be null", response);
- assertNotNull("response source should not be null", response.getSource());
- }
-
- @Test
- public void testDelete() throws Exception {
- //first, INDEX a value
- Map<String, String> map = createIndexedData();
- sendBody("direct:index", map);
- String indexId = template.requestBody("direct:index", map, String.class);
- assertNotNull("indexId should be set", indexId);
-
- //now, verify GET succeeded
- GetResponse response = template.requestBody("direct:get", indexId, GetResponse.class);
- assertNotNull("response should not be null", response);
- assertNotNull("response source should not be null", response.getSource());
-
- //now, perform DELETE
- DeleteResponse deleteResponse = template.requestBody("direct:delete", indexId, DeleteResponse.class);
- assertNotNull("response should not be null", deleteResponse);
-
- //now, verify GET fails to find the indexed value
- response = template.requestBody("direct:get", indexId, GetResponse.class);
- assertNotNull("response should not be null", response);
- assertNull("response source should be null", response.getSource());
- }
-
- @Test
- public void testSearch() throws Exception {
- //first, INDEX a value
- Map<String, String> map = createIndexedData();
- sendBody("direct:index", map);
-
- //now, verify GET succeeded
- Map<String, Object> actualQuery = new HashMap<String, Object>();
- actualQuery.put("content", "searchtest");
- Map<String, Object> match = new HashMap<String, Object>();
- match.put("match", actualQuery);
- Map<String, Object> query = new HashMap<String, Object>();
- query.put("query", match);
- SearchResponse response = template.requestBody("direct:search", query, SearchResponse.class);
- assertNotNull("response should not be null", response);
- assertNotNull("response hits should be == 1", response.getHits().totalHits());
- }
-
- @Test
- public void testIndexWithHeaders() throws Exception {
- Map<String, String> map = createIndexedData();
- Map<String, Object> headers = new HashMap<String, Object>();
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
- headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
- headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
-
- String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
- assertNotNull("indexId should be set", indexId);
- }
-
- @Test
- public void testIndexWithIDInHeader() throws Exception {
- Map<String, String> map = createIndexedData();
- Map<String, Object> headers = new HashMap<String, Object>();
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
- headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
- headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
- headers.put(ElasticsearchConstants.PARAM_INDEX_ID, "123");
-
- String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
- assertNotNull("indexId should be set", indexId);
- assertEquals("indexId should be equals to the provided id", "123", indexId);
- }
-
- @Test
- public void testUpdateWithIDInHeader() throws Exception {
- Map<String, String> map = createIndexedData();
- Map<String, Object> headers = new HashMap<String, Object>();
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
- headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
- headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
- headers.put(ElasticsearchConstants.PARAM_INDEX_ID, "123");
-
- String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
- assertNotNull("indexId should be set", indexId);
- assertEquals("indexId should be equals to the provided id", "123", indexId);
-
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_UPDATE);
-
- indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
- assertNotNull("indexId should be set", indexId);
- assertEquals("indexId should be equals to the provided id", "123", indexId);
- }
-
- @Test
- @Ignore("need to setup the cluster IP for this test")
- public void indexWithIp() throws Exception {
- Map<String, String> map = createIndexedData();
- Map<String, Object> headers = new HashMap<String, Object>();
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
- headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
- headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
-
- String indexId = template.requestBodyAndHeaders("direct:indexWithIp", map, headers, String.class);
- assertNotNull("indexId should be set", indexId);
- }
-
- @Test
- @Ignore("need to setup the cluster IP/Port for this test")
- public void indexWithIpAndPort() throws Exception {
- Map<String, String> map = createIndexedData();
- Map<String, Object> headers = new HashMap<String, Object>();
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
- headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
- headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
-
- String indexId = template.requestBodyAndHeaders("direct:indexWithIpAndPort", map, headers, String.class);
- assertNotNull("indexId should be set", indexId);
- }
-
- @Test
- @Ignore("need to setup the cluster with multiple nodes for this test")
- public void indexWithTransportAddresses() throws Exception {
- Map<String, String> map = createIndexedData();
- Map<String, Object> headers = new HashMap<String, Object>();
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
- headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
- headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
-
- String indexId = template.requestBodyAndHeaders("direct:indexWithTransportAddresses", map, headers, String.class);
- assertNotNull("indexId should be set", indexId);
- }
-
- @Test
- @Ignore("need to setup the cluster with multiple nodes for this test")
- public void indexWithIpAndTransportAddresses() throws Exception {
- Map<String, String> map = createIndexedData();
- Map<String, Object> headers = new HashMap<String, Object>();
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
- headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
- headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
-
- //should ignore transport addresses configuration
- String indexId = template.requestBodyAndHeaders("direct:indexWithIpAndTransportAddresses", map, headers, String.class);
- assertNotNull("indexId should be set", indexId);
- }
-
- @Test
- public void testGetWithHeaders() throws Exception {
- //first, INDEX a value
- Map<String, String> map = createIndexedData();
- Map<String, Object> headers = new HashMap<String, Object>();
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
- headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
- headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
-
- String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
-
- //now, verify GET
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_GET_BY_ID);
- GetResponse response = template.requestBodyAndHeaders("direct:start", indexId, headers, GetResponse.class);
- assertNotNull("response should not be null", response);
- assertNotNull("response source should not be null", response.getSource());
- }
-
- @Test
- public void testDeleteWithHeaders() throws Exception {
- //first, INDEX a value
- Map<String, String> map = createIndexedData();
- Map<String, Object> headers = new HashMap<String, Object>();
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
- headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
- headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
-
- String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
-
- //now, verify GET
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_GET_BY_ID);
- GetResponse response = template.requestBodyAndHeaders("direct:start", indexId, headers, GetResponse.class);
- assertNotNull("response should not be null", response);
- assertNotNull("response source should not be null", response.getSource());
-
- //now, perform DELETE
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_DELETE);
- DeleteResponse deleteResponse = template.requestBodyAndHeaders("direct:start", indexId, headers, DeleteResponse.class);
- assertNotNull("response should not be null", deleteResponse);
-
- //now, verify GET fails to find the indexed value
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_GET_BY_ID);
- response = template.requestBodyAndHeaders("direct:start", indexId, headers, GetResponse.class);
- assertNotNull("response should not be null", response);
- assertNull("response source should be null", response.getSource());
- }
-
- @Test
- public void indexRequestBody() throws Exception {
- String prefix = createPrefix();
-
- // given
- IndexRequest request = new IndexRequest(prefix + "foo", prefix + "bar", prefix + "testId");
- request.source("{\"" + prefix + "content\": \"" + prefix + "hello\"}");
-
- // when
- String documentId = template.requestBody("direct:index", request, String.class);
-
- // then
- assertThat(documentId, equalTo(prefix + "testId"));
- }
-
- @Test
- public void updateRequestBody() throws Exception {
- String prefix = createPrefix();
-
- // first index data
- IndexRequest indexRequest = new IndexRequest(prefix + "foo", prefix + "bar", prefix + "testId");
- indexRequest.source("{\"" + prefix + "content\": \"" + prefix + "hello\"}");
- template.requestBody("direct:index", indexRequest, String.class);
-
- // then update
- UpdateRequest request = new UpdateRequest(prefix + "foo", prefix + "bar", prefix + "testId");
- request.doc("{\"" + prefix + "content2\": \"" + prefix + "hello2\"}");
- String documentId = template.requestBody("direct:update", request, String.class);
-
- assertThat(documentId, equalTo(prefix + "testId"));
- }
-
- @Test
- public void getRequestBody() throws Exception {
- String prefix = createPrefix();
-
- // given
- GetRequest request = new GetRequest(prefix + "foo").type(prefix + "bar");
-
- // when
- String documentId = template.requestBody("direct:index",
- new IndexRequest(prefix + "foo", prefix + "bar", prefix + "testId")
- .source("{\"" + prefix + "content\": \"" + prefix + "hello\"}"), String.class);
- GetResponse response = template.requestBody("direct:get",
- request.id(documentId), GetResponse.class);
-
- // then
- assertThat(response, notNullValue());
- assertThat(prefix + "hello", equalTo(response.getSourceAsMap().get(prefix + "content")));
- }
-
- @Test
- public void deleteRequestBody() throws Exception {
- String prefix = createPrefix();
-
- // given
- DeleteRequest request = new DeleteRequest(prefix + "foo").type(prefix + "bar");
-
- // when
- String documentId = template.requestBody("direct:index",
- new IndexRequest("" + prefix + "foo", "" + prefix + "bar", "" + prefix + "testId")
- .source("{\"" + prefix + "content\": \"" + prefix + "hello\"}"), String.class);
- DeleteResponse response = template.requestBody("direct:delete",
- request.id(documentId), DeleteResponse.class);
-
- // then
- assertThat(response, notNullValue());
- assertThat(documentId, equalTo(response.getId()));
- }
-
- @Test
- public void bulkIndexRequestBody() throws Exception {
- String prefix = createPrefix();
-
- // given
- BulkRequest request = new BulkRequest();
- request.add(new IndexRequest(prefix + "foo", prefix + "bar", prefix + "baz")
- .source("{\"" + prefix + "content\": \"" + prefix + "hello\"}"));
-
- // when
- @SuppressWarnings("unchecked")
- List<String> indexedDocumentIds = template.requestBody("direct:bulk_index", request, List.class);
-
- // then
- assertThat(indexedDocumentIds, notNullValue());
- assertThat(indexedDocumentIds.size(), equalTo(1));
- assertThat(indexedDocumentIds, hasItem(prefix + "baz"));
- }
-
- @Test
- public void bulkRequestBody() throws Exception {
- String prefix = createPrefix();
-
- // given
- BulkRequest request = new BulkRequest();
- request.add(new IndexRequest(prefix + "foo", prefix + "bar", prefix + "baz")
- .source("{\"" + prefix + "content\": \"" + prefix + "hello\"}"));
-
- // when
- BulkResponse response = template.requestBody("direct:bulk", request, BulkResponse.class);
-
- // then
- assertThat(response, notNullValue());
- assertEquals(prefix + "baz", response.getItems()[0].getId());
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() {
- from("direct:start").to("elasticsearch://local");
- from("direct:index").to("elasticsearch://local?operation=INDEX&indexName=twitter&indexType=tweet");
- from("direct:update").to("elasticsearch://local?operation=UPDATE&indexName=twitter&indexType=tweet");
- from("direct:indexWithReplication").to("elasticsearch://local?operation=INDEX&indexName=twitter&indexType=tweet&replicationType=SYNC");
- from("direct:indexWithWriteConsistency").to("elasticsearch://local?operation=INDEX&indexName=twitter&indexType=tweet&consistencyLevel=ONE");
- 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:search").to("elasticsearch://local?operation=SEARCH&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");
- //from("direct:indexWithTransportAddresses").to("elasticsearch://elasticsearch?operation=INDEX&indexName=twitter&indexType=tweet&transportAddresses=localhost:9300,localhost:9301");
- //from("direct:indexWithIpAndTransportAddresses").
- //to("elasticsearch://elasticsearch?operation=INDEX&indexName=twitter&indexType=tweet&ip=localhost&port=9300&transportAddresses=localhost:4444,localhost:5555");
- }
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/2ccd02f1/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchGetSearchDeleteUpdateTest.java
----------------------------------------------------------------------
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchGetSearchDeleteUpdateTest.java b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchGetSearchDeleteUpdateTest.java
new file mode 100644
index 0000000..d02ce86
--- /dev/null
+++ b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchGetSearchDeleteUpdateTest.java
@@ -0,0 +1,224 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.elasticsearch;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.delete.DeleteResponse;
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.notNullValue;
+
+public class ElasticsearchGetSearchDeleteUpdateTest extends ElasticsearchBaseTest {
+
+ @Test
+ public void testGet() throws Exception {
+ //first, INDEX a value
+ Map<String, String> map = createIndexedData();
+ sendBody("direct:index", map);
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull("indexId should be set", indexId);
+
+ //now, verify GET succeeded
+ GetResponse response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertNotNull("response should not be null", response);
+ assertNotNull("response source should not be null", response.getSource());
+ }
+
+ @Test
+ public void testDelete() throws Exception {
+ //first, INDEX a value
+ Map<String, String> map = createIndexedData();
+ sendBody("direct:index", map);
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull("indexId should be set", indexId);
+
+ //now, verify GET succeeded
+ GetResponse response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertNotNull("response should not be null", response);
+ assertNotNull("response source should not be null", response.getSource());
+
+ //now, perform DELETE
+ DeleteResponse deleteResponse = template.requestBody("direct:delete", indexId, DeleteResponse.class);
+ assertNotNull("response should not be null", deleteResponse);
+
+ //now, verify GET fails to find the indexed value
+ response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertNotNull("response should not be null", response);
+ assertNull("response source should be null", response.getSource());
+ }
+
+ @Test
+ public void testSearch() throws Exception {
+ //first, INDEX a value
+ Map<String, String> map = createIndexedData();
+ sendBody("direct:index", map);
+
+ //now, verify GET succeeded
+ Map<String, Object> actualQuery = new HashMap<String, Object>();
+ actualQuery.put("content", "searchtest");
+ Map<String, Object> match = new HashMap<String, Object>();
+ match.put("match", actualQuery);
+ Map<String, Object> query = new HashMap<String, Object>();
+ query.put("query", match);
+ SearchResponse response = template.requestBody("direct:search", query, SearchResponse.class);
+ assertNotNull("response should not be null", response);
+ assertNotNull("response hits should be == 1", response.getHits().totalHits());
+ }
+
+ @Test
+ public void testUpdate() throws Exception {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull("indexId should be set", indexId);
+
+ Map<String, String> newMap = new HashMap<>();
+ newMap.put(createPrefix() + "key2", createPrefix() + "value2");
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_INDEX_ID, indexId);
+ indexId = template.requestBodyAndHeaders("direct:update", newMap, headers, String.class);
+ assertNotNull("indexId should be set", indexId);
+ }
+
+ @Test
+ public void testGetWithHeaders() throws Exception {
+ //first, INDEX a value
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<String, Object>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
+
+ String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
+
+ //now, verify GET
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_GET_BY_ID);
+ GetResponse response = template.requestBodyAndHeaders("direct:start", indexId, headers, GetResponse.class);
+ assertNotNull("response should not be null", response);
+ assertNotNull("response source should not be null", response.getSource());
+ }
+
+ @Test
+ public void testDeleteWithHeaders() throws Exception {
+ //first, INDEX a value
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<String, Object>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
+
+ String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
+
+ //now, verify GET
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_GET_BY_ID);
+ GetResponse response = template.requestBodyAndHeaders("direct:start", indexId, headers, GetResponse.class);
+ assertNotNull("response should not be null", response);
+ assertNotNull("response source should not be null", response.getSource());
+
+ //now, perform DELETE
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_DELETE);
+ DeleteResponse deleteResponse = template.requestBodyAndHeaders("direct:start", indexId, headers, DeleteResponse.class);
+ assertNotNull("response should not be null", deleteResponse);
+
+ //now, verify GET fails to find the indexed value
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_GET_BY_ID);
+ response = template.requestBodyAndHeaders("direct:start", indexId, headers, GetResponse.class);
+ assertNotNull("response should not be null", response);
+ assertNull("response source should be null", response.getSource());
+ }
+
+ @Test
+ public void testUpdateWithIDInHeader() throws Exception {
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<String, Object>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_ID, "123");
+
+ String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
+ assertNotNull("indexId should be set", indexId);
+ assertEquals("indexId should be equals to the provided id", "123", indexId);
+
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_UPDATE);
+
+ indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
+ assertNotNull("indexId should be set", indexId);
+ assertEquals("indexId should be equals to the provided id", "123", indexId);
+ }
+
+ @Test
+ public void getRequestBody() throws Exception {
+ String prefix = createPrefix();
+
+ // given
+ GetRequest request = new GetRequest(prefix + "foo").type(prefix + "bar");
+
+ // when
+ String documentId = template.requestBody("direct:index",
+ new IndexRequest(prefix + "foo", prefix + "bar", prefix + "testId")
+ .source("{\"" + prefix + "content\": \"" + prefix + "hello\"}"), String.class);
+ GetResponse response = template.requestBody("direct:get",
+ request.id(documentId), GetResponse.class);
+
+ // then
+ assertThat(response, notNullValue());
+ assertThat(prefix + "hello", equalTo(response.getSourceAsMap().get(prefix + "content")));
+ }
+
+ @Test
+ public void deleteRequestBody() throws Exception {
+ String prefix = createPrefix();
+
+ // given
+ DeleteRequest request = new DeleteRequest(prefix + "foo").type(prefix + "bar");
+
+ // when
+ String documentId = template.requestBody("direct:index",
+ new IndexRequest("" + prefix + "foo", "" + prefix + "bar", "" + prefix + "testId")
+ .source("{\"" + prefix + "content\": \"" + prefix + "hello\"}"), String.class);
+ DeleteResponse response = template.requestBody("direct:delete",
+ request.id(documentId), DeleteResponse.class);
+
+ // then
+ assertThat(response, notNullValue());
+ assertThat(documentId, equalTo(response.getId()));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").to("elasticsearch://local");
+ 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:search").to("elasticsearch://local?operation=SEARCH&indexName=twitter&indexType=tweet");
+ from("direct:update").to("elasticsearch://local?operation=UPDATE&indexName=twitter&indexType=tweet");
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/2ccd02f1/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchIndexTest.java
----------------------------------------------------------------------
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchIndexTest.java b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchIndexTest.java
new file mode 100644
index 0000000..ae01e5b
--- /dev/null
+++ b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchIndexTest.java
@@ -0,0 +1,145 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.elasticsearch;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.Ignore;
+import org.junit.Test;
+
+public class ElasticsearchIndexTest extends ElasticsearchBaseTest {
+
+ @Test
+ public void testIndex() throws Exception {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull("indexId should be set", indexId);
+ }
+
+ @Test
+ public void testIndexWithReplication() throws Exception {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:indexWithReplication", map, String.class);
+ assertNotNull("indexId should be set", indexId);
+ }
+
+ @Test
+ public void testIndexWithWriteConsistency() throws Exception {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:indexWithWriteConsistency", map, String.class);
+ assertNotNull("indexId should be set", indexId);
+ }
+
+ @Test
+ public void testIndexWithHeaders() throws Exception {
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<String, Object>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
+
+ String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
+ assertNotNull("indexId should be set", indexId);
+ }
+
+ @Test
+ public void testIndexWithIDInHeader() throws Exception {
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<String, Object>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_ID, "123");
+
+ String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
+ assertNotNull("indexId should be set", indexId);
+ assertEquals("indexId should be equals to the provided id", "123", indexId);
+ }
+
+ @Test
+ @Ignore("need to setup the cluster IP for this test")
+ public void indexWithIp() throws Exception {
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<String, Object>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
+
+ String indexId = template.requestBodyAndHeaders("direct:indexWithIp", map, headers, String.class);
+ assertNotNull("indexId should be set", indexId);
+ }
+
+ @Test
+ @Ignore("need to setup the cluster IP/Port for this test")
+ public void indexWithIpAndPort() throws Exception {
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<String, Object>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
+
+ String indexId = template.requestBodyAndHeaders("direct:indexWithIpAndPort", map, headers, String.class);
+ assertNotNull("indexId should be set", indexId);
+ }
+
+ @Test
+ @Ignore("need to setup the cluster with multiple nodes for this test")
+ public void indexWithTransportAddresses() throws Exception {
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<String, Object>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
+
+ String indexId = template.requestBodyAndHeaders("direct:indexWithTransportAddresses", map, headers, String.class);
+ assertNotNull("indexId should be set", indexId);
+ }
+
+ @Test
+ @Ignore("need to setup the cluster with multiple nodes for this test")
+ public void indexWithIpAndTransportAddresses() throws Exception {
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<String, Object>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
+
+ //should ignore transport addresses configuration
+ String indexId = template.requestBodyAndHeaders("direct:indexWithIpAndTransportAddresses", map, headers, String.class);
+ assertNotNull("indexId should be set", indexId);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").to("elasticsearch://local");
+ from("direct:index").to("elasticsearch://local?operation=INDEX&indexName=twitter&indexType=tweet");
+ from("direct:indexWithReplication").to("elasticsearch://local?operation=INDEX&indexName=twitter&indexType=tweet&replicationType=SYNC");
+ from("direct:indexWithWriteConsistency").to("elasticsearch://local?operation=INDEX&indexName=twitter&indexType=tweet&consistencyLevel=ONE");
+ //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");
+ //from("direct:indexWithTransportAddresses").to("elasticsearch://elasticsearch?operation=INDEX&indexName=twitter&indexType=tweet&transportAddresses=localhost:9300,localhost:9301");
+ //from("direct:indexWithIpAndTransportAddresses").
+ //to("elasticsearch://elasticsearch?operation=INDEX&indexName=twitter&indexType=tweet&ip=localhost&port=9300&transportAddresses=localhost:4444,localhost:5555");
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/2ccd02f1/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/SpringElasticsearchTest.java
----------------------------------------------------------------------
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/SpringElasticsearchTest.java b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/SpringElasticsearchTest.java
deleted file mode 100644
index 506175e..0000000
--- a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/SpringElasticsearchTest.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.camel.component.elasticsearch;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.camel.EndpointInject;
-import org.apache.camel.Produce;
-import org.apache.camel.ProducerTemplate;
-import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.test.spring.CamelSpringTestSupport;
-
-import org.junit.Test;
-
-import org.springframework.context.support.AbstractXmlApplicationContext;
-import org.springframework.context.support.ClassPathXmlApplicationContext;
-
-public class SpringElasticsearchTest extends CamelSpringTestSupport {
-
- @Produce(uri = "direct:index")
- protected ProducerTemplate producer;
-
- @EndpointInject(uri = "mock:result")
- protected MockEndpoint mock;
-
- @Override
- protected AbstractXmlApplicationContext createApplicationContext() {
- deleteDirectory("target/data");
- return new ClassPathXmlApplicationContext("org/apache/camel/component/elasticsearch/SpringElasticsearchTest-context.xml");
- }
-
- @Override
- public boolean isCreateCamelContextPerClass() {
- // let's speed up the tests using the same context
- return true;
- }
-
- @Test
- public void testSendBody() throws Exception {
- mock.expectedMinimumMessageCount(1);
-
- Map<String, String> body = new HashMap<String, String>();
- body.put("content", "test");
- producer.sendBody(body);
-
- mock.assertIsSatisfied();
- }
-
- @Test
- public void testSendBodyAndHeaders() throws Exception {
- mock.expectedMinimumMessageCount(1);
-
- Map<String, String> body = new HashMap<String, String>();
- body.put("content", "test");
-
- Map<String, Object> headers = new HashMap<String, Object>();
- headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
- headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
- headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");
-
- producer.sendBodyAndHeaders(body, headers);
-
- mock.assertIsSatisfied();
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/2ccd02f1/components/camel-elasticsearch/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/components/camel-elasticsearch/src/test/resources/log4j.properties b/components/camel-elasticsearch/src/test/resources/log4j.properties
index 019ec0e..993e98d 100644
--- a/components/camel-elasticsearch/src/test/resources/log4j.properties
+++ b/components/camel-elasticsearch/src/test/resources/log4j.properties
@@ -19,11 +19,11 @@
# The logging properties used during tests
#
log4j.rootLogger=INFO, file
-log4j.logger.org.springframework=WARN
# uncomment this to turn on debug of camel
#log4j.logger.org.apache.camel=DEBUG
#log4j.logger.org.apache.camel.component.elasticsearch=TRACE
+#log4j.logger.org.elasticsearch=DEBUG
# CONSOLE appender not used by default
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
http://git-wip-us.apache.org/repos/asf/camel/blob/2ccd02f1/components/camel-elasticsearch/src/test/resources/org/apache/camel/component/elasticsearch/SpringElasticsearchTest-context.xml
----------------------------------------------------------------------
diff --git a/components/camel-elasticsearch/src/test/resources/org/apache/camel/component/elasticsearch/SpringElasticsearchTest-context.xml b/components/camel-elasticsearch/src/test/resources/org/apache/camel/component/elasticsearch/SpringElasticsearchTest-context.xml
deleted file mode 100644
index fc9cd7f..0000000
--- a/components/camel-elasticsearch/src/test/resources/org/apache/camel/component/elasticsearch/SpringElasticsearchTest-context.xml
+++ /dev/null
@@ -1,32 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- Licensed to the Apache Software Foundation (ASF) under one or more
- contributor license agreements. See the NOTICE file distributed with
- this work for additional information regarding copyright ownership.
- The ASF licenses this file to You under the Apache License, Version 2.0
- (the "License"); you may not use this file except in compliance with
- the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
--->
-<beans xmlns="http://www.springframework.org/schema/beans"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="
- http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
- http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
-
- <camelContext xmlns="http://camel.apache.org/schema/spring">
- <route>
- <from uri="direct:index" />
- <to uri="elasticsearch://local?operation=INDEX&indexName=twitter&indexType=tweet"/>
- <to uri="mock:result"/>
- </route>
- </camelContext>
-
-</beans>