JAMES-2398 Modify injections and configuration to allow several indexes, with different mappings, pointed by different aliases.
Project: http://git-wip-us.apache.org/repos/asf/james-project/repo Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/3ae482a7 Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/3ae482a7 Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/3ae482a7 Branch: refs/heads/master Commit: 3ae482a711c5a6b5310713553adfbb6102298ff9 Parents: 6013f91 Author: benwa <[email protected]> Authored: Thu May 17 16:17:58 2018 +0700 Committer: Matthieu Baechler <[email protected]> Committed: Mon May 21 11:37:52 2018 +0200 ---------------------------------------------------------------------- .../backends/es/DeleteByQueryPerformer.java | 9 +- .../backends/es/ElasticSearchConstants.java | 27 -- .../james/backends/es/ElasticSearchIndexer.java | 115 +++++++++ .../org/apache/james/backends/es/Indexer.java | 104 -------- .../james/backends/es/IndexerSupplier.java | 24 -- .../backends/es/ElasticSearchIndexerTest.java | 245 ++++++++++++++++++ .../apache/james/backends/es/IndexerTest.java | 256 ------------------- .../MailboxElasticSearchConstants.java | 5 + .../elasticsearch/MailboxIndexerSupplier.java | 46 ---- ...lasticSearchListeningMessageSearchIndex.java | 21 +- .../search/ElasticSearchSearcher.java | 14 +- .../ElasticSearchIntegrationTest.java | 13 +- ...icSearchListeningMessageSearchIndexTest.java | 30 +-- .../host/ElasticSearchHostSystem.java | 11 +- .../mailbox/ElasticSearchConfiguration.java | 80 ++++-- .../mailbox/ElasticSearchMailboxModule.java | 103 ++++---- .../mailbox/ElasticSearchConfigurationTest.java | 106 +++++++- src/site/xdoc/server/config-elasticsearch.xml | 17 +- 18 files changed, 621 insertions(+), 605 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/james-project/blob/3ae482a7/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/DeleteByQueryPerformer.java ---------------------------------------------------------------------- diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/DeleteByQueryPerformer.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/DeleteByQueryPerformer.java index 6c8c2e3..a267b3a 100644 --- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/DeleteByQueryPerformer.java +++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/DeleteByQueryPerformer.java @@ -21,9 +21,6 @@ package org.apache.james.backends.es; import java.util.concurrent.ExecutorService; -import javax.inject.Inject; -import javax.inject.Named; - import org.apache.james.backends.es.search.ScrollIterable; import org.elasticsearch.action.ListenableActionFuture; import org.elasticsearch.action.bulk.BulkRequestBuilder; @@ -46,13 +43,13 @@ public class DeleteByQueryPerformer { private final AliasName aliasName; private final TypeName typeName; - @Inject - public DeleteByQueryPerformer(Client client, @Named("AsyncExecutor") ExecutorService executor, @Named(ElasticSearchConstants.WRITE_ALIAS) AliasName aliasName, TypeName typeName) { + public DeleteByQueryPerformer(Client client, ExecutorService executor, + AliasName aliasName, TypeName typeName) { this(client, executor, DEFAULT_BATCH_SIZE, aliasName, typeName); } @VisibleForTesting - public DeleteByQueryPerformer(Client client, @Named("AsyncExecutor") ExecutorService executor, int batchSize, AliasName aliasName, TypeName typeName) { + public DeleteByQueryPerformer(Client client, ExecutorService executor, int batchSize, AliasName aliasName, TypeName typeName) { this.client = client; this.executor = executor; this.batchSize = batchSize; http://git-wip-us.apache.org/repos/asf/james-project/blob/3ae482a7/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchConstants.java ---------------------------------------------------------------------- diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchConstants.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchConstants.java deleted file mode 100644 index 7429a3b..0000000 --- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchConstants.java +++ /dev/null @@ -1,27 +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; - -public interface ElasticSearchConstants { - String WRITE_ALIAS = "injectWriteAlias"; - String READ_ALIAS = "injectReadAlias"; - - String MAILBOX_INDEX = "mailboxIndex"; -} http://git-wip-us.apache.org/repos/asf/james-project/blob/3ae482a7/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchIndexer.java ---------------------------------------------------------------------- diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchIndexer.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchIndexer.java new file mode 100644 index 0000000..ec3f161 --- /dev/null +++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchIndexer.java @@ -0,0 +1,115 @@ +/**************************************************************** + * 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 java.util.concurrent.ExecutorService; + +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.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; + +public class ElasticSearchIndexer { + private static int DEBUG_MAX_LENGTH_CONTENT = 1000; + private static final int DEFAULT_BATCH_SIZE = 100; + + private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchIndexer.class); + + private final Client client; + private final DeleteByQueryPerformer deleteByQueryPerformer; + private final AliasName aliasName; + private final TypeName typeName; + + public ElasticSearchIndexer(Client client, ExecutorService executor, + AliasName aliasName, + TypeName typeName) { + this(client, executor, aliasName, typeName, DEFAULT_BATCH_SIZE); + } + + @VisibleForTesting + public ElasticSearchIndexer(Client client, ExecutorService executor, + AliasName aliasName, + TypeName typeName, + int batchSize) { + this.client = client; + this.deleteByQueryPerformer = new DeleteByQueryPerformer(client, executor, batchSize, aliasName, typeName); + this.aliasName = aliasName; + this.typeName = typeName; + } + + 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(); + } + + 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(); + } + } + + 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(); + } + } + + 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/3ae482a7/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/Indexer.java ---------------------------------------------------------------------- diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/Indexer.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/Indexer.java deleted file mode 100644 index 9a3c83b..0000000 --- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/Indexer.java +++ /dev/null @@ -1,104 +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 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 Indexer { - private static int DEBUG_MAX_LENGTH_CONTENT = 1000; - - private static final Logger LOGGER = LoggerFactory.getLogger(Indexer.class); - - private final Client client; - private final DeleteByQueryPerformer deleteByQueryPerformer; - private final AliasName aliasName; - private final TypeName typeName; - - public Indexer(Client client, DeleteByQueryPerformer deleteByQueryPerformer, - AliasName aliasName, - TypeName typeName) { - this.client = client; - this.deleteByQueryPerformer = deleteByQueryPerformer; - this.aliasName = aliasName; - this.typeName = typeName; - } - - 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(); - } - - 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(); - } - } - - 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(); - } - } - - 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/3ae482a7/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/IndexerSupplier.java ---------------------------------------------------------------------- diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/IndexerSupplier.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/IndexerSupplier.java deleted file mode 100644 index 8a3edea..0000000 --- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/IndexerSupplier.java +++ /dev/null @@ -1,24 +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; - -public interface IndexerSupplier { - Indexer getIndexer(); -} http://git-wip-us.apache.org/repos/asf/james-project/blob/3ae482a7/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 new file mode 100644 index 0000000..29f6226 --- /dev/null +++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchIndexerTest.java @@ -0,0 +1,245 @@ +/**************************************************************** + * 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.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.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() { + node = embeddedElasticSearch.getNode(); + TestingClientProvider clientProvider = new TestingClientProvider(node); + new IndexCreationFactory() + .useIndex(INDEX_NAME) + .addAlias(ALIAS_NAME) + .createIndexAndAliases(clientProvider.get()); + testee = new ElasticSearchIndexer(clientProvider.get(), Executors.newSingleThreadExecutor(), ALIAS_NAME, TYPE_NAME, MINIMUM_BATCH_SIZE); + } + + @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/3ae482a7/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/IndexerTest.java ---------------------------------------------------------------------- diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/IndexerTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/IndexerTest.java deleted file mode 100644 index 2abbdda..0000000 --- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/IndexerTest.java +++ /dev/null @@ -1,256 +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.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 IndexerTest { - - 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 Indexer testee; - - @Before - public void setup() { - 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 Indexer(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/3ae482a7/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/MailboxElasticSearchConstants.java ---------------------------------------------------------------------- diff --git a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/MailboxElasticSearchConstants.java b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/MailboxElasticSearchConstants.java index 1017933..c347738 100644 --- a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/MailboxElasticSearchConstants.java +++ b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/MailboxElasticSearchConstants.java @@ -24,6 +24,11 @@ import org.apache.james.backends.es.IndexName; import org.apache.james.backends.es.TypeName; public interface MailboxElasticSearchConstants { + + interface InjectionNames { + String MAILBOX = "mailbox"; + } + AliasName DEFAULT_MAILBOX_WRITE_ALIAS = new AliasName("mailboxWriteAlias"); AliasName DEFAULT_MAILBOX_READ_ALIAS = new AliasName("mailboxReadAlias"); IndexName DEFAULT_MAILBOX_INDEX = new IndexName("mailbox_v1"); http://git-wip-us.apache.org/repos/asf/james-project/blob/3ae482a7/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/MailboxIndexerSupplier.java ---------------------------------------------------------------------- diff --git a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/MailboxIndexerSupplier.java b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/MailboxIndexerSupplier.java deleted file mode 100644 index e8a2143..0000000 --- a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/MailboxIndexerSupplier.java +++ /dev/null @@ -1,46 +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.mailbox.elasticsearch; - -import javax.inject.Inject; -import javax.inject.Named; - -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.Indexer; -import org.apache.james.backends.es.IndexerSupplier; -import org.apache.james.backends.es.TypeName; -import org.elasticsearch.client.Client; - -public class MailboxIndexerSupplier implements IndexerSupplier { - private final Indexer indexer; - - @Inject - public MailboxIndexerSupplier(Client client, DeleteByQueryPerformer deleteByQueryPerformer, - @Named(ElasticSearchConstants.WRITE_ALIAS) AliasName aliasName, - TypeName typeName) { - this.indexer = new Indexer(client, deleteByQueryPerformer, aliasName, typeName); - } - - @Override - public Indexer getIndexer() { - return indexer; - } -} http://git-wip-us.apache.org/repos/asf/james-project/blob/3ae482a7/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java ---------------------------------------------------------------------- diff --git a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java index 68d7af9..a367e59 100644 --- a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java +++ b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java @@ -30,14 +30,13 @@ import java.util.stream.Collectors; import javax.inject.Inject; import javax.inject.Named; -import org.apache.james.backends.es.ElasticSearchConstants; -import org.apache.james.backends.es.Indexer; -import org.apache.james.backends.es.IndexerSupplier; +import org.apache.james.backends.es.ElasticSearchIndexer; import org.apache.james.backends.es.UpdatedRepresentation; import org.apache.james.mailbox.MailboxManager.MessageCapabilities; import org.apache.james.mailbox.MailboxManager.SearchCapabilities; import org.apache.james.mailbox.MailboxSession; import org.apache.james.mailbox.MessageUid; +import org.apache.james.mailbox.elasticsearch.MailboxElasticSearchConstants; import org.apache.james.mailbox.elasticsearch.json.JsonMessageConstants; import org.apache.james.mailbox.elasticsearch.json.MessageToElasticSearchJson; import org.apache.james.mailbox.elasticsearch.search.ElasticSearchSearcher; @@ -63,16 +62,16 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchListeningMessageSearchIndex.class); private static final String ID_SEPARATOR = ":"; - private final Indexer indexer; + private final ElasticSearchIndexer elasticSearchIndexer; private final ElasticSearchSearcher searcher; private final MessageToElasticSearchJson messageToElasticSearchJson; @Inject public ElasticSearchListeningMessageSearchIndex(MessageMapperFactory factory, - @Named(ElasticSearchConstants.MAILBOX_INDEX) IndexerSupplier indexer, + @Named(MailboxElasticSearchConstants.InjectionNames.MAILBOX) ElasticSearchIndexer indexer, ElasticSearchSearcher searcher, MessageToElasticSearchJson messageToElasticSearchJson) { super(factory); - this.indexer = indexer.getIndexer(); + this.elasticSearchIndexer = indexer; this.messageToElasticSearchJson = messageToElasticSearchJson; this.searcher = searcher; } @@ -129,7 +128,7 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe mailbox.getMailboxId(), session.getUser().getUserName(), message.getUid()); - indexer.indexMessage(indexIdFor(mailbox, message.getUid()), messageToElasticSearchJson.convertToJson(message, ImmutableList.of(session.getUser()))); + elasticSearchIndexer.indexMessage(indexIdFor(mailbox, message.getUid()), messageToElasticSearchJson.convertToJson(message, ImmutableList.of(session.getUser()))); } catch (Exception e) { try { LOGGER.warn("Indexing mailbox {}-{} of user {} on message {} without attachments ", @@ -138,7 +137,7 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe session.getUser().getUserName(), message.getUid(), e); - indexer.indexMessage(indexIdFor(mailbox, message.getUid()), messageToElasticSearchJson.convertToJsonWithoutAttachment(message, ImmutableList.of(session.getUser()))); + elasticSearchIndexer.indexMessage(indexIdFor(mailbox, message.getUid()), messageToElasticSearchJson.convertToJsonWithoutAttachment(message, ImmutableList.of(session.getUser()))); } catch (JsonProcessingException e1) { LOGGER.error("Error when indexing mailbox {}-{} of user {} on message {} without its attachment", mailbox.getName(), @@ -153,7 +152,7 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe @Override public void delete(MailboxSession session, Mailbox mailbox, List<MessageUid> expungedUids) throws MailboxException { try { - indexer.deleteMessages(expungedUids.stream() + elasticSearchIndexer.deleteMessages(expungedUids.stream() .map(uid -> indexIdFor(mailbox, uid)) .collect(Collectors.toList())); } catch (Exception e) { @@ -166,7 +165,7 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe @Override public void deleteAll(MailboxSession session, Mailbox mailbox) throws MailboxException { try { - indexer.deleteAllMatchingQuery( + elasticSearchIndexer.deleteAllMatchingQuery( termQuery( JsonMessageConstants.MAILBOX_ID, mailbox.getMailboxId().serialize())); @@ -178,7 +177,7 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe @Override public void update(MailboxSession session, Mailbox mailbox, List<UpdatedFlags> updatedFlagsList) throws MailboxException { try { - indexer.updateMessages(updatedFlagsList.stream() + elasticSearchIndexer.updateMessages(updatedFlagsList.stream() .map(updatedFlags -> createUpdatedDocumentPartFromUpdatedFlags(mailbox, updatedFlags)) .collect(Collectors.toList())); } catch (Exception e) { http://git-wip-us.apache.org/repos/asf/james-project/blob/3ae482a7/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/search/ElasticSearchSearcher.java ---------------------------------------------------------------------- diff --git a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/search/ElasticSearchSearcher.java b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/search/ElasticSearchSearcher.java index cab7a53..68b6a32 100644 --- a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/search/ElasticSearchSearcher.java +++ b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/search/ElasticSearchSearcher.java @@ -23,11 +23,7 @@ import java.util.Collection; import java.util.Optional; import java.util.stream.Stream; -import javax.inject.Inject; -import javax.inject.Named; - import org.apache.james.backends.es.AliasName; -import org.apache.james.backends.es.ElasticSearchConstants; import org.apache.james.backends.es.TypeName; import org.apache.james.backends.es.search.ScrollIterable; import org.apache.james.mailbox.MessageUid; @@ -51,9 +47,10 @@ import org.slf4j.LoggerFactory; public class ElasticSearchSearcher { + public static final int DEFAULT_SEARCH_SIZE = 100; + private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchSearcher.class); private static final TimeValue TIMEOUT = new TimeValue(60000); - private static final int DEFAULT_SIZE = 100; private final Client client; private final QueryConverter queryConverter; @@ -63,13 +60,6 @@ public class ElasticSearchSearcher { private final AliasName aliasName; private final TypeName typeName; - @Inject - public ElasticSearchSearcher(Client client, QueryConverter queryConverter, - MailboxId.Factory mailboxIdFactory, MessageId.Factory messageIdFactory, - @Named(ElasticSearchConstants.READ_ALIAS) AliasName aliasName, TypeName typeName) { - this(client, queryConverter, DEFAULT_SIZE, mailboxIdFactory, messageIdFactory, aliasName, typeName); - } - public ElasticSearchSearcher(Client client, QueryConverter queryConverter, int size, MailboxId.Factory mailboxIdFactory, MessageId.Factory messageIdFactory, AliasName aliasName, TypeName typeName) { http://git-wip-us.apache.org/repos/asf/james-project/blob/3ae482a7/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 cc5c679..444422f 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 @@ -25,7 +25,7 @@ import java.nio.charset.StandardCharsets; import java.time.ZoneId; import java.util.concurrent.Executors; -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.NodeMappingFactory; @@ -109,14 +109,11 @@ public class ElasticSearchIntegrationTest extends AbstractMessageSearchIndexTest ElasticSearchListeningMessageSearchIndex elasticSearchListeningMessageSearchIndex = new ElasticSearchListeningMessageSearchIndex( storeMailboxManager.getMapperFactory(), - new MailboxIndexerSupplier(client, - new DeleteByQueryPerformer(client, - Executors.newSingleThreadExecutor(), - BATCH_SIZE, - MailboxElasticSearchConstants.DEFAULT_MAILBOX_WRITE_ALIAS, - MailboxElasticSearchConstants.MESSAGE_TYPE), + new ElasticSearchIndexer(client, + Executors.newSingleThreadExecutor(), MailboxElasticSearchConstants.DEFAULT_MAILBOX_WRITE_ALIAS, - MailboxElasticSearchConstants.MESSAGE_TYPE), + MailboxElasticSearchConstants.MESSAGE_TYPE, + BATCH_SIZE), new ElasticSearchSearcher(client, new QueryConverter(new CriterionConverter()), SEARCH_SIZE, new InMemoryId.Factory(), storeMailboxManager.getMessageIdFactory(), MailboxElasticSearchConstants.DEFAULT_MAILBOX_READ_ALIAS, http://git-wip-us.apache.org/repos/asf/james-project/blob/3ae482a7/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndexTest.java ---------------------------------------------------------------------- diff --git a/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndexTest.java b/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndexTest.java index 8a07886..0f1325c 100644 --- a/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndexTest.java +++ b/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndexTest.java @@ -32,7 +32,7 @@ import java.util.Optional; import javax.mail.Flags; -import org.apache.james.backends.es.Indexer; +import org.apache.james.backends.es.ElasticSearchIndexer; import org.apache.james.backends.es.UpdatedRepresentation; import org.apache.james.mailbox.MailboxSession; import org.apache.james.mailbox.MailboxSession.User; @@ -67,7 +67,7 @@ public class ElasticSearchListeningMessageSearchIndexTest { public static final String EXPECTED_JSON_CONTENT = "json content"; public static final String USERNAME = "username"; - private Indexer indexer; + private ElasticSearchIndexer elasticSearchIndexer; private MessageToElasticSearchJson messageToElasticSearchJson; private ElasticSearchListeningMessageSearchIndex testee; private MailboxSession session; @@ -80,9 +80,9 @@ public class ElasticSearchListeningMessageSearchIndexTest { messageToElasticSearchJson = mock(MessageToElasticSearchJson.class); ElasticSearchSearcher elasticSearchSearcher = mock(ElasticSearchSearcher.class); - indexer = mock(Indexer.class); + elasticSearchIndexer = mock(ElasticSearchIndexer.class); - testee = new ElasticSearchListeningMessageSearchIndex(mapperFactory, indexer, elasticSearchSearcher, messageToElasticSearchJson); + testee = new ElasticSearchListeningMessageSearchIndex(mapperFactory, elasticSearchIndexer, elasticSearchSearcher, messageToElasticSearchJson); session = new MockMailboxSession(USERNAME); users = ImmutableList.of(session.getUser()); } @@ -102,7 +102,7 @@ public class ElasticSearchListeningMessageSearchIndexTest { testee.add(session, mailbox, message); //Then - verify(indexer).indexMessage(eq(ELASTIC_SEARCH_ID), eq(EXPECTED_JSON_CONTENT)); + verify(elasticSearchIndexer).indexMessage(eq(ELASTIC_SEARCH_ID), eq(EXPECTED_JSON_CONTENT)); } @SuppressWarnings("unchecked") @@ -125,7 +125,7 @@ public class ElasticSearchListeningMessageSearchIndexTest { testee.add(session, mailbox, message); //Then - verify(indexer).indexMessage(eq(ELASTIC_SEARCH_ID), eq(EXPECTED_JSON_CONTENT)); + verify(elasticSearchIndexer).indexMessage(eq(ELASTIC_SEARCH_ID), eq(EXPECTED_JSON_CONTENT)); } private MailboxMessage mockedMessage(MessageUid messageId) throws IOException { @@ -166,14 +166,14 @@ public class ElasticSearchListeningMessageSearchIndexTest { .thenReturn(MAILBOX_ID); BulkResponse expectedBulkResponse = mock(BulkResponse.class); - when(indexer.deleteMessages(any(List.class))) + when(elasticSearchIndexer.deleteMessages(any(List.class))) .thenReturn(Optional.of(expectedBulkResponse)); //When testee.delete(session, mailbox, Lists.newArrayList(MESSAGE_UID)); //Then - verify(indexer).deleteMessages(eq(Lists.newArrayList(ELASTIC_SEARCH_ID))); + verify(elasticSearchIndexer).deleteMessages(eq(Lists.newArrayList(ELASTIC_SEARCH_ID))); } @Test @@ -189,14 +189,14 @@ public class ElasticSearchListeningMessageSearchIndexTest { .thenReturn(MAILBOX_ID); BulkResponse expectedBulkResponse = mock(BulkResponse.class); - when(indexer.deleteMessages(any(List.class))) + when(elasticSearchIndexer.deleteMessages(any(List.class))) .thenReturn(Optional.of(expectedBulkResponse)); //When testee.delete(session, mailbox, Lists.newArrayList(MESSAGE_UID, messageId2, messageId3, messageId4, messageId5)); //Then - verify(indexer).deleteMessages(eq(Lists.newArrayList(ELASTIC_SEARCH_ID, "12:2", "12:3", "12:4", "12:5"))); + verify(elasticSearchIndexer).deleteMessages(eq(Lists.newArrayList(ELASTIC_SEARCH_ID, "12:2", "12:3", "12:4", "12:5"))); } @Test @@ -207,7 +207,7 @@ public class ElasticSearchListeningMessageSearchIndexTest { when(mailbox.getMailboxId()) .thenReturn(MAILBOX_ID); - when(indexer.deleteMessages(any(List.class))) + when(elasticSearchIndexer.deleteMessages(any(List.class))) .thenThrow(new ElasticsearchException("")); //When @@ -241,7 +241,7 @@ public class ElasticSearchListeningMessageSearchIndexTest { //Then ImmutableList<UpdatedRepresentation> expectedUpdatedRepresentations = ImmutableList.of(new UpdatedRepresentation(ELASTIC_SEARCH_ID, "json updated content")); - verify(indexer).updateMessages(expectedUpdatedRepresentations); + verify(elasticSearchIndexer).updateMessages(expectedUpdatedRepresentations); } @Test @@ -259,7 +259,7 @@ public class ElasticSearchListeningMessageSearchIndexTest { .thenReturn(MAILBOX_ID); ImmutableList<UpdatedRepresentation> expectedUpdatedRepresentations = ImmutableList.of(new UpdatedRepresentation(ELASTIC_SEARCH_ID, "json updated content")); - when(indexer.updateMessages(expectedUpdatedRepresentations)) + when(elasticSearchIndexer.updateMessages(expectedUpdatedRepresentations)) .thenThrow(new ElasticsearchException("")); //When @@ -281,7 +281,7 @@ public class ElasticSearchListeningMessageSearchIndexTest { //Then QueryBuilder expectedQueryBuilder = QueryBuilders.termQuery("mailboxId", "12"); - verify(indexer).deleteAllMatchingQuery(refEq(expectedQueryBuilder)); + verify(elasticSearchIndexer).deleteAllMatchingQuery(refEq(expectedQueryBuilder)); } @Test @@ -292,7 +292,7 @@ public class ElasticSearchListeningMessageSearchIndexTest { .thenReturn(MAILBOX_ID); doThrow(RuntimeException.class) - .when(indexer).deleteAllMatchingQuery(QueryBuilders.termQuery("mailboxId", "12")); + .when(elasticSearchIndexer).deleteAllMatchingQuery(QueryBuilders.termQuery("mailboxId", "12")); //When testee.deleteAll(session, mailbox); http://git-wip-us.apache.org/repos/asf/james-project/blob/3ae482a7/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 2356f61..cc8af21 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 @@ -26,7 +26,7 @@ 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.ElasticSearchIndexer; import org.apache.james.backends.es.EmbeddedElasticSearch; import org.apache.james.backends.es.IndexCreationFactory; import org.apache.james.backends.es.NodeMappingFactory; @@ -39,7 +39,6 @@ import org.apache.james.mailbox.MailboxManager; import org.apache.james.mailbox.acl.SimpleGroupMembershipResolver; import org.apache.james.mailbox.elasticsearch.IndexAttachments; import org.apache.james.mailbox.elasticsearch.MailboxElasticSearchConstants; -import org.apache.james.mailbox.elasticsearch.MailboxIndexerSupplier; import org.apache.james.mailbox.elasticsearch.MailboxMappingFactory; import org.apache.james.mailbox.elasticsearch.events.ElasticSearchListeningMessageSearchIndex; import org.apache.james.mailbox.elasticsearch.json.MessageToElasticSearchJson; @@ -105,12 +104,14 @@ public class ElasticSearchHostSystem extends JamesImapHostSystem { ElasticSearchListeningMessageSearchIndex searchIndex = new ElasticSearchListeningMessageSearchIndex( factory, - new MailboxIndexerSupplier(client, - new DeleteByQueryPerformer(client, Executors.newSingleThreadExecutor(), MailboxElasticSearchConstants.DEFAULT_MAILBOX_WRITE_ALIAS, MailboxElasticSearchConstants.MESSAGE_TYPE), + new ElasticSearchIndexer(client, + Executors.newSingleThreadExecutor(), MailboxElasticSearchConstants.DEFAULT_MAILBOX_WRITE_ALIAS, MailboxElasticSearchConstants.MESSAGE_TYPE), new ElasticSearchSearcher(client, - new QueryConverter(new CriterionConverter()), new InMemoryId.Factory(), messageIdFactory, + new QueryConverter(new CriterionConverter()), + ElasticSearchSearcher.DEFAULT_SEARCH_SIZE, + new InMemoryId.Factory(), messageIdFactory, MailboxElasticSearchConstants.DEFAULT_MAILBOX_READ_ALIAS, MailboxElasticSearchConstants.MESSAGE_TYPE), new MessageToElasticSearchJson(new DefaultTextExtractor(), ZoneId.systemDefault(), IndexAttachments.YES)); http://git-wip-us.apache.org/repos/asf/james-project/blob/3ae482a7/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchConfiguration.java ---------------------------------------------------------------------- diff --git a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchConfiguration.java b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchConfiguration.java index 2fc32e9..f4ed4c9 100644 --- a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchConfiguration.java +++ b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchConfiguration.java @@ -31,6 +31,7 @@ import org.apache.james.backends.es.IndexName; import org.apache.james.mailbox.elasticsearch.IndexAttachments; import org.apache.james.mailbox.elasticsearch.MailboxElasticSearchConstants; import org.apache.james.util.Host; +import org.apache.james.util.OptionalUtils; import com.github.steveash.guavate.Guavate; import com.google.common.annotations.VisibleForTesting; @@ -41,10 +42,13 @@ public class ElasticSearchConfiguration { public static final String ELASTICSEARCH_MASTER_HOST = "elasticsearch.masterHost"; public static final String ELASTICSEARCH_PORT = "elasticsearch.port"; public static final String ELASTICSEARCH_INDEX_NAME = "elasticsearch.index.name"; + public static final String ELASTICSEARCH_INDEX_MAILBOX_NAME = "elasticsearch.index.mailbox.name"; public static final String ELASTICSEARCH_NB_REPLICA = "elasticsearch.nb.replica"; public static final String ELASTICSEARCH_NB_SHARDS = "elasticsearch.nb.shards"; public static final String ELASTICSEARCH_ALIAS_READ_NAME = "elasticsearch.alias.read.name"; public static final String ELASTICSEARCH_ALIAS_WRITE_NAME = "elasticsearch.alias.write.name"; + public static final String ELASTICSEARCH_ALIAS_READ_MAILBOX_NAME = "elasticsearch.alias.read.mailbox.name"; + public static final String ELASTICSEARCH_ALIAS_WRITE_MAILBOX_NAME = "elasticsearch.alias.write.mailbox.name"; public static final String ELASTICSEARCH_RETRY_CONNECTION_MIN_DELAY = "elasticsearch.retryConnection.minDelay"; public static final String ELASTICSEARCH_RETRY_CONNECTION_MAX_RETRIES = "elasticsearch.retryConnection.maxRetries"; public static final String ELASTICSEARCH_INDEX_ATTACHMENTS = "elasticsearch.indexAttachments"; @@ -76,16 +80,9 @@ public class ElasticSearchConfiguration { int minDelay = configuration.getInt(ELASTICSEARCH_RETRY_CONNECTION_MIN_DELAY, DEFAULT_CONNECTION_MIN_DELAY); IndexAttachments indexAttachments = provideIndexAttachments(configuration); ImmutableList<Host> hosts = getHosts(configuration); - - AliasName readAlias = Optional.ofNullable(configuration.getString(ELASTICSEARCH_ALIAS_READ_NAME)) - .map(AliasName::new) - .orElse(MailboxElasticSearchConstants.DEFAULT_MAILBOX_READ_ALIAS); - AliasName writeAlias = Optional.ofNullable(configuration.getString(ELASTICSEARCH_ALIAS_WRITE_NAME)) - .map(AliasName::new) - .orElse(MailboxElasticSearchConstants.DEFAULT_MAILBOX_WRITE_ALIAS); - IndexName indexName = Optional.ofNullable(configuration.getString(ELASTICSEARCH_INDEX_NAME)) - .map(IndexName::new) - .orElse(MailboxElasticSearchConstants.DEFAULT_MAILBOX_INDEX); + AliasName readAlias = computeMailboxReadAlias(configuration); + AliasName writeAlias = computeMailboxWriteAlias(configuration); + IndexName indexName = computeMailboxIndexName(configuration); return new ElasticSearchConfiguration( hosts, @@ -99,6 +96,33 @@ public class ElasticSearchConfiguration { indexAttachments); } + public static IndexName computeMailboxIndexName(PropertiesConfiguration configuration) { + return OptionalUtils.or( + Optional.ofNullable(configuration.getString(ELASTICSEARCH_INDEX_MAILBOX_NAME)) + .map(IndexName::new), + Optional.ofNullable(configuration.getString(ELASTICSEARCH_INDEX_NAME)) + .map(IndexName::new)) + .orElse(MailboxElasticSearchConstants.DEFAULT_MAILBOX_INDEX); + } + + public static AliasName computeMailboxWriteAlias(PropertiesConfiguration configuration) { + return OptionalUtils.or( + Optional.ofNullable(configuration.getString(ELASTICSEARCH_ALIAS_WRITE_MAILBOX_NAME)) + .map(AliasName::new), + Optional.ofNullable(configuration.getString(ELASTICSEARCH_ALIAS_WRITE_NAME)) + .map(AliasName::new)) + .orElse(MailboxElasticSearchConstants.DEFAULT_MAILBOX_WRITE_ALIAS); + } + + public static AliasName computeMailboxReadAlias(PropertiesConfiguration configuration) { + return OptionalUtils.or( + Optional.ofNullable(configuration.getString(ELASTICSEARCH_ALIAS_READ_MAILBOX_NAME)) + .map(AliasName::new), + Optional.ofNullable(configuration.getString(ELASTICSEARCH_ALIAS_READ_NAME)) + .map(AliasName::new)) + .orElse(MailboxElasticSearchConstants.DEFAULT_MAILBOX_READ_ALIAS); + } + private static IndexAttachments provideIndexAttachments(PropertiesConfiguration configuration) { if (configuration.getBoolean(ELASTICSEARCH_INDEX_ATTACHMENTS, DEFAULT_INDEX_ATTACHMENTS)) { return IndexAttachments.YES; @@ -142,22 +166,22 @@ public class ElasticSearchConfiguration { } private final ImmutableList<Host> hosts; - private final IndexName indexName; - private final AliasName readAliasName; - private final AliasName writeAliasName; + private final IndexName indexMailboxName; + private final AliasName readAliasMailboxName; + private final AliasName writeAliasMailboxName; private final int nbShards; private final int nbReplica; private final int minDelay; private final int maxRetries; private final IndexAttachments indexAttachment; - public ElasticSearchConfiguration(ImmutableList<Host> hosts, IndexName indexName, AliasName readAliasName, - AliasName writeAliasName, int nbShards, int nbReplica, int minDelay, + public ElasticSearchConfiguration(ImmutableList<Host> hosts, IndexName indexMailboxName, AliasName readAliasMailboxName, + AliasName writeAliasMailboxName, int nbShards, int nbReplica, int minDelay, int maxRetries, IndexAttachments indexAttachment) { this.hosts = hosts; - this.indexName = indexName; - this.readAliasName = readAliasName; - this.writeAliasName = writeAliasName; + this.indexMailboxName = indexMailboxName; + this.readAliasMailboxName = readAliasMailboxName; + this.writeAliasMailboxName = writeAliasMailboxName; this.nbShards = nbShards; this.nbReplica = nbReplica; this.minDelay = minDelay; @@ -169,16 +193,16 @@ public class ElasticSearchConfiguration { return hosts; } - public IndexName getIndexName() { - return indexName; + public IndexName getIndexMailboxName() { + return indexMailboxName; } - public AliasName getReadAliasName() { - return readAliasName; + public AliasName getReadAliasMailboxName() { + return readAliasMailboxName; } - public AliasName getWriteAliasName() { - return writeAliasName; + public AliasName getWriteAliasMailboxName() { + return writeAliasMailboxName; } public int getNbShards() { @@ -212,16 +236,16 @@ public class ElasticSearchConfiguration { && Objects.equals(this.maxRetries, that.maxRetries) && Objects.equals(this.indexAttachment, that.indexAttachment) && Objects.equals(this.hosts, that.hosts) - && Objects.equals(this.indexName, that.indexName) - && Objects.equals(this.readAliasName, that.readAliasName) - && Objects.equals(this.writeAliasName, that.writeAliasName); + && Objects.equals(this.indexMailboxName, that.indexMailboxName) + && Objects.equals(this.readAliasMailboxName, that.readAliasMailboxName) + && Objects.equals(this.writeAliasMailboxName, that.writeAliasMailboxName); } return false; } @Override public final int hashCode() { - return Objects.hash(hosts, indexName, readAliasName, writeAliasName, nbShards, + return Objects.hash(hosts, indexMailboxName, readAliasMailboxName, writeAliasMailboxName, nbShards, nbReplica, minDelay, maxRetries, indexAttachment); } } http://git-wip-us.apache.org/repos/asf/james-project/blob/3ae482a7/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 811bd96..a530529 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 @@ -19,28 +19,30 @@ package org.apache.james.modules.mailbox; +import static org.apache.james.mailbox.elasticsearch.search.ElasticSearchSearcher.DEFAULT_SEARCH_SIZE; + import java.io.FileNotFoundException; import java.time.LocalDateTime; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; import javax.inject.Named; import javax.inject.Singleton; import org.apache.commons.configuration.ConfigurationException; import org.apache.commons.configuration.PropertiesConfiguration; -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.IndexCreationFactory; -import org.apache.james.backends.es.IndexName; -import org.apache.james.backends.es.IndexerSupplier; import org.apache.james.backends.es.NodeMappingFactory; -import org.apache.james.backends.es.TypeName; import org.apache.james.mailbox.elasticsearch.IndexAttachments; import org.apache.james.mailbox.elasticsearch.MailboxElasticSearchConstants; -import org.apache.james.mailbox.elasticsearch.MailboxIndexerSupplier; import org.apache.james.mailbox.elasticsearch.MailboxMappingFactory; import org.apache.james.mailbox.elasticsearch.events.ElasticSearchListeningMessageSearchIndex; +import org.apache.james.mailbox.elasticsearch.query.QueryConverter; +import org.apache.james.mailbox.elasticsearch.search.ElasticSearchSearcher; +import org.apache.james.mailbox.model.MailboxId; +import org.apache.james.mailbox.model.MessageId; import org.apache.james.mailbox.store.search.ListeningMessageSearchIndex; import org.apache.james.mailbox.store.search.MessageSearchIndex; import org.apache.james.utils.PropertiesProvider; @@ -53,7 +55,6 @@ import org.slf4j.LoggerFactory; import com.google.inject.AbstractModule; import com.google.inject.Provides; import com.google.inject.Scopes; -import com.google.inject.name.Names; import com.nurkiewicz.asyncretry.AsyncRetryExecutor; public class ElasticSearchMailboxModule extends AbstractModule { @@ -63,14 +64,39 @@ public class ElasticSearchMailboxModule extends AbstractModule { @Override protected void configure() { - bind(TypeName.class).toInstance(MailboxElasticSearchConstants.MESSAGE_TYPE); bind(ElasticSearchListeningMessageSearchIndex.class).in(Scopes.SINGLETON); bind(MessageSearchIndex.class).to(ElasticSearchListeningMessageSearchIndex.class); bind(ListeningMessageSearchIndex.class).to(ElasticSearchListeningMessageSearchIndex.class); - bind(MailboxIndexerSupplier.class).in(Scopes.SINGLETON); - bind(IndexerSupplier.class) - .annotatedWith(Names.named(ElasticSearchConstants.MAILBOX_INDEX)) - .to(MailboxIndexerSupplier.class); + } + + @Provides + @Singleton + @Named(MailboxElasticSearchConstants.InjectionNames.MAILBOX) + private ElasticSearchIndexer createMailboxElasticSearchIndexer(Client client, + @Named("AsyncExecutor") ExecutorService executor, + ElasticSearchConfiguration configuration) { + return new ElasticSearchIndexer( + client, + executor, + configuration.getWriteAliasMailboxName(), + MailboxElasticSearchConstants.MESSAGE_TYPE); + } + + @Provides + @Singleton + private ElasticSearchSearcher createMailboxElasticSearchSearcher(Client client, + QueryConverter queryConverter, + MailboxId.Factory mailboxIdFactory, + MessageId.Factory messageIdFactory, + ElasticSearchConfiguration configuration) { + return new ElasticSearchSearcher( + client, + queryConverter, + DEFAULT_SEARCH_SIZE, + mailboxIdFactory, + messageIdFactory, + configuration.getReadAliasMailboxName(), + MailboxElasticSearchConstants.MESSAGE_TYPE); } @Provides @@ -86,54 +112,37 @@ public class ElasticSearchMailboxModule extends AbstractModule { } @Provides - protected IndexName provideIndexName(ElasticSearchConfiguration configuration) { - return configuration.getIndexName(); - } - - @Provides - @Named(ElasticSearchConstants.READ_ALIAS) - protected AliasName provideReadAliasName(ElasticSearchConfiguration configuration) { - return configuration.getReadAliasName(); - } - - @Provides - @Named(ElasticSearchConstants.WRITE_ALIAS) - protected AliasName provideWriteAliasName(ElasticSearchConfiguration configuration) { - return configuration.getWriteAliasName(); - } - - @Provides - @Singleton - protected IndexCreationFactory provideIndexCreationFactory(ElasticSearchConfiguration configuration) { - return new IndexCreationFactory() - .useIndex(configuration.getIndexName()) - .addAlias(configuration.getReadAliasName()) - .addAlias(configuration.getWriteAliasName()) - .nbShards(configuration.getNbShards()) - .nbReplica(configuration.getNbReplica()); - } - - @Provides @Singleton protected Client provideClient(ElasticSearchConfiguration configuration, - IndexCreationFactory indexCreationFactory, - AsyncRetryExecutor executor) throws ExecutionException, InterruptedException { + AsyncRetryExecutor executor) throws ExecutionException, InterruptedException { return RetryExecutorUtil.retryOnExceptions(executor, configuration.getMaxRetries(), configuration.getMinDelay(), NoNodeAvailableException.class) - .getWithRetry(context -> connectToCluster(configuration, indexCreationFactory)) + .getWithRetry(context -> connectToCluster(configuration)) .get(); } - private Client connectToCluster(ElasticSearchConfiguration configuration, IndexCreationFactory indexCreationFactory) { + private Client connectToCluster(ElasticSearchConfiguration configuration) { LOGGER.info("Trying to connect to ElasticSearch service at {}", LocalDateTime.now()); Client client = ClientProviderImpl.fromHosts(configuration.getHosts()).get(); - indexCreationFactory.createIndexAndAliases(client); - return NodeMappingFactory.applyMapping(client, - configuration.getIndexName(), + createMailboxIndexCreationFactory(configuration).createIndexAndAliases(client); + + NodeMappingFactory.applyMapping(client, + configuration.getIndexMailboxName(), MailboxElasticSearchConstants.MESSAGE_TYPE, MailboxMappingFactory.getMappingContent()); + + return client; + } + + protected IndexCreationFactory createMailboxIndexCreationFactory(ElasticSearchConfiguration configuration) { + return new IndexCreationFactory() + .useIndex(configuration.getIndexMailboxName()) + .addAlias(configuration.getReadAliasMailboxName()) + .addAlias(configuration.getWriteAliasMailboxName()) + .nbShards(configuration.getNbShards()) + .nbReplica(configuration.getNbReplica()); } @Provides --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
