JAMES-2398 Move ElasticSearch mailbox index to mailbox project
Project: http://git-wip-us.apache.org/repos/asf/james-project/repo Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/cdf8ce73 Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/cdf8ce73 Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/cdf8ce73 Branch: refs/heads/master Commit: cdf8ce73726ed642cf7eec8ef23b040b31123074 Parents: 4fe5a22 Author: Raphael Ouazana <[email protected]> Authored: Wed May 16 15:36:19 2018 +0200 Committer: Matthieu Baechler <[email protected]> Committed: Mon May 21 11:37:43 2018 +0200 ---------------------------------------------------------------------- .../es/ElasticSearchMailboxIndexer.java | 112 -------- .../backends/es/ElasticSearchIndexerTest.java | 257 ------------------ .../ElasticSearchMailboxIndexer.java | 118 ++++++++ .../ElasticSearchIntegrationTest.java | 1 - .../ElasticSearchMailboxIndexerTest.java | 266 +++++++++++++++++++ .../host/ElasticSearchHostSystem.java | 2 +- .../mailbox/ElasticSearchMailboxModule.java | 2 +- 7 files changed, 386 insertions(+), 372 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/james-project/blob/cdf8ce73/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchMailboxIndexer.java ---------------------------------------------------------------------- diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchMailboxIndexer.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchMailboxIndexer.java deleted file mode 100644 index 35d5849..0000000 --- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchMailboxIndexer.java +++ /dev/null @@ -1,112 +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.james.backends.es; - -import java.util.List; -import java.util.Optional; - -import javax.inject.Inject; -import javax.inject.Named; - -import org.apache.commons.lang3.StringUtils; -import org.elasticsearch.action.bulk.BulkRequestBuilder; -import org.elasticsearch.action.bulk.BulkResponse; -import org.elasticsearch.action.index.IndexResponse; -import org.elasticsearch.client.Client; -import org.elasticsearch.common.ValidationException; -import org.elasticsearch.index.query.QueryBuilder; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.google.common.base.Preconditions; - -public class ElasticSearchMailboxIndexer implements ElasticSearchIndexer { - private static int DEBUG_MAX_LENGTH_CONTENT = 1000; - - private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchMailboxIndexer.class); - - private final Client client; - private final DeleteByQueryPerformer deleteByQueryPerformer; - private final AliasName aliasName; - private final TypeName typeName; - - @Inject - public ElasticSearchMailboxIndexer(Client client, DeleteByQueryPerformer deleteByQueryPerformer, - @Named(ElasticSearchConstants.WRITE_ALIAS) AliasName aliasName, - TypeName typeName) { - this.client = client; - this.deleteByQueryPerformer = deleteByQueryPerformer; - this.aliasName = aliasName; - this.typeName = typeName; - } - - @Override - public IndexResponse indexMessage(String id, String content) { - checkArgument(content); - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("Indexing {}: {}", id, StringUtils.left(content, DEBUG_MAX_LENGTH_CONTENT)); - } - return client.prepareIndex(aliasName.getValue(), typeName.getValue(), id) - .setSource(content) - .get(); - } - - @Override - public Optional<BulkResponse> updateMessages(List<UpdatedRepresentation> updatedDocumentParts) { - try { - Preconditions.checkNotNull(updatedDocumentParts); - BulkRequestBuilder bulkRequestBuilder = client.prepareBulk(); - updatedDocumentParts.forEach(updatedDocumentPart -> bulkRequestBuilder.add( - client.prepareUpdate( - aliasName.getValue(), - typeName.getValue(), - updatedDocumentPart.getId()) - .setDoc(updatedDocumentPart.getUpdatedDocumentPart()))); - return Optional.of(bulkRequestBuilder.get()); - } catch (ValidationException e) { - LOGGER.warn("Error while updating index", e); - return Optional.empty(); - } - } - - @Override - public Optional<BulkResponse> deleteMessages(List<String> ids) { - try { - BulkRequestBuilder bulkRequestBuilder = client.prepareBulk(); - ids.forEach(id -> bulkRequestBuilder.add( - client.prepareDelete( - aliasName.getValue(), - typeName.getValue(), - id))); - return Optional.of(bulkRequestBuilder.get()); - } catch (ValidationException e) { - LOGGER.warn("Error while deleting index", e); - return Optional.empty(); - } - } - - @Override - public void deleteAllMatchingQuery(QueryBuilder queryBuilder) { - deleteByQueryPerformer.perform(queryBuilder); - } - - private void checkArgument(String content) { - Preconditions.checkArgument(content != null, "content should be provided"); - } -} http://git-wip-us.apache.org/repos/asf/james-project/blob/cdf8ce73/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchIndexerTest.java ---------------------------------------------------------------------- diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchIndexerTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchIndexerTest.java deleted file mode 100644 index c811b35..0000000 --- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchIndexerTest.java +++ /dev/null @@ -1,257 +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.james.backends.es; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.elasticsearch.index.query.QueryBuilders.termQuery; - -import java.io.IOException; -import java.util.concurrent.Executors; - -import org.apache.james.backends.es.utils.TestingClientProvider; -import org.elasticsearch.action.search.SearchResponse; -import org.elasticsearch.client.Client; -import org.elasticsearch.index.query.QueryBuilder; -import org.elasticsearch.index.query.QueryBuilders; -import org.elasticsearch.node.Node; -import org.junit.Before; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.RuleChain; -import org.junit.rules.TemporaryFolder; - -import com.google.common.collect.Lists; - -public class ElasticSearchIndexerTest { - - private static final int MINIMUM_BATCH_SIZE = 1; - private static final IndexName INDEX_NAME = new IndexName("index_name"); - private static final AliasName ALIAS_NAME = new AliasName("alias_name"); - private static final TypeName TYPE_NAME = new TypeName("type_name"); - private TemporaryFolder temporaryFolder = new TemporaryFolder(); - private EmbeddedElasticSearch embeddedElasticSearch = new EmbeddedElasticSearch(temporaryFolder, INDEX_NAME); - - @Rule - public RuleChain ruleChain = RuleChain.outerRule(temporaryFolder).around(embeddedElasticSearch); - - private Node node; - private ElasticSearchIndexer testee; - - @Before - public void setup() throws IOException { - node = embeddedElasticSearch.getNode(); - TestingClientProvider clientProvider = new TestingClientProvider(node); - new IndexCreationFactory() - .useIndex(INDEX_NAME) - .addAlias(ALIAS_NAME) - .createIndexAndAliases(clientProvider.get()); - DeleteByQueryPerformer deleteByQueryPerformer = new DeleteByQueryPerformer(clientProvider.get(), - Executors.newSingleThreadExecutor(), - MINIMUM_BATCH_SIZE, - ALIAS_NAME, - TYPE_NAME) { - @Override - public void perform(QueryBuilder queryBuilder) { - doDeleteByQuery(queryBuilder); - } - }; - testee = new ElasticSearchMailboxIndexer(clientProvider.get(), deleteByQueryPerformer, ALIAS_NAME, TYPE_NAME); - } - - @Test - public void indexMessageShouldWork() throws Exception { - String messageId = "1"; - String content = "{\"message\": \"trying out Elasticsearch\"}"; - - testee.indexMessage(messageId, content); - embeddedElasticSearch.awaitForElasticSearch(); - - try (Client client = node.client()) { - SearchResponse searchResponse = client.prepareSearch(INDEX_NAME.getValue()) - .setTypes(TYPE_NAME.getValue()) - .setQuery(QueryBuilders.matchQuery("message", "trying")) - .get(); - assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1); - } - } - - @Test(expected = IllegalArgumentException.class) - public void indexMessageShouldThrowWhenJsonIsNull() throws InterruptedException { - testee.indexMessage("1", null); - } - - @Test - public void updateMessages() throws Exception { - String messageId = "1"; - String content = "{\"message\": \"trying out Elasticsearch\",\"field\":\"Should be unchanged\"}"; - - testee.indexMessage(messageId, content); - embeddedElasticSearch.awaitForElasticSearch(); - - testee.updateMessages(Lists.newArrayList(new UpdatedRepresentation(messageId, "{\"message\": \"mastering out Elasticsearch\"}"))); - embeddedElasticSearch.awaitForElasticSearch(); - - try (Client client = node.client()) { - SearchResponse searchResponse = client.prepareSearch(INDEX_NAME.getValue()) - .setTypes(TYPE_NAME.getValue()) - .setQuery(QueryBuilders.matchQuery("message", "mastering")) - .get(); - assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1); - } - - try (Client client = node.client()) { - SearchResponse searchResponse = client.prepareSearch(INDEX_NAME.getValue()) - .setTypes(TYPE_NAME.getValue()) - .setQuery(QueryBuilders.matchQuery("field", "unchanged")) - .get(); - assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1); - } - } - - @Test(expected = IllegalArgumentException.class) - public void updateMessageShouldThrowWhenJsonIsNull() throws InterruptedException { - testee.updateMessages(Lists.newArrayList(new UpdatedRepresentation("1", null))); - } - - @Test(expected = IllegalArgumentException.class) - public void updateMessageShouldThrowWhenIdIsNull() throws InterruptedException { - testee.updateMessages(Lists.newArrayList(new UpdatedRepresentation(null, "{\"message\": \"mastering out Elasticsearch\"}"))); - } - - @Test(expected = IllegalArgumentException.class) - public void updateMessageShouldThrowWhenJsonIsEmpty() throws InterruptedException { - testee.updateMessages(Lists.newArrayList(new UpdatedRepresentation("1", ""))); - } - - @Test(expected = IllegalArgumentException.class) - public void updateMessageShouldThrowWhenIdIsEmpty() throws InterruptedException { - testee.updateMessages(Lists.newArrayList(new UpdatedRepresentation("", "{\"message\": \"mastering out Elasticsearch\"}"))); - } - - @Test - public void deleteByQueryShouldWorkOnSingleMessage() throws Exception { - String messageId = "1:2"; - String content = "{\"message\": \"trying out Elasticsearch\", \"property\":\"1\"}"; - - testee.indexMessage(messageId, content); - embeddedElasticSearch.awaitForElasticSearch(); - - testee.deleteAllMatchingQuery(termQuery("property", "1")); - embeddedElasticSearch.awaitForElasticSearch(); - - try (Client client = node.client()) { - SearchResponse searchResponse = client.prepareSearch(INDEX_NAME.getValue()) - .setTypes(TYPE_NAME.getValue()) - .setQuery(QueryBuilders.matchAllQuery()) - .get(); - assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(0); - } - } - - @Test - public void deleteByQueryShouldWorkWhenMultipleMessages() throws Exception { - String messageId = "1:1"; - String content = "{\"message\": \"trying out Elasticsearch\", \"property\":\"1\"}"; - - testee.indexMessage(messageId, content); - - String messageId2 = "1:2"; - String content2 = "{\"message\": \"trying out Elasticsearch 2\", \"property\":\"1\"}"; - - testee.indexMessage(messageId2, content2); - - String messageId3 = "2:3"; - String content3 = "{\"message\": \"trying out Elasticsearch 3\", \"property\":\"2\"}"; - - testee.indexMessage(messageId3, content3); - embeddedElasticSearch.awaitForElasticSearch(); - - testee.deleteAllMatchingQuery(termQuery("property", "1")); - embeddedElasticSearch.awaitForElasticSearch(); - - try (Client client = node.client()) { - SearchResponse searchResponse = client.prepareSearch(INDEX_NAME.getValue()) - .setTypes(TYPE_NAME.getValue()) - .setQuery(QueryBuilders.matchAllQuery()) - .get(); - assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1); - } - } - - @Test - public void deleteMessage() throws Exception { - String messageId = "1:2"; - String content = "{\"message\": \"trying out Elasticsearch\"}"; - - testee.indexMessage(messageId, content); - embeddedElasticSearch.awaitForElasticSearch(); - - testee.deleteMessages(Lists.newArrayList(messageId)); - embeddedElasticSearch.awaitForElasticSearch(); - - try (Client client = node.client()) { - SearchResponse searchResponse = client.prepareSearch(INDEX_NAME.getValue()) - .setTypes(TYPE_NAME.getValue()) - .setQuery(QueryBuilders.matchAllQuery()) - .get(); - assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(0); - } - } - - @Test - public void deleteShouldWorkWhenMultipleMessages() throws Exception { - String messageId = "1:1"; - String content = "{\"message\": \"trying out Elasticsearch\", \"mailboxId\":\"1\"}"; - - testee.indexMessage(messageId, content); - - String messageId2 = "1:2"; - String content2 = "{\"message\": \"trying out Elasticsearch 2\", \"mailboxId\":\"1\"}"; - - testee.indexMessage(messageId2, content2); - - String messageId3 = "2:3"; - String content3 = "{\"message\": \"trying out Elasticsearch 3\", \"mailboxId\":\"2\"}"; - - testee.indexMessage(messageId3, content3); - embeddedElasticSearch.awaitForElasticSearch(); - - testee.deleteMessages(Lists.newArrayList(messageId, messageId3)); - embeddedElasticSearch.awaitForElasticSearch(); - - try (Client client = node.client()) { - SearchResponse searchResponse = client.prepareSearch(INDEX_NAME.getValue()) - .setTypes(TYPE_NAME.getValue()) - .setQuery(QueryBuilders.matchAllQuery()) - .get(); - assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1); - } - } - - @Test - public void updateMessagesShouldNotThrowWhenEmptyList() throws Exception { - testee.updateMessages(Lists.newArrayList()); - } - - @Test - public void deleteMessagesShouldNotThrowWhenEmptyList() throws Exception { - testee.deleteMessages(Lists.newArrayList()); - } -} http://git-wip-us.apache.org/repos/asf/james-project/blob/cdf8ce73/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/ElasticSearchMailboxIndexer.java ---------------------------------------------------------------------- diff --git a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/ElasticSearchMailboxIndexer.java b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/ElasticSearchMailboxIndexer.java new file mode 100644 index 0000000..f240baa --- /dev/null +++ b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/ElasticSearchMailboxIndexer.java @@ -0,0 +1,118 @@ +/**************************************************************** + * 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.james.mailbox.elasticsearch; + +import java.util.List; +import java.util.Optional; + +import javax.inject.Inject; +import javax.inject.Named; + +import org.apache.commons.lang3.StringUtils; +import org.apache.james.backends.es.AliasName; +import org.apache.james.backends.es.DeleteByQueryPerformer; +import org.apache.james.backends.es.ElasticSearchConstants; +import org.apache.james.backends.es.ElasticSearchIndexer; +import org.apache.james.backends.es.TypeName; +import org.apache.james.backends.es.UpdatedRepresentation; +import org.elasticsearch.action.bulk.BulkRequestBuilder; +import org.elasticsearch.action.bulk.BulkResponse; +import org.elasticsearch.action.index.IndexResponse; +import org.elasticsearch.client.Client; +import org.elasticsearch.common.ValidationException; +import org.elasticsearch.index.query.QueryBuilder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.base.Preconditions; + +public class ElasticSearchMailboxIndexer implements ElasticSearchIndexer { + private static int DEBUG_MAX_LENGTH_CONTENT = 1000; + + private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchMailboxIndexer.class); + + private final Client client; + private final DeleteByQueryPerformer deleteByQueryPerformer; + private final AliasName aliasName; + private final TypeName typeName; + + @Inject + public ElasticSearchMailboxIndexer(Client client, DeleteByQueryPerformer deleteByQueryPerformer, + @Named(ElasticSearchConstants.WRITE_ALIAS) AliasName aliasName, + TypeName typeName) { + this.client = client; + this.deleteByQueryPerformer = deleteByQueryPerformer; + this.aliasName = aliasName; + this.typeName = typeName; + } + + @Override + public IndexResponse indexMessage(String id, String content) { + checkArgument(content); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Indexing {}: {}", id, StringUtils.left(content, DEBUG_MAX_LENGTH_CONTENT)); + } + return client.prepareIndex(aliasName.getValue(), typeName.getValue(), id) + .setSource(content) + .get(); + } + + @Override + public Optional<BulkResponse> updateMessages(List<UpdatedRepresentation> updatedDocumentParts) { + try { + Preconditions.checkNotNull(updatedDocumentParts); + BulkRequestBuilder bulkRequestBuilder = client.prepareBulk(); + updatedDocumentParts.forEach(updatedDocumentPart -> bulkRequestBuilder.add( + client.prepareUpdate( + aliasName.getValue(), + typeName.getValue(), + updatedDocumentPart.getId()) + .setDoc(updatedDocumentPart.getUpdatedDocumentPart()))); + return Optional.of(bulkRequestBuilder.get()); + } catch (ValidationException e) { + LOGGER.warn("Error while updating index", e); + return Optional.empty(); + } + } + + @Override + public Optional<BulkResponse> deleteMessages(List<String> ids) { + try { + BulkRequestBuilder bulkRequestBuilder = client.prepareBulk(); + ids.forEach(id -> bulkRequestBuilder.add( + client.prepareDelete( + aliasName.getValue(), + typeName.getValue(), + id))); + return Optional.of(bulkRequestBuilder.get()); + } catch (ValidationException e) { + LOGGER.warn("Error while deleting index", e); + return Optional.empty(); + } + } + + @Override + public void deleteAllMatchingQuery(QueryBuilder queryBuilder) { + deleteByQueryPerformer.perform(queryBuilder); + } + + private void checkArgument(String content) { + Preconditions.checkArgument(content != null, "content should be provided"); + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/cdf8ce73/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIntegrationTest.java ---------------------------------------------------------------------- diff --git a/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIntegrationTest.java b/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIntegrationTest.java index dcb8308..7cc78f2 100644 --- a/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIntegrationTest.java +++ b/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIntegrationTest.java @@ -26,7 +26,6 @@ import java.time.ZoneId; import java.util.concurrent.Executors; import org.apache.james.backends.es.DeleteByQueryPerformer; -import org.apache.james.backends.es.ElasticSearchMailboxIndexer; import org.apache.james.backends.es.EmbeddedElasticSearch; import org.apache.james.backends.es.IndexCreationFactory; import org.apache.james.backends.es.NodeMappingFactory; http://git-wip-us.apache.org/repos/asf/james-project/blob/cdf8ce73/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchMailboxIndexerTest.java ---------------------------------------------------------------------- diff --git a/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchMailboxIndexerTest.java b/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchMailboxIndexerTest.java new file mode 100644 index 0000000..f33ac57 --- /dev/null +++ b/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchMailboxIndexerTest.java @@ -0,0 +1,266 @@ +/**************************************************************** + * 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.james.mailbox.elasticsearch; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.elasticsearch.index.query.QueryBuilders.termQuery; + +import java.io.IOException; +import java.util.concurrent.Executors; + +import org.apache.james.backends.es.AliasName; +import org.apache.james.backends.es.DeleteByQueryPerformer; +import org.apache.james.backends.es.ElasticSearchIndexer; +import org.apache.james.backends.es.EmbeddedElasticSearch; +import org.apache.james.backends.es.IndexCreationFactory; +import org.apache.james.backends.es.IndexName; +import org.apache.james.backends.es.TypeName; +import org.apache.james.backends.es.UpdatedRepresentation; +import org.apache.james.backends.es.utils.TestingClientProvider; +import org.apache.james.mailbox.elasticsearch.ElasticSearchMailboxIndexer; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.client.Client; +import org.elasticsearch.index.query.QueryBuilder; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.node.Node; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.RuleChain; +import org.junit.rules.TemporaryFolder; + +import com.google.common.collect.Lists; + +public class ElasticSearchMailboxIndexerTest { + + private static final int MINIMUM_BATCH_SIZE = 1; + private static final IndexName INDEX_NAME = new IndexName("index_name"); + private static final AliasName ALIAS_NAME = new AliasName("alias_name"); + private static final TypeName TYPE_NAME = new TypeName("type_name"); + private TemporaryFolder temporaryFolder = new TemporaryFolder(); + private EmbeddedElasticSearch embeddedElasticSearch = new EmbeddedElasticSearch(temporaryFolder, INDEX_NAME); + + @Rule + public RuleChain ruleChain = RuleChain.outerRule(temporaryFolder).around(embeddedElasticSearch); + + private Node node; + private ElasticSearchIndexer testee; + + @Before + public void setup() throws IOException { + node = embeddedElasticSearch.getNode(); + TestingClientProvider clientProvider = new TestingClientProvider(node); + new IndexCreationFactory() + .useIndex(INDEX_NAME) + .addAlias(ALIAS_NAME) + .createIndexAndAliases(clientProvider.get()); + DeleteByQueryPerformer deleteByQueryPerformer = new DeleteByQueryPerformer(clientProvider.get(), + Executors.newSingleThreadExecutor(), + MINIMUM_BATCH_SIZE, + ALIAS_NAME, + TYPE_NAME) { + @Override + public void perform(QueryBuilder queryBuilder) { + doDeleteByQuery(queryBuilder); + } + }; + testee = new ElasticSearchMailboxIndexer(clientProvider.get(), deleteByQueryPerformer, ALIAS_NAME, TYPE_NAME); + } + + @Test + public void indexMessageShouldWork() throws Exception { + String messageId = "1"; + String content = "{\"message\": \"trying out Elasticsearch\"}"; + + testee.indexMessage(messageId, content); + embeddedElasticSearch.awaitForElasticSearch(); + + try (Client client = node.client()) { + SearchResponse searchResponse = client.prepareSearch(INDEX_NAME.getValue()) + .setTypes(TYPE_NAME.getValue()) + .setQuery(QueryBuilders.matchQuery("message", "trying")) + .get(); + assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1); + } + } + + @Test(expected = IllegalArgumentException.class) + public void indexMessageShouldThrowWhenJsonIsNull() throws InterruptedException { + testee.indexMessage("1", null); + } + + @Test + public void updateMessages() throws Exception { + String messageId = "1"; + String content = "{\"message\": \"trying out Elasticsearch\",\"field\":\"Should be unchanged\"}"; + + testee.indexMessage(messageId, content); + embeddedElasticSearch.awaitForElasticSearch(); + + testee.updateMessages(Lists.newArrayList(new UpdatedRepresentation(messageId, "{\"message\": \"mastering out Elasticsearch\"}"))); + embeddedElasticSearch.awaitForElasticSearch(); + + try (Client client = node.client()) { + SearchResponse searchResponse = client.prepareSearch(INDEX_NAME.getValue()) + .setTypes(TYPE_NAME.getValue()) + .setQuery(QueryBuilders.matchQuery("message", "mastering")) + .get(); + assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1); + } + + try (Client client = node.client()) { + SearchResponse searchResponse = client.prepareSearch(INDEX_NAME.getValue()) + .setTypes(TYPE_NAME.getValue()) + .setQuery(QueryBuilders.matchQuery("field", "unchanged")) + .get(); + assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1); + } + } + + @Test(expected = IllegalArgumentException.class) + public void updateMessageShouldThrowWhenJsonIsNull() throws InterruptedException { + testee.updateMessages(Lists.newArrayList(new UpdatedRepresentation("1", null))); + } + + @Test(expected = IllegalArgumentException.class) + public void updateMessageShouldThrowWhenIdIsNull() throws InterruptedException { + testee.updateMessages(Lists.newArrayList(new UpdatedRepresentation(null, "{\"message\": \"mastering out Elasticsearch\"}"))); + } + + @Test(expected = IllegalArgumentException.class) + public void updateMessageShouldThrowWhenJsonIsEmpty() throws InterruptedException { + testee.updateMessages(Lists.newArrayList(new UpdatedRepresentation("1", ""))); + } + + @Test(expected = IllegalArgumentException.class) + public void updateMessageShouldThrowWhenIdIsEmpty() throws InterruptedException { + testee.updateMessages(Lists.newArrayList(new UpdatedRepresentation("", "{\"message\": \"mastering out Elasticsearch\"}"))); + } + + @Test + public void deleteByQueryShouldWorkOnSingleMessage() throws Exception { + String messageId = "1:2"; + String content = "{\"message\": \"trying out Elasticsearch\", \"property\":\"1\"}"; + + testee.indexMessage(messageId, content); + embeddedElasticSearch.awaitForElasticSearch(); + + testee.deleteAllMatchingQuery(termQuery("property", "1")); + embeddedElasticSearch.awaitForElasticSearch(); + + try (Client client = node.client()) { + SearchResponse searchResponse = client.prepareSearch(INDEX_NAME.getValue()) + .setTypes(TYPE_NAME.getValue()) + .setQuery(QueryBuilders.matchAllQuery()) + .get(); + assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(0); + } + } + + @Test + public void deleteByQueryShouldWorkWhenMultipleMessages() throws Exception { + String messageId = "1:1"; + String content = "{\"message\": \"trying out Elasticsearch\", \"property\":\"1\"}"; + + testee.indexMessage(messageId, content); + + String messageId2 = "1:2"; + String content2 = "{\"message\": \"trying out Elasticsearch 2\", \"property\":\"1\"}"; + + testee.indexMessage(messageId2, content2); + + String messageId3 = "2:3"; + String content3 = "{\"message\": \"trying out Elasticsearch 3\", \"property\":\"2\"}"; + + testee.indexMessage(messageId3, content3); + embeddedElasticSearch.awaitForElasticSearch(); + + testee.deleteAllMatchingQuery(termQuery("property", "1")); + embeddedElasticSearch.awaitForElasticSearch(); + + try (Client client = node.client()) { + SearchResponse searchResponse = client.prepareSearch(INDEX_NAME.getValue()) + .setTypes(TYPE_NAME.getValue()) + .setQuery(QueryBuilders.matchAllQuery()) + .get(); + assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1); + } + } + + @Test + public void deleteMessage() throws Exception { + String messageId = "1:2"; + String content = "{\"message\": \"trying out Elasticsearch\"}"; + + testee.indexMessage(messageId, content); + embeddedElasticSearch.awaitForElasticSearch(); + + testee.deleteMessages(Lists.newArrayList(messageId)); + embeddedElasticSearch.awaitForElasticSearch(); + + try (Client client = node.client()) { + SearchResponse searchResponse = client.prepareSearch(INDEX_NAME.getValue()) + .setTypes(TYPE_NAME.getValue()) + .setQuery(QueryBuilders.matchAllQuery()) + .get(); + assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(0); + } + } + + @Test + public void deleteShouldWorkWhenMultipleMessages() throws Exception { + String messageId = "1:1"; + String content = "{\"message\": \"trying out Elasticsearch\", \"mailboxId\":\"1\"}"; + + testee.indexMessage(messageId, content); + + String messageId2 = "1:2"; + String content2 = "{\"message\": \"trying out Elasticsearch 2\", \"mailboxId\":\"1\"}"; + + testee.indexMessage(messageId2, content2); + + String messageId3 = "2:3"; + String content3 = "{\"message\": \"trying out Elasticsearch 3\", \"mailboxId\":\"2\"}"; + + testee.indexMessage(messageId3, content3); + embeddedElasticSearch.awaitForElasticSearch(); + + testee.deleteMessages(Lists.newArrayList(messageId, messageId3)); + embeddedElasticSearch.awaitForElasticSearch(); + + try (Client client = node.client()) { + SearchResponse searchResponse = client.prepareSearch(INDEX_NAME.getValue()) + .setTypes(TYPE_NAME.getValue()) + .setQuery(QueryBuilders.matchAllQuery()) + .get(); + assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1); + } + } + + @Test + public void updateMessagesShouldNotThrowWhenEmptyList() throws Exception { + testee.updateMessages(Lists.newArrayList()); + } + + @Test + public void deleteMessagesShouldNotThrowWhenEmptyList() throws Exception { + testee.deleteMessages(Lists.newArrayList()); + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/cdf8ce73/mpt/impl/imap-mailbox/elasticsearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/ElasticSearchHostSystem.java ---------------------------------------------------------------------- diff --git a/mpt/impl/imap-mailbox/elasticsearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/ElasticSearchHostSystem.java b/mpt/impl/imap-mailbox/elasticsearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/ElasticSearchHostSystem.java index c84e2ee..d98f5b8 100644 --- a/mpt/impl/imap-mailbox/elasticsearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/ElasticSearchHostSystem.java +++ b/mpt/impl/imap-mailbox/elasticsearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/ElasticSearchHostSystem.java @@ -27,7 +27,6 @@ import java.util.concurrent.Executors; import org.apache.commons.io.FileUtils; import org.apache.commons.lang.NotImplementedException; import org.apache.james.backends.es.DeleteByQueryPerformer; -import org.apache.james.backends.es.ElasticSearchMailboxIndexer; import org.apache.james.backends.es.EmbeddedElasticSearch; import org.apache.james.backends.es.IndexCreationFactory; import org.apache.james.backends.es.NodeMappingFactory; @@ -38,6 +37,7 @@ import org.apache.james.imap.main.DefaultImapDecoderFactory; import org.apache.james.imap.processor.main.DefaultImapProcessorFactory; import org.apache.james.mailbox.MailboxManager; import org.apache.james.mailbox.acl.SimpleGroupMembershipResolver; +import org.apache.james.mailbox.elasticsearch.ElasticSearchMailboxIndexer; import org.apache.james.mailbox.elasticsearch.IndexAttachments; import org.apache.james.mailbox.elasticsearch.MailboxElasticSearchConstants; import org.apache.james.mailbox.elasticsearch.MailboxMappingFactory; http://git-wip-us.apache.org/repos/asf/james-project/blob/cdf8ce73/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchMailboxModule.java ---------------------------------------------------------------------- diff --git a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchMailboxModule.java b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchMailboxModule.java index fa20148..2544c91 100644 --- a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchMailboxModule.java +++ b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchMailboxModule.java @@ -32,11 +32,11 @@ import org.apache.james.backends.es.AliasName; import org.apache.james.backends.es.ClientProviderImpl; import org.apache.james.backends.es.ElasticSearchConstants; import org.apache.james.backends.es.ElasticSearchIndexer; -import org.apache.james.backends.es.ElasticSearchMailboxIndexer; import org.apache.james.backends.es.IndexCreationFactory; import org.apache.james.backends.es.IndexName; import org.apache.james.backends.es.NodeMappingFactory; import org.apache.james.backends.es.TypeName; +import org.apache.james.mailbox.elasticsearch.ElasticSearchMailboxIndexer; import org.apache.james.mailbox.elasticsearch.IndexAttachments; import org.apache.james.mailbox.elasticsearch.MailboxElasticSearchConstants; import org.apache.james.mailbox.elasticsearch.MailboxMappingFactory; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
