You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by sb...@apache.org on 2016/10/10 21:27:17 UTC
[2/6] incubator-streams git commit: tests improved, all tests passing
tests improved, all tests passing
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/e23901c2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/e23901c2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/e23901c2
Branch: refs/heads/master
Commit: e23901c2ed9b0766f28ae91bc57c89782fef100b
Parents: be3627e
Author: Steve Blackmon @steveblackmon <sb...@apache.org>
Authored: Wed Oct 5 10:37:34 2016 -0500
Committer: Steve Blackmon @steveblackmon <sb...@apache.org>
Committed: Wed Oct 5 10:37:34 2016 -0500
----------------------------------------------------------------------
.../streams-persist-elasticsearch/pom.xml | 25 ++-
.../elasticsearch/test/ElasticsearchITs.java | 18 ++
.../test/ElasticsearchParentChildUpdaterIT.java | 159 ++++++++++++++
.../test/ElasticsearchParentChildWriterIT.java | 195 +++++++++++++++++
.../test/ElasticsearchPersistUpdaterIT.java | 208 ++++++++++++++++++
.../test/ElasticsearchPersistWriterIT.java | 124 +++--------
...ElasticsearchPersistWriterParentChildIT.java | 210 -------------------
.../resources/ActivityChildObjectParent.json | 2 +-
.../ElasticsearchParentChildUpdaterIT.conf | 8 +
.../ElasticsearchParentChildWriterIT.conf | 8 +
.../ElasticsearchPersistUpdaterIT.conf | 8 +
...ElasticsearchPersistWriterParentChildIT.conf | 8 -
12 files changed, 645 insertions(+), 328 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e23901c2/streams-contrib/streams-persist-elasticsearch/pom.xml
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/pom.xml b/streams-contrib/streams-persist-elasticsearch/pom.xml
index fe7f798..7891baa 100644
--- a/streams-contrib/streams-persist-elasticsearch/pom.xml
+++ b/streams-contrib/streams-persist-elasticsearch/pom.xml
@@ -115,10 +115,6 @@
<scope>test</scope>
<type>test-jar</type>
</dependency>
- <dependency>
- <groupId>org.hamcrest</groupId>
- <artifactId>hamcrest-all</artifactId>
- </dependency>
</dependencies>
<dependencyManagement>
<dependencies>
@@ -145,11 +141,6 @@
<version>${lucene.version}</version>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.hamcrest</groupId>
- <artifactId>hamcrest-all</artifactId>
- <version>1.3</version>
- </dependency>
</dependencies>
</dependencyManagement>
<build>
@@ -233,6 +224,22 @@
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <version>${failsafe.plugin.version}</version>
+ <configuration>
+ <!-- Run integration test suite rather than individual tests. -->
+ <excludes>
+ <exclude>**/*Test.java</exclude>
+ <exclude>**/*Tests.java</exclude>
+ <include>**/*IT.java</include>
+ </excludes>
+ <includes>
+ <include>**/*ITs.java</include>
+ </includes>
+ </configuration>
+ </plugin>
</plugins>
</build>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e23901c2/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchITs.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchITs.java b/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchITs.java
new file mode 100644
index 0000000..504172e
--- /dev/null
+++ b/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchITs.java
@@ -0,0 +1,18 @@
+package org.apache.streams.elasticsearch.test;
+
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+@RunWith(Suite.class)
+@Suite.SuiteClasses({
+ ElasticsearchPersistWriterIT.class,
+ ElasticsearchPersistUpdaterIT.class,
+ ElasticsearchParentChildWriterIT.class,
+ ElasticsearchParentChildUpdaterIT.class,
+ DatumFromMetadataProcessorIT.class
+})
+
+public class ElasticsearchITs {
+ // the class remains empty,
+ // used only as a holder for the above annotations
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e23901c2/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchParentChildUpdaterIT.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchParentChildUpdaterIT.java b/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchParentChildUpdaterIT.java
new file mode 100644
index 0000000..c37f920
--- /dev/null
+++ b/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchParentChildUpdaterIT.java
@@ -0,0 +1,159 @@
+/*
+ * 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
+ *
+ * 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.streams.elasticsearch.test;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Strings;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigParseOptions;
+import org.apache.commons.io.Charsets;
+import org.apache.commons.io.IOUtils;
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.config.StreamsConfiguration;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.elasticsearch.ElasticsearchClientManager;
+import org.apache.streams.elasticsearch.ElasticsearchPersistUpdater;
+import org.apache.streams.elasticsearch.ElasticsearchPersistWriter;
+import org.apache.streams.elasticsearch.ElasticsearchWriterConfiguration;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.ActivityObject;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
+import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
+import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
+import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateRequestBuilder;
+import org.elasticsearch.action.count.CountRequest;
+import org.elasticsearch.action.count.CountResponse;
+import org.elasticsearch.action.search.SearchRequestBuilder;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.client.Requests;
+import org.elasticsearch.cluster.health.ClusterHealthStatus;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.junit.Before;
+import org.junit.Test;
+import org.reflections.Reflections;
+import org.reflections.scanners.SubTypesScanner;
+import org.reflections.util.ClasspathHelper;
+import org.reflections.util.ConfigurationBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.InputStream;
+import java.net.URL;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+import static junit.framework.TestCase.assertTrue;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+/**
+ * Created by sblackmon on 10/20/14.
+ */
+public class ElasticsearchParentChildUpdaterIT {
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(ElasticsearchParentChildUpdaterIT.class);
+
+ private static ObjectMapper MAPPER = StreamsJacksonMapper.getInstance();
+
+ protected ElasticsearchWriterConfiguration testConfiguration;
+ protected Client testClient;
+
+ Set<Class<? extends ActivityObject>> objectTypes;
+
+ List<String> files;
+
+ @Before
+ public void prepareTest() throws Exception {
+
+ Config reference = ConfigFactory.load();
+ File conf_file = new File("target/test-classes/ElasticsearchParentChildUpdaterIT.conf");
+ assert(conf_file.exists());
+ Config testResourceConfig = ConfigFactory.parseFileAnySyntax(conf_file, ConfigParseOptions.defaults().setAllowMissing(false));
+ Properties es_properties = new Properties();
+ InputStream es_stream = new FileInputStream("elasticsearch.properties");
+ es_properties.load(es_stream);
+ Config esProps = ConfigFactory.parseProperties(es_properties);
+ Config typesafe = testResourceConfig.withFallback(esProps).withFallback(reference).resolve();
+ StreamsConfiguration streams = StreamsConfigurator.detectConfiguration(typesafe);
+ testConfiguration = new ComponentConfigurator<>(ElasticsearchWriterConfiguration.class).detectConfiguration(typesafe, "elasticsearch");
+ testClient = new ElasticsearchClientManager(testConfiguration).getClient();
+
+ ClusterHealthRequest clusterHealthRequest = Requests.clusterHealthRequest();
+ ClusterHealthResponse clusterHealthResponse = testClient.admin().cluster().health(clusterHealthRequest).actionGet();
+ assertNotEquals(clusterHealthResponse.getStatus(), ClusterHealthStatus.RED);
+
+ IndicesExistsRequest indicesExistsRequest = Requests.indicesExistsRequest(testConfiguration.getIndex());
+ IndicesExistsResponse indicesExistsResponse = testClient.admin().indices().exists(indicesExistsRequest).actionGet();
+ assertTrue(indicesExistsResponse.isExists());
+
+ Reflections reflections = new Reflections(new ConfigurationBuilder()
+ .setUrls(ClasspathHelper.forPackage("org.apache.streams.pojo.json"))
+ .setScanners(new SubTypesScanner()));
+ objectTypes = reflections.getSubTypesOf(ActivityObject.class);
+
+ InputStream testActivityFolderStream = ElasticsearchParentChildUpdaterIT.class.getClassLoader()
+ .getResourceAsStream("activities");
+ files = IOUtils.readLines(testActivityFolderStream, Charsets.UTF_8);
+
+ }
+
+ @Test
+ public void testPersistUpdater() throws Exception {
+
+ ElasticsearchPersistUpdater testPersistUpdater = new ElasticsearchPersistUpdater(testConfiguration);
+ testPersistUpdater.prepare(null);
+
+ for( String file : files) {
+ LOGGER.info("File: " + file );
+ InputStream testActivityFileStream = ElasticsearchParentChildUpdaterIT.class.getClassLoader()
+ .getResourceAsStream("activities/" + file);
+ Activity activity = MAPPER.readValue(testActivityFileStream, Activity.class);
+ activity.setAdditionalProperty("updated", Boolean.TRUE);
+ StreamsDatum datum = new StreamsDatum(activity, activity.getVerb());
+ if( !Strings.isNullOrEmpty(activity.getObject().getObjectType())) {
+ datum.getMetadata().put("parent", activity.getObject().getObjectType());
+ datum.getMetadata().put("type", "activity");
+ testPersistUpdater.write(datum);
+ LOGGER.info("Updated: " + activity.getVerb() );
+ }
+ }
+
+ testPersistUpdater.cleanUp();
+
+ SearchRequestBuilder countUpdatedRequest = testClient
+ .prepareSearch(testConfiguration.getIndex())
+ .setTypes("activity")
+ .setQuery(QueryBuilders.queryStringQuery("updated:true"));
+ SearchResponse countUpdatedResponse = countUpdatedRequest.execute().actionGet();
+
+ assertEquals(84, countUpdatedResponse.getHits().getTotalHits());
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e23901c2/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchParentChildWriterIT.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchParentChildWriterIT.java b/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchParentChildWriterIT.java
new file mode 100644
index 0000000..7254913
--- /dev/null
+++ b/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchParentChildWriterIT.java
@@ -0,0 +1,195 @@
+/*
+ * 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
+ *
+ * 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.streams.elasticsearch.test;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigParseOptions;
+import org.apache.commons.io.Charsets;
+import org.apache.commons.io.IOUtils;
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.config.StreamsConfiguration;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.elasticsearch.ElasticsearchClientManager;
+import org.apache.streams.elasticsearch.ElasticsearchPersistUpdater;
+import org.apache.streams.elasticsearch.ElasticsearchPersistWriter;
+import org.apache.streams.elasticsearch.ElasticsearchWriterConfiguration;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.ActivityObject;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
+import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
+import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
+import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateRequestBuilder;
+import org.elasticsearch.action.count.CountRequest;
+import org.elasticsearch.action.count.CountResponse;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchRequestBuilder;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.client.Requests;
+import org.elasticsearch.cluster.health.ClusterHealthStatus;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.junit.Before;
+import org.junit.Test;
+import org.reflections.Reflections;
+import org.reflections.scanners.SubTypesScanner;
+import org.reflections.util.ClasspathHelper;
+import org.reflections.util.ConfigurationBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.InputStream;
+import java.net.URL;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+import static junit.framework.TestCase.assertTrue;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+
+/**
+ * Created by sblackmon on 10/20/14.
+ */
+public class ElasticsearchParentChildWriterIT {
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(ElasticsearchParentChildWriterIT.class);
+
+ private static ObjectMapper MAPPER = StreamsJacksonMapper.getInstance();
+
+ protected ElasticsearchWriterConfiguration testConfiguration;
+ protected Client testClient;
+
+ Set<Class<? extends ActivityObject>> objectTypes;
+
+ List<String> files;
+
+ @Before
+ public void prepareTest() throws Exception {
+
+ Config reference = ConfigFactory.load();
+ File conf_file = new File("target/test-classes/ElasticsearchParentChildWriterIT.conf");
+ assert(conf_file.exists());
+ Config testResourceConfig = ConfigFactory.parseFileAnySyntax(conf_file, ConfigParseOptions.defaults().setAllowMissing(false));
+ Properties es_properties = new Properties();
+ InputStream es_stream = new FileInputStream("elasticsearch.properties");
+ es_properties.load(es_stream);
+ Config esProps = ConfigFactory.parseProperties(es_properties);
+ Config typesafe = testResourceConfig.withFallback(esProps).withFallback(reference).resolve();
+ StreamsConfiguration streams = StreamsConfigurator.detectConfiguration(typesafe);
+ testConfiguration = new ComponentConfigurator<>(ElasticsearchWriterConfiguration.class).detectConfiguration(typesafe, "elasticsearch");
+ testClient = new ElasticsearchClientManager(testConfiguration).getClient();
+
+ ClusterHealthRequest clusterHealthRequest = Requests.clusterHealthRequest();
+ ClusterHealthResponse clusterHealthResponse = testClient.admin().cluster().health(clusterHealthRequest).actionGet();
+ assertNotEquals(clusterHealthResponse.getStatus(), ClusterHealthStatus.RED);
+
+ IndicesExistsRequest indicesExistsRequest = Requests.indicesExistsRequest(testConfiguration.getIndex());
+ IndicesExistsResponse indicesExistsResponse = testClient.admin().indices().exists(indicesExistsRequest).actionGet();
+ if(indicesExistsResponse.isExists()) {
+ DeleteIndexRequest deleteIndexRequest = Requests.deleteIndexRequest(testConfiguration.getIndex());
+ DeleteIndexResponse deleteIndexResponse = testClient.admin().indices().delete(deleteIndexRequest).actionGet();
+ assertTrue(deleteIndexResponse.isAcknowledged());
+ };
+
+ PutIndexTemplateRequestBuilder putTemplateRequestBuilder = testClient.admin().indices().preparePutTemplate("mappings");
+ URL templateURL = ElasticsearchParentChildWriterIT.class.getResource("/ActivityChildObjectParent.json");
+ ObjectNode template = MAPPER.readValue(templateURL, ObjectNode.class);
+ String templateSource = MAPPER.writeValueAsString(template);
+ putTemplateRequestBuilder.setSource(templateSource);
+
+ testClient.admin().indices().putTemplate(putTemplateRequestBuilder.request()).actionGet();
+
+ Reflections reflections = new Reflections(new ConfigurationBuilder()
+ .setUrls(ClasspathHelper.forPackage("org.apache.streams.pojo.json"))
+ .setScanners(new SubTypesScanner()));
+ objectTypes = reflections.getSubTypesOf(ActivityObject.class);
+
+ InputStream testActivityFolderStream = ElasticsearchParentChildWriterIT.class.getClassLoader()
+ .getResourceAsStream("activities");
+ files = IOUtils.readLines(testActivityFolderStream, Charsets.UTF_8);
+
+ }
+
+ @Test
+ public void testPersistWriter() throws Exception {
+
+ IndicesExistsRequest indicesExistsRequest = Requests.indicesExistsRequest(testConfiguration.getIndex());
+ IndicesExistsResponse indicesExistsResponse = testClient.admin().indices().exists(indicesExistsRequest).actionGet();
+ if(indicesExistsResponse.isExists()) {
+ DeleteIndexRequest deleteIndexRequest = Requests.deleteIndexRequest(testConfiguration.getIndex());
+ DeleteIndexResponse deleteIndexResponse = testClient.admin().indices().delete(deleteIndexRequest).actionGet();
+ };
+
+ ElasticsearchPersistWriter testPersistWriter = new ElasticsearchPersistWriter(testConfiguration);
+ testPersistWriter.prepare(null);
+
+ for( Class objectType : objectTypes ) {
+ Object object = objectType.newInstance();
+ ActivityObject activityObject = MAPPER.convertValue(object, ActivityObject.class);
+ StreamsDatum datum = new StreamsDatum(activityObject, activityObject.getObjectType());
+ datum.getMetadata().put("type", "object");
+ testPersistWriter.write( datum );
+ }
+
+ for( String file : files) {
+ LOGGER.info("File: " + file );
+ InputStream testActivityFileStream = ElasticsearchParentChildWriterIT.class.getClassLoader()
+ .getResourceAsStream("activities/" + file);
+ Activity activity = MAPPER.readValue(testActivityFileStream, Activity.class);
+ StreamsDatum datum = new StreamsDatum(activity, activity.getVerb());
+ if( !Strings.isNullOrEmpty(activity.getObject().getObjectType())) {
+ datum.getMetadata().put("parent", activity.getObject().getObjectType());
+ datum.getMetadata().put("type", "activity");
+ testPersistWriter.write(datum);
+ LOGGER.info("Wrote: " + activity.getVerb());
+ }
+ }
+
+ testPersistWriter.cleanUp();
+
+ SearchRequestBuilder countParentRequest = testClient
+ .prepareSearch(testConfiguration.getIndex())
+ .setTypes("object");
+ SearchResponse countParentResponse = countParentRequest.execute().actionGet();
+
+ assertEquals(41, countParentResponse.getHits().getTotalHits());
+
+ SearchRequestBuilder countChildRequest = testClient
+ .prepareSearch(testConfiguration.getIndex())
+ .setTypes("activity");
+ SearchResponse countChildResponse = countChildRequest.execute().actionGet();
+
+ assertEquals(84, countChildResponse.getHits().getTotalHits());
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e23901c2/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchPersistUpdaterIT.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchPersistUpdaterIT.java b/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchPersistUpdaterIT.java
new file mode 100644
index 0000000..ab6337f
--- /dev/null
+++ b/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchPersistUpdaterIT.java
@@ -0,0 +1,208 @@
+/*
+ * 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
+ *
+ * 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.streams.elasticsearch.test;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigParseOptions;
+import org.apache.commons.io.Charsets;
+import org.apache.commons.io.IOUtils;
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.config.StreamsConfiguration;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.elasticsearch.ElasticsearchClientManager;
+import org.apache.streams.elasticsearch.ElasticsearchPersistUpdater;
+import org.apache.streams.elasticsearch.ElasticsearchPersistWriter;
+import org.apache.streams.elasticsearch.ElasticsearchWriterConfiguration;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.Actor;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
+import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
+import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
+import org.elasticsearch.action.search.SearchRequestBuilder;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.client.Requests;
+import org.elasticsearch.cluster.health.ClusterHealthStatus;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.InputStream;
+import java.util.List;
+import java.util.Properties;
+
+import static junit.framework.TestCase.assertTrue;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+/**
+ * Created by sblackmon on 10/20/14.
+ */
+public class ElasticsearchPersistUpdaterIT {
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(ElasticsearchPersistUpdaterIT.class);
+
+ private static ObjectMapper MAPPER = StreamsJacksonMapper.getInstance();
+
+ protected ElasticsearchWriterConfiguration testConfiguration;
+ protected Client testClient;
+
+ @Before
+ public void prepareTest() throws Exception {
+
+ Config reference = ConfigFactory.load();
+ File conf_file = new File("target/test-classes/ElasticsearchPersistUpdaterIT.conf");
+ assert(conf_file.exists());
+ Config testResourceConfig = ConfigFactory.parseFileAnySyntax(conf_file, ConfigParseOptions.defaults().setAllowMissing(false));
+ Properties es_properties = new Properties();
+ InputStream es_stream = new FileInputStream("elasticsearch.properties");
+ es_properties.load(es_stream);
+ Config esProps = ConfigFactory.parseProperties(es_properties);
+ Config typesafe = testResourceConfig.withFallback(esProps).withFallback(reference).resolve();
+ StreamsConfiguration streams = StreamsConfigurator.detectConfiguration(typesafe);
+ testConfiguration = new ComponentConfigurator<>(ElasticsearchWriterConfiguration.class).detectConfiguration(typesafe, "elasticsearch");
+ testClient = new ElasticsearchClientManager(testConfiguration).getClient();
+
+ ClusterHealthRequest clusterHealthRequest = Requests.clusterHealthRequest();
+ ClusterHealthResponse clusterHealthResponse = testClient.admin().cluster().health(clusterHealthRequest).actionGet();
+ assertNotEquals(clusterHealthResponse.getStatus(), ClusterHealthStatus.RED);
+
+ IndicesExistsRequest indicesExistsRequest = Requests.indicesExistsRequest(testConfiguration.getIndex());
+ IndicesExistsResponse indicesExistsResponse = testClient.admin().indices().exists(indicesExistsRequest).actionGet();
+ assertTrue(indicesExistsResponse.isExists());
+
+ }
+
+ @Test
+ public void testPersistUpdater() throws Exception {
+
+ IndicesExistsRequest indicesExistsRequest = Requests.indicesExistsRequest(testConfiguration.getIndex());
+ IndicesExistsResponse indicesExistsResponse = testClient.admin().indices().exists(indicesExistsRequest).actionGet();
+ assertTrue(indicesExistsResponse.isExists());
+
+ SearchRequestBuilder countRequest = testClient
+ .prepareSearch(testConfiguration.getIndex())
+ .setTypes(testConfiguration.getType());
+ SearchResponse countResponse = countRequest.execute().actionGet();
+
+ long count = countResponse.getHits().getTotalHits();
+
+ ElasticsearchPersistUpdater testPersistUpdater = new ElasticsearchPersistUpdater(testConfiguration);
+ testPersistUpdater.prepare(null);
+
+ InputStream testActivityFolderStream = ElasticsearchPersistUpdaterIT.class.getClassLoader()
+ .getResourceAsStream("activities");
+ List<String> files = IOUtils.readLines(testActivityFolderStream, Charsets.UTF_8);
+
+ for( String file : files) {
+ LOGGER.info("File: " + file );
+ InputStream testActivityFileStream = ElasticsearchPersistUpdaterIT.class.getClassLoader()
+ .getResourceAsStream("activities/" + file);
+ Activity activity = MAPPER.readValue(testActivityFileStream, Activity.class);
+ Activity update = new Activity();
+ update.setAdditionalProperty("updated", Boolean.TRUE);
+ update.setAdditionalProperty("str", "str");
+ update.setAdditionalProperty("long", 10l);
+ update.setActor(
+ (Actor) new Actor()
+ .withAdditionalProperty("updated", Boolean.TRUE)
+ .withAdditionalProperty("double", 10d)
+ .withAdditionalProperty("map",
+ MAPPER.createObjectNode().set("field", MAPPER.createArrayNode().add("item"))));
+
+ StreamsDatum datum = new StreamsDatum(update, activity.getVerb());
+ testPersistUpdater.write( datum );
+ LOGGER.info("Updated: " + activity.getVerb() );
+ }
+
+ testPersistUpdater.cleanUp();
+
+ SearchRequestBuilder updatedCountRequest = testClient
+ .prepareSearch(testConfiguration.getIndex())
+ .setTypes(testConfiguration.getType())
+ .setQuery(QueryBuilders.existsQuery("updated"));
+ SearchResponse updatedCount = updatedCountRequest.execute().actionGet();
+
+ LOGGER.info("updated: {}", updatedCount.getHits().getTotalHits());
+
+ assertEquals(count, updatedCount.getHits().getTotalHits());
+
+ SearchRequestBuilder actorUpdatedCountRequest = testClient
+ .prepareSearch(testConfiguration.getIndex())
+ .setTypes(testConfiguration.getType())
+ .setQuery(QueryBuilders.termQuery("actor.updated", true));
+ SearchResponse actorUpdatedCount = actorUpdatedCountRequest.execute().actionGet();
+
+ LOGGER.info("actor.updated: {}", actorUpdatedCount.getHits().getTotalHits());
+
+ assertEquals(count, actorUpdatedCount.getHits().getTotalHits());
+
+ SearchRequestBuilder strUpdatedCountRequest = testClient
+ .prepareSearch(testConfiguration.getIndex())
+ .setTypes(testConfiguration.getType())
+ .setQuery(QueryBuilders.termQuery("str", "str"));
+ SearchResponse strUpdatedCount = strUpdatedCountRequest.execute().actionGet();
+
+ LOGGER.info("strupdated: {}", strUpdatedCount.getHits().getTotalHits());
+
+ assertEquals(count, strUpdatedCount.getHits().getTotalHits());
+
+ SearchRequestBuilder longUpdatedCountRequest = testClient
+ .prepareSearch(testConfiguration.getIndex())
+ .setTypes(testConfiguration.getType())
+ .setQuery(QueryBuilders.rangeQuery("long").from(9).to(11));
+ SearchResponse longUpdatedCount = longUpdatedCountRequest.execute().actionGet();
+
+ LOGGER.info("longupdated: {}", longUpdatedCount.getHits().getTotalHits());
+
+ assertEquals(count, longUpdatedCount.getHits().getTotalHits());
+
+ SearchRequestBuilder doubleUpdatedCountRequest = testClient
+ .prepareSearch(testConfiguration.getIndex())
+ .setTypes(testConfiguration.getType())
+ .setQuery(QueryBuilders.rangeQuery("long").from(9).to(11));
+ SearchResponse doubleUpdatedCount = doubleUpdatedCountRequest.execute().actionGet();
+
+ LOGGER.info("doubleupdated: {}", doubleUpdatedCount.getHits().getTotalHits());
+
+ assertEquals(count, doubleUpdatedCount.getHits().getTotalHits());
+
+ SearchRequestBuilder mapUpdatedCountRequest = testClient
+ .prepareSearch(testConfiguration.getIndex())
+ .setTypes(testConfiguration.getType())
+ .setQuery(QueryBuilders.termQuery("actor.map.field", "item"));
+ SearchResponse mapUpdatedCount = mapUpdatedCountRequest.execute().actionGet();
+
+ LOGGER.info("mapfieldupdated: {}", mapUpdatedCount.getHits().getTotalHits());
+
+ assertEquals(count, mapUpdatedCount.getHits().getTotalHits());
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e23901c2/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchPersistWriterIT.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchPersistWriterIT.java b/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchPersistWriterIT.java
index cf2fdfd..e4dfba2 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchPersistWriterIT.java
+++ b/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchPersistWriterIT.java
@@ -37,15 +37,22 @@ import org.apache.streams.elasticsearch.ElasticsearchWriterConfiguration;
import org.apache.streams.jackson.StreamsJacksonMapper;
import org.apache.streams.pojo.json.Activity;
import org.apache.streams.pojo.json.Actor;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
+import org.elasticsearch.action.search.SearchRequestBuilder;
+import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
+import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.elasticsearch.index.query.QueryBuilders;
import org.junit.Before;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -57,6 +64,7 @@ import java.util.*;
import static junit.framework.TestCase.assertTrue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
/**
* Created by sblackmon on 10/20/14.
@@ -86,23 +94,23 @@ public class ElasticsearchPersistWriterIT {
testConfiguration = new ComponentConfigurator<>(ElasticsearchWriterConfiguration.class).detectConfiguration(typesafe, "elasticsearch");
testClient = new ElasticsearchClientManager(testConfiguration).getClient();
- }
-
- @Test
- public void testPersist() throws Exception {
- testPersistWriter();
- testPersistUpdater();
- }
-
- void testPersistWriter() throws Exception {
+ ClusterHealthRequest clusterHealthRequest = Requests.clusterHealthRequest();
+ ClusterHealthResponse clusterHealthResponse = testClient.admin().cluster().health(clusterHealthRequest).actionGet();
+ assertNotEquals(clusterHealthResponse.getStatus(), ClusterHealthStatus.RED);
IndicesExistsRequest indicesExistsRequest = Requests.indicesExistsRequest(testConfiguration.getIndex());
IndicesExistsResponse indicesExistsResponse = testClient.admin().indices().exists(indicesExistsRequest).actionGet();
if(indicesExistsResponse.isExists()) {
DeleteIndexRequest deleteIndexRequest = Requests.deleteIndexRequest(testConfiguration.getIndex());
DeleteIndexResponse deleteIndexResponse = testClient.admin().indices().delete(deleteIndexRequest).actionGet();
+ assertTrue(deleteIndexResponse.isAcknowledged());
};
+ }
+
+ @Test
+ public void testPersistWriter() throws Exception {
+
ElasticsearchPersistWriter testPersistWriter = new ElasticsearchPersistWriter(testConfiguration);
testPersistWriter.prepare(null);
@@ -118,101 +126,17 @@ public class ElasticsearchPersistWriterIT {
StreamsDatum datum = new StreamsDatum(activity, activity.getVerb());
testPersistWriter.write( datum );
LOGGER.info("Wrote: " + activity.getVerb() );
- }
-
- testPersistWriter.cleanUp();
-
- long count = testClient.count(testClient.prepareCount().request()).actionGet().getCount();
-
- assert(count == 89);
-
- }
-
- void testPersistUpdater() throws Exception {
-
- IndicesExistsRequest indicesExistsRequest = Requests.indicesExistsRequest(testConfiguration.getIndex());
- IndicesExistsResponse indicesExistsResponse = testClient.admin().indices().exists(indicesExistsRequest).actionGet();
- assertTrue(indicesExistsResponse.isExists());
-
- long count = testClient.count(testClient.prepareCount().request()).actionGet().getCount();
-
- ElasticsearchPersistUpdater testPersistUpdater = new ElasticsearchPersistUpdater(testConfiguration);
- testPersistUpdater.prepare(null);
-
- InputStream testActivityFolderStream = ElasticsearchPersistWriterIT.class.getClassLoader()
- .getResourceAsStream("activities");
- List<String> files = IOUtils.readLines(testActivityFolderStream, Charsets.UTF_8);
-
- for( String file : files) {
- LOGGER.info("File: " + file );
- InputStream testActivityFileStream = ElasticsearchPersistWriterIT.class.getClassLoader()
- .getResourceAsStream("activities/" + file);
- Activity activity = MAPPER.readValue(testActivityFileStream, Activity.class);
- Activity update = new Activity();
- update.setAdditionalProperty("updated", Boolean.TRUE);
- update.setAdditionalProperty("str", "str");
- update.setAdditionalProperty("long", 10l);
- update.setActor(
- (Actor) new Actor()
- .withAdditionalProperty("updated", Boolean.TRUE)
- .withAdditionalProperty("double", 10d)
- .withAdditionalProperty("map",
- MAPPER.createObjectNode().set("field", MAPPER.createArrayNode().add("item"))));
-
- StreamsDatum datum = new StreamsDatum(update, activity.getVerb());
- testPersistUpdater.write( datum );
- LOGGER.info("Updated: " + activity.getVerb() );
}
- testPersistUpdater.cleanUp();
-
- long updated = testClient.prepareCount().setQuery(
- QueryBuilders.existsQuery("updated")
- ).execute().actionGet().getCount();
-
- LOGGER.info("updated: {}", updated);
-
- assertEquals(count, updated);
-
- long actorupdated = testClient.prepareCount().setQuery(
- QueryBuilders.termQuery("actor.updated", true)
- ).execute().actionGet().getCount();
-
- LOGGER.info("actor.updated: {}", actorupdated);
-
- assertEquals(count, actorupdated);
-
- long strupdated = testClient.prepareCount().setQuery(
- QueryBuilders.termQuery("str", "str")
- ).execute().actionGet().getCount();
+ testPersistWriter.cleanUp();
- LOGGER.info("strupdated: {}", strupdated);
+ SearchRequestBuilder countRequest = testClient
+ .prepareSearch(testConfiguration.getIndex())
+ .setTypes(testConfiguration.getType());
+ SearchResponse countResponse = countRequest.execute().actionGet();
- assertEquals(count, strupdated);
-
- long longupdated = testClient.prepareCount().setQuery(
- QueryBuilders.rangeQuery("long").from(9).to(11)
- ).execute().actionGet().getCount();
-
- LOGGER.info("longupdated: {}", longupdated);
-
- assertEquals(count, longupdated);
-
- long doubleupdated = testClient.prepareCount().setQuery(
- QueryBuilders.rangeQuery("long").from(9).to(11)
- ).execute().actionGet().getCount();
-
- LOGGER.info("doubleupdated: {}", doubleupdated);
-
- assertEquals(count, doubleupdated);
-
- long mapfieldupdated = testClient.prepareCount().setQuery(
- QueryBuilders.termQuery("actor.map.field", "item")
- ).execute().actionGet().getCount();
-
- LOGGER.info("mapfieldupdated: {}", mapfieldupdated);
-
- assertEquals(count, mapfieldupdated);
+ assertEquals(89, countResponse.getHits().getTotalHits());
}
+
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e23901c2/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchPersistWriterParentChildIT.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchPersistWriterParentChildIT.java b/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchPersistWriterParentChildIT.java
deleted file mode 100644
index f70ccf8..0000000
--- a/streams-contrib/streams-persist-elasticsearch/src/test/java/org/apache/streams/elasticsearch/test/ElasticsearchPersistWriterParentChildIT.java
+++ /dev/null
@@ -1,210 +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
- *
- * 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.streams.elasticsearch.test;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import com.google.common.base.Strings;
-import com.google.common.collect.Lists;
-import com.typesafe.config.Config;
-import com.typesafe.config.ConfigFactory;
-import com.typesafe.config.ConfigParseOptions;
-import org.apache.commons.io.Charsets;
-import org.apache.commons.io.IOUtils;
-import org.apache.streams.config.ComponentConfigurator;
-import org.apache.streams.config.StreamsConfiguration;
-import org.apache.streams.config.StreamsConfigurator;
-import org.apache.streams.core.StreamsDatum;
-import org.apache.streams.elasticsearch.ElasticsearchClientManager;
-import org.apache.streams.elasticsearch.ElasticsearchPersistUpdater;
-import org.apache.streams.elasticsearch.ElasticsearchPersistWriter;
-import org.apache.streams.elasticsearch.ElasticsearchWriterConfiguration;
-import org.apache.streams.jackson.StreamsJacksonMapper;
-import org.apache.streams.pojo.json.Activity;
-import org.apache.streams.pojo.json.ActivityObject;
-import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
-import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
-import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
-import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
-import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateRequestBuilder;
-import org.elasticsearch.action.count.CountRequest;
-import org.elasticsearch.action.count.CountResponse;
-import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.action.search.SearchRequestBuilder;
-import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.client.Client;
-import org.elasticsearch.client.Requests;
-import org.elasticsearch.index.query.QueryBuilders;
-import org.junit.Before;
-import org.junit.Test;
-import org.reflections.Reflections;
-import org.reflections.scanners.SubTypesScanner;
-import org.reflections.util.ClasspathHelper;
-import org.reflections.util.ConfigurationBuilder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.InputStream;
-import java.net.URL;
-import java.util.List;
-import java.util.Properties;
-import java.util.Set;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-
-/**
- * Created by sblackmon on 10/20/14.
- */
-public class ElasticsearchPersistWriterParentChildIT {
-
- private final static Logger LOGGER = LoggerFactory.getLogger(ElasticsearchPersistWriterParentChildIT.class);
-
- private static ObjectMapper MAPPER = StreamsJacksonMapper.getInstance();
-
- protected ElasticsearchWriterConfiguration testConfiguration;
- protected Client testClient;
-
- Set<Class<? extends ActivityObject>> objectTypes;
-
- List<String> files;
-
- @Before
- public void prepareTest() throws Exception {
-
- Config reference = ConfigFactory.load();
- File conf_file = new File("target/test-classes/ElasticsearchPersistWriterParentChildIT.conf");
- assert(conf_file.exists());
- Config testResourceConfig = ConfigFactory.parseFileAnySyntax(conf_file, ConfigParseOptions.defaults().setAllowMissing(false));
- Properties es_properties = new Properties();
- InputStream es_stream = new FileInputStream("elasticsearch.properties");
- es_properties.load(es_stream);
- Config esProps = ConfigFactory.parseProperties(es_properties);
- Config typesafe = testResourceConfig.withFallback(esProps).withFallback(reference).resolve();
- StreamsConfiguration streams = StreamsConfigurator.detectConfiguration(typesafe);
- testConfiguration = new ComponentConfigurator<>(ElasticsearchWriterConfiguration.class).detectConfiguration(typesafe, "elasticsearch");
- testClient = new ElasticsearchClientManager(testConfiguration).getClient();
-
- PutIndexTemplateRequestBuilder putTemplateRequestBuilder = testClient.admin().indices().preparePutTemplate("mappings");
- URL templateURL = ElasticsearchPersistWriterParentChildIT.class.getResource("/ActivityChildObjectParent.json");
- ObjectNode template = MAPPER.readValue(templateURL, ObjectNode.class);
- String templateSource = MAPPER.writeValueAsString(template);
- putTemplateRequestBuilder.setSource(templateSource);
-
- testClient.admin().indices().putTemplate(putTemplateRequestBuilder.request()).actionGet();
-
- Reflections reflections = new Reflections(new ConfigurationBuilder()
- .setUrls(ClasspathHelper.forPackage("org.apache.streams.pojo.json"))
- .setScanners(new SubTypesScanner()));
- objectTypes = reflections.getSubTypesOf(ActivityObject.class);
-
- InputStream testActivityFolderStream = ElasticsearchPersistWriterParentChildIT.class.getClassLoader()
- .getResourceAsStream("activities");
- files = IOUtils.readLines(testActivityFolderStream, Charsets.UTF_8);
-
- }
-
- @Test
- public void testPersist() throws Exception {
- testPersistWriter();
- testPersistUpdater();
- }
-
- void testPersistWriter() throws Exception {
-
- IndicesExistsRequest indicesExistsRequest = Requests.indicesExistsRequest(testConfiguration.getIndex());
- IndicesExistsResponse indicesExistsResponse = testClient.admin().indices().exists(indicesExistsRequest).actionGet();
- if(indicesExistsResponse.isExists()) {
- DeleteIndexRequest deleteIndexRequest = Requests.deleteIndexRequest(testConfiguration.getIndex());
- DeleteIndexResponse deleteIndexResponse = testClient.admin().indices().delete(deleteIndexRequest).actionGet();
- };
-
- ElasticsearchPersistWriter testPersistWriter = new ElasticsearchPersistWriter(testConfiguration);
- testPersistWriter.prepare(null);
-
- for( Class objectType : objectTypes ) {
- Object object = objectType.newInstance();
- ActivityObject activityObject = MAPPER.convertValue(object, ActivityObject.class);
- StreamsDatum datum = new StreamsDatum(activityObject, activityObject.getObjectType());
- datum.getMetadata().put("type", "object");
- testPersistWriter.write( datum );
- }
-
- for( String file : files) {
- LOGGER.info("File: " + file );
- InputStream testActivityFileStream = ElasticsearchPersistWriterParentChildIT.class.getClassLoader()
- .getResourceAsStream("activities/" + file);
- Activity activity = MAPPER.readValue(testActivityFileStream, Activity.class);
- StreamsDatum datum = new StreamsDatum(activity, activity.getVerb());
- if( !Strings.isNullOrEmpty(activity.getObject().getObjectType())) {
- datum.getMetadata().put("parent", activity.getObject().getObjectType());
- datum.getMetadata().put("type", "activity");
- testPersistWriter.write(datum);
- LOGGER.info("Wrote: " + activity.getVerb());
- }
- }
-
- testPersistWriter.cleanUp();
-
- CountRequest countParentRequest = Requests.countRequest(testConfiguration.getIndex()).types("object");
- CountResponse countParentResponse = testClient.count(countParentRequest).actionGet();
-
- assertEquals(41, countParentResponse.getCount());
-
- CountRequest countChildRequest = Requests.countRequest(testConfiguration.getIndex()).types("activity");
- CountResponse countChildResponse = testClient.count(countChildRequest).actionGet();
-
- assertEquals(84, countChildResponse.getCount());
-
- }
-
- void testPersistUpdater() throws Exception {
-
- ElasticsearchPersistUpdater testPersistUpdater = new ElasticsearchPersistUpdater(testConfiguration);
- testPersistUpdater.prepare(null);
-
- for( String file : files) {
- LOGGER.info("File: " + file );
- InputStream testActivityFileStream = ElasticsearchPersistWriterParentChildIT.class.getClassLoader()
- .getResourceAsStream("activities/" + file);
- Activity activity = MAPPER.readValue(testActivityFileStream, Activity.class);
- activity.setAdditionalProperty("updated", Boolean.TRUE);
- StreamsDatum datum = new StreamsDatum(activity, activity.getVerb());
- if( !Strings.isNullOrEmpty(activity.getObject().getObjectType())) {
- datum.getMetadata().put("parent", activity.getObject().getObjectType());
- datum.getMetadata().put("type", "activity");
- testPersistUpdater.write(datum);
- LOGGER.info("Updated: " + activity.getVerb() );
- }
- }
-
- testPersistUpdater.cleanUp();
-
- SearchRequestBuilder countUpdatedRequest = testClient
- .prepareSearch(testConfiguration.getIndex())
- .setTypes("activity")
- .setQuery(QueryBuilders.queryStringQuery("updated:true"));
- SearchResponse countUpdatedResponse = countUpdatedRequest.execute().actionGet();
-
- assertEquals(84, countUpdatedResponse.getHits().getTotalHits());
-
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e23901c2/streams-contrib/streams-persist-elasticsearch/src/test/resources/ActivityChildObjectParent.json
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/test/resources/ActivityChildObjectParent.json b/streams-contrib/streams-persist-elasticsearch/src/test/resources/ActivityChildObjectParent.json
index 14f90a8..923c648 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/test/resources/ActivityChildObjectParent.json
+++ b/streams-contrib/streams-persist-elasticsearch/src/test/resources/ActivityChildObjectParent.json
@@ -2,7 +2,7 @@
"$license": [
"http://www.apache.org/licenses/LICENSE-2.0"
],
- "template": "*",
+ "template": "elasticsearch_persist_writer_parent_child_it",
"order": 100,
"mappings": {
"object": {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e23901c2/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchParentChildUpdaterIT.conf
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchParentChildUpdaterIT.conf b/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchParentChildUpdaterIT.conf
new file mode 100644
index 0000000..70a53d9
--- /dev/null
+++ b/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchParentChildUpdaterIT.conf
@@ -0,0 +1,8 @@
+elasticsearch {
+ hosts += ${es.tcp.host}
+ port = ${es.tcp.port}
+ clusterName = "elasticsearch"
+ index = "elasticsearch_persist_writer_parent_child_it"
+ batchSize = 5
+ refresh = true
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e23901c2/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchParentChildWriterIT.conf
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchParentChildWriterIT.conf b/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchParentChildWriterIT.conf
new file mode 100644
index 0000000..70a53d9
--- /dev/null
+++ b/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchParentChildWriterIT.conf
@@ -0,0 +1,8 @@
+elasticsearch {
+ hosts += ${es.tcp.host}
+ port = ${es.tcp.port}
+ clusterName = "elasticsearch"
+ index = "elasticsearch_persist_writer_parent_child_it"
+ batchSize = 5
+ refresh = true
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e23901c2/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchPersistUpdaterIT.conf
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchPersistUpdaterIT.conf b/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchPersistUpdaterIT.conf
new file mode 100644
index 0000000..4eb787f
--- /dev/null
+++ b/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchPersistUpdaterIT.conf
@@ -0,0 +1,8 @@
+elasticsearch {
+ hosts += ${es.tcp.host}
+ port = ${es.tcp.port}
+ clusterName = "elasticsearch"
+ index = "elasticsearch_persist_writer_it"
+ type = "activity"
+ refresh = true
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e23901c2/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchPersistWriterParentChildIT.conf
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchPersistWriterParentChildIT.conf b/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchPersistWriterParentChildIT.conf
deleted file mode 100644
index 70a53d9..0000000
--- a/streams-contrib/streams-persist-elasticsearch/src/test/resources/ElasticsearchPersistWriterParentChildIT.conf
+++ /dev/null
@@ -1,8 +0,0 @@
-elasticsearch {
- hosts += ${es.tcp.host}
- port = ${es.tcp.port}
- clusterName = "elasticsearch"
- index = "elasticsearch_persist_writer_parent_child_it"
- batchSize = 5
- refresh = true
-}
\ No newline at end of file