JAMES-2398 Allow to bind ElasticSearchIndexer to different implementations
Project: http://git-wip-us.apache.org/repos/asf/james-project/repo Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/4fe5a220 Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/4fe5a220 Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/4fe5a220 Branch: refs/heads/master Commit: 4fe5a22075856f621b7dc11352386fe9da0d4321 Parents: 570d1ea Author: Raphael Ouazana <[email protected]> Authored: Wed May 16 15:32:15 2018 +0200 Committer: Matthieu Baechler <[email protected]> Committed: Mon May 21 11:37:43 2018 +0200 ---------------------------------------------------------------------- .../backends/es/ElasticSearchConstants.java | 2 + .../james/backends/es/ElasticSearchIndexer.java | 82 +------------- .../es/ElasticSearchMailboxIndexer.java | 112 +++++++++++++++++++ .../backends/es/ElasticSearchIndexerTest.java | 2 +- ...lasticSearchListeningMessageSearchIndex.java | 7 +- .../ElasticSearchIntegrationTest.java | 4 +- .../host/ElasticSearchHostSystem.java | 4 +- .../mailbox/ElasticSearchMailboxModule.java | 8 ++ 8 files changed, 137 insertions(+), 84 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/james-project/blob/4fe5a220/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 index 8830208..7429a3b 100644 --- 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 @@ -22,4 +22,6 @@ 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/4fe5a220/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 index 5b59b20..cefbd58 100644 --- 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 @@ -21,88 +21,16 @@ 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 ElasticSearchIndexer { - private static int DEBUG_MAX_LENGTH_CONTENT = 1000; +public interface ElasticSearchIndexer { + IndexResponse indexMessage(String id, String content); - 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; + Optional<BulkResponse> updateMessages(List<UpdatedRepresentation> updatedDocumentParts); - @Inject - public ElasticSearchIndexer(Client client, DeleteByQueryPerformer deleteByQueryPerformer, - @Named(ElasticSearchConstants.WRITE_ALIAS) AliasName aliasName, - TypeName typeName) { - this.client = client; - this.deleteByQueryPerformer = deleteByQueryPerformer; - this.aliasName = aliasName; - this.typeName = typeName; - } + Optional<BulkResponse> deleteMessages(List<String> ids); - 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"); - } + void deleteAllMatchingQuery(QueryBuilder queryBuilder); } http://git-wip-us.apache.org/repos/asf/james-project/blob/4fe5a220/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 new file mode 100644 index 0000000..35d5849 --- /dev/null +++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchMailboxIndexer.java @@ -0,0 +1,112 @@ +/**************************************************************** + * 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/4fe5a220/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 index c2ab50d..c811b35 100644 --- 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 @@ -72,7 +72,7 @@ public class ElasticSearchIndexerTest { doDeleteByQuery(queryBuilder); } }; - testee = new ElasticSearchIndexer(clientProvider.get(), deleteByQueryPerformer, ALIAS_NAME, TYPE_NAME); + testee = new ElasticSearchMailboxIndexer(clientProvider.get(), deleteByQueryPerformer, ALIAS_NAME, TYPE_NAME); } @Test http://git-wip-us.apache.org/repos/asf/james-project/blob/4fe5a220/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 eda8f7b..fb8c321 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 @@ -28,7 +28,9 @@ import java.util.Optional; 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.ElasticSearchIndexer; import org.apache.james.backends.es.UpdatedRepresentation; import org.apache.james.mailbox.MailboxManager.MessageCapabilities; @@ -65,8 +67,9 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe private final MessageToElasticSearchJson messageToElasticSearchJson; @Inject - public ElasticSearchListeningMessageSearchIndex(MessageMapperFactory factory, ElasticSearchIndexer indexer, - ElasticSearchSearcher searcher, MessageToElasticSearchJson messageToElasticSearchJson) { + public ElasticSearchListeningMessageSearchIndex(MessageMapperFactory factory, + @Named(ElasticSearchConstants.MAILBOX_INDEX) ElasticSearchIndexer indexer, + ElasticSearchSearcher searcher, MessageToElasticSearchJson messageToElasticSearchJson) { super(factory); this.indexer = indexer; this.messageToElasticSearchJson = messageToElasticSearchJson; http://git-wip-us.apache.org/repos/asf/james-project/blob/4fe5a220/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 49d92b8..dcb8308 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,7 @@ 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.ElasticSearchMailboxIndexer; import org.apache.james.backends.es.EmbeddedElasticSearch; import org.apache.james.backends.es.IndexCreationFactory; import org.apache.james.backends.es.NodeMappingFactory; @@ -110,7 +110,7 @@ public class ElasticSearchIntegrationTest extends AbstractMessageSearchIndexTest ElasticSearchListeningMessageSearchIndex elasticSearchListeningMessageSearchIndex = new ElasticSearchListeningMessageSearchIndex( storeMailboxManager.getMapperFactory(), - new ElasticSearchIndexer(client, + new ElasticSearchMailboxIndexer(client, new DeleteByQueryPerformer(client, Executors.newSingleThreadExecutor(), BATCH_SIZE, http://git-wip-us.apache.org/repos/asf/james-project/blob/4fe5a220/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 ea9b789..c84e2ee 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,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.ElasticSearchMailboxIndexer; import org.apache.james.backends.es.EmbeddedElasticSearch; import org.apache.james.backends.es.IndexCreationFactory; import org.apache.james.backends.es.NodeMappingFactory; @@ -105,7 +105,7 @@ public class ElasticSearchHostSystem extends JamesImapHostSystem { ElasticSearchListeningMessageSearchIndex searchIndex = new ElasticSearchListeningMessageSearchIndex( factory, - new ElasticSearchIndexer(client, + new ElasticSearchMailboxIndexer(client, new DeleteByQueryPerformer(client, Executors.newSingleThreadExecutor(), MailboxElasticSearchConstants.DEFAULT_MAILBOX_WRITE_ALIAS, MailboxElasticSearchConstants.MESSAGE_TYPE), MailboxElasticSearchConstants.DEFAULT_MAILBOX_WRITE_ALIAS, MailboxElasticSearchConstants.MESSAGE_TYPE), http://git-wip-us.apache.org/repos/asf/james-project/blob/4fe5a220/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 485f008..fa20148 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 @@ -31,6 +31,8 @@ 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.ElasticSearchMailboxIndexer; import org.apache.james.backends.es.IndexCreationFactory; import org.apache.james.backends.es.IndexName; import org.apache.james.backends.es.NodeMappingFactory; @@ -51,6 +53,7 @@ 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 { @@ -64,6 +67,11 @@ public class ElasticSearchMailboxModule extends AbstractModule { bind(ElasticSearchListeningMessageSearchIndex.class).in(Scopes.SINGLETON); bind(MessageSearchIndex.class).to(ElasticSearchListeningMessageSearchIndex.class); bind(ListeningMessageSearchIndex.class).to(ElasticSearchListeningMessageSearchIndex.class); + bind(ElasticSearchMailboxIndexer.class).in(Scopes.SINGLETON); + bind(ElasticSearchIndexer.class) + .annotatedWith(Names.named(ElasticSearchConstants.MAILBOX_INDEX)) + .to(ElasticSearchMailboxIndexer.class); +; } @Provides --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
