You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by rc...@apache.org on 2023/10/10 02:16:07 UTC
[james-project] branch master updated (a6336d718c -> 05adac3392)
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
from a6336d718c RspamdListener configuration to not report added (#1741)
new 243e0498c3 JAMES-3728 add moved mailboxId field to Expunged, Added Event
new 0d83600ed8 JAMES-3728 OpenSearch search index: upon move, GET then INDEX
new 05adac3392 JAMES-3728 OpenSearch index message support Index strategy
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/james/mailbox/events/MailboxEvents.java | 35 +++-
.../java/org/apache/james/mailbox/EventTest.java | 8 +-
.../apache/james/mailbox/MailboxListenerTest.java | 9 +-
.../james/event/json/MailboxEventSerializer.scala | 18 +-
.../james/event/json/AddedSerializationTest.java | 43 ++++-
.../event/json/ExpungedSerializationTest.java | 43 ++++-
.../opensearch/OpenSearchMailboxConfiguration.java | 39 ++++-
.../OpenSearchListeningMessageSearchIndex.java | 190 +++++++++++++++++++--
.../mailbox/opensearch/json/IndexableMessage.java | 2 +-
.../opensearch/json/MessageToOpenSearchJson.java | 26 +++
.../opensearch/OpenSearchIntegrationTest.java | 99 ++++++++++-
.../OpenSearchOptimizeMoveIntegrationTest.java} | 12 +-
.../OpenSearchListeningMessageSearchIndexTest.java | 11 +-
.../opensearch/search/OpenSearchSearcherTest.java | 4 +-
.../james/mailbox/store/StoreMessageIdManager.java | 71 +++++---
.../james/mailbox/store/StoreMessageManager.java | 2 +
.../james/mailbox/store/event/EventFactory.java | 122 ++++++++++---
.../store/search/ListeningMessageSearchIndex.java | 2 +-
.../search/AbstractMessageSearchIndexTest.java | 4 +-
.../elasticsearch/host/OpenSearchHostSystem.java | 5 +-
.../modules/ROOT/pages/configure/opensearch.adoc | 5 +
.../sample-configuration/opensearch.properties | 4 +
.../event/PopulateEmailQueryViewListenerTest.java | 3 +-
.../james/webadmin/routes/MailboxesRoutesTest.java | 4 +-
.../webadmin/routes/UserMailboxesRoutesTest.java | 4 +-
src/site/xdoc/server/config-opensearch.xml | 4 +
26 files changed, 664 insertions(+), 105 deletions(-)
copy mailbox/opensearch/src/{main/java/org/apache/james/mailbox/opensearch/IndexHeaders.java => test/java/org/apache/james/mailbox/opensearch/OpenSearchOptimizeMoveIntegrationTest.java} (80%)
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 02/03: JAMES-3728 OpenSearch search index: upon move, GET then INDEX
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 0d83600ed8c9463426078174ceaed558355f44b9
Author: Tung Tran <vt...@linagora.com>
AuthorDate: Thu Sep 28 09:56:21 2023 +0700
JAMES-3728 OpenSearch search index: upon move, GET then INDEX
---
.../OpenSearchListeningMessageSearchIndex.java | 135 ++++++++++++++++++++-
.../mailbox/opensearch/json/IndexableMessage.java | 2 +-
.../opensearch/json/MessageToOpenSearchJson.java | 26 ++++
.../opensearch/OpenSearchIntegrationTest.java | 90 ++++++++++++++
.../store/search/ListeningMessageSearchIndex.java | 2 +-
.../search/AbstractMessageSearchIndexTest.java | 4 +-
6 files changed, 250 insertions(+), 9 deletions(-)
diff --git a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java
index 5284ec9d2e..3aaa627e04 100644
--- a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java
+++ b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java
@@ -43,6 +43,7 @@ import org.apache.james.backends.opensearch.DocumentId;
import org.apache.james.backends.opensearch.OpenSearchIndexer;
import org.apache.james.backends.opensearch.RoutingKey;
import org.apache.james.backends.opensearch.UpdatedRepresentation;
+import org.apache.james.events.Event;
import org.apache.james.events.Group;
import org.apache.james.mailbox.FlagsBuilder;
import org.apache.james.mailbox.MailboxManager.MessageCapabilities;
@@ -50,15 +51,18 @@ import org.apache.james.mailbox.MailboxManager.SearchCapabilities;
import org.apache.james.mailbox.MailboxSession;
import org.apache.james.mailbox.MessageUid;
import org.apache.james.mailbox.SessionProvider;
+import org.apache.james.mailbox.events.MailboxEvents;
import org.apache.james.mailbox.model.Mailbox;
import org.apache.james.mailbox.model.MailboxId;
import org.apache.james.mailbox.model.MessageId;
+import org.apache.james.mailbox.model.MessageMetaData;
import org.apache.james.mailbox.model.SearchQuery;
import org.apache.james.mailbox.model.UpdatedFlags;
import org.apache.james.mailbox.opensearch.MailboxOpenSearchConstants;
import org.apache.james.mailbox.opensearch.json.MessageToOpenSearchJson;
import org.apache.james.mailbox.opensearch.search.OpenSearchSearcher;
import org.apache.james.mailbox.store.MailboxSessionMapperFactory;
+import org.apache.james.mailbox.store.mail.MessageMapper.FetchType;
import org.apache.james.mailbox.store.mail.model.MailboxMessage;
import org.apache.james.mailbox.store.search.ListeningMessageSearchIndex;
import org.opensearch.client.json.JsonData;
@@ -81,8 +85,16 @@ import com.google.common.collect.ImmutableList;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SynchronousSink;
+import reactor.core.scheduler.Schedulers;
public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearchIndex {
+
+ static class OpenSearchNotFoundException extends RuntimeException {
+ public OpenSearchNotFoundException(String message) {
+ super(message);
+ }
+ }
+
private static final int FLAGS_UPDATE_PROCESSING_WINDOW_SIZE = 32;
public static class OpenSearchListeningMessageSearchIndexGroup extends Group {
@@ -101,6 +113,8 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
private final MessageToOpenSearchJson messageToOpenSearchJson;
private final RoutingKey.Factory<MailboxId> routingKeyFactory;
private final MessageId.Factory messageIdFactory;
+ private final SessionProvider sessionProvider;
+ private final MailboxSessionMapperFactory factory;
@Inject
public OpenSearchListeningMessageSearchIndex(MailboxSessionMapperFactory factory,
@@ -109,6 +123,8 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
OpenSearchSearcher searcher, MessageToOpenSearchJson messageToOpenSearchJson,
SessionProvider sessionProvider, RoutingKey.Factory<MailboxId> routingKeyFactory, MessageId.Factory messageIdFactory) {
super(factory, searchOverrides, sessionProvider);
+ this.sessionProvider = sessionProvider;
+ this.factory = factory;
this.openSearchIndexer = indexer;
this.messageToOpenSearchJson = messageToOpenSearchJson;
this.searcher = searcher;
@@ -131,7 +147,113 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
SearchCapabilities.AttachmentFileName,
SearchCapabilities.PartialEmailMatch);
}
-
+
+ @Override
+ public Mono<Void> reactiveEvent(Event event) {
+ MailboxSession systemSession = sessionProvider.createSystemSession(event.getUsername());
+ return handleMailboxEvent(event, systemSession, (MailboxEvents.MailboxEvent) event)
+ .then(Mono.fromRunnable(() -> factory.endProcessingRequest(systemSession)));
+ }
+
+ private Mono<Void> handleMailboxEvent(Event event, MailboxSession session, MailboxEvents.MailboxEvent mailboxEvent) {
+ MailboxId mailboxId = mailboxEvent.getMailboxId();
+
+ if (event instanceof MailboxEvents.Added) {
+ return handleAddedEvent(session, (MailboxEvents.Added) event, mailboxId);
+ } else if (event instanceof MailboxEvents.Expunged) {
+ return handleExpungedEvent((MailboxEvents.Expunged) event, session, mailboxId);
+ } else if (event instanceof MailboxEvents.FlagsUpdated) {
+ MailboxEvents.FlagsUpdated flagsUpdated = (MailboxEvents.FlagsUpdated) event;
+ return update(session, mailboxId, flagsUpdated.getUpdatedFlags());
+ } else if (event instanceof MailboxEvents.MailboxDeletion) {
+ return deleteAll(session, mailboxId);
+ } else {
+ return Mono.empty();
+ }
+ }
+
+ private Mono<Void> handleExpungedEvent(MailboxEvents.Expunged expunged, MailboxSession session, MailboxId mailboxId) {
+ if (expunged.isMoved()) {
+ return Mono.justOrEmpty(expunged.movedToMailboxId())
+ .flatMap(movedToMailboxId -> Flux.fromIterable(expunged.getExpunged().values())
+ .concatMap(oldMessageData -> handleExpungedMessage(oldMessageData, mailboxId, movedToMailboxId, session))
+ .then(Mono.defer(() -> delete(session, mailboxId, expunged.getUids()))));
+ } else {
+ return delete(session, mailboxId, expunged.getUids());
+ }
+ }
+
+ private Mono<Void> handleExpungedMessage(MessageMetaData expungedMessageMetaData,
+ MailboxId movedFromMailboxId,
+ MailboxId movedToMailboxId,
+ MailboxSession session) {
+ return newIndexByIndexedDocument(expungedMessageMetaData.getUid(), movedFromMailboxId, movedToMailboxId, session, expungedMessageMetaData.getMessageId())
+ .onErrorResume(OpenSearchNotFoundException.class,
+ notFoundException -> {
+ LOGGER.warn("Can not find message {} in mailbox {} for reindexing",
+ expungedMessageMetaData.getUid(),
+ movedFromMailboxId.serialize());
+ return notFoundHandleFallBack(expungedMessageMetaData, movedToMailboxId, session);
+ });
+ }
+
+ private Mono<Void> notFoundHandleFallBack(MessageMetaData expungedMessageMetaData, MailboxId movedToMailboxId, MailboxSession session) {
+ return retrieveMailboxMessage(session, expungedMessageMetaData.getMessageId(), movedToMailboxId, FetchType.FULL)
+ .publishOn(Schedulers.parallel())
+ .flatMap(mailboxMessage -> factory.getMailboxMapper(session)
+ .findMailboxById(movedToMailboxId)
+ .flatMap(mailbox -> add(session, mailbox, mailboxMessage)));
+ }
+
+ private Mono<Void> newIndexByIndexedDocument(MessageUid oldMessageUid,
+ MailboxId movedFromMailboxId,
+ MailboxId movedToMailboxId,
+ MailboxSession session,
+ MessageId messageId) {
+ RoutingKey movedFromRoutingKey = routingKeyFactory.from(movedFromMailboxId);
+ return openSearchIndexer.get(indexIdFor(movedFromMailboxId, oldMessageUid), movedFromRoutingKey)
+ .filter(GetResponse::found)
+ .switchIfEmpty(Mono.error(() -> new OpenSearchNotFoundException("Can not find message " + oldMessageUid + " in mailbox " + movedFromMailboxId.serialize())))
+ .mapNotNull(GetResponse::source)
+ .flatMap(document -> updateDocumentThenIndex(messageId, movedToMailboxId, document, session));
+ }
+
+ private Mono<Void> updateDocumentThenIndex(MessageId messageId,
+ MailboxId movedToMailboxId,
+ ObjectNode origin,
+ MailboxSession session) {
+ return retrieveMailboxMessage(session, messageId, movedToMailboxId, FetchType.METADATA)
+ .flatMap(mailboxMessage -> Mono.fromCallable(() -> {
+ messageToOpenSearchJson.updateMessageUid(origin, mailboxMessage.getUid());
+ messageToOpenSearchJson.updateMailboxId(origin, movedToMailboxId);
+ mailboxMessage.getSaveDate().ifPresent(newSaveDate -> messageToOpenSearchJson.updateSaveDate(origin, newSaveDate));
+ return origin;
+ }).map(messageToOpenSearchJson::toString)
+ .flatMap(jsonContent -> add(movedToMailboxId, mailboxMessage.getUid(), jsonContent)));
+ }
+
+ private Mono<MailboxMessage> retrieveMailboxMessage(MailboxSession session,
+ MessageId messageId,
+ MailboxId movedToMailboxId,
+ FetchType fetchType) {
+ return factory.getMessageIdMapper(session)
+ .findReactive(List.of(messageId), fetchType)
+ .filter(mailboxMessage -> mailboxMessage.getMailboxId().equals(movedToMailboxId))
+ .next();
+ }
+
+ private Mono<Void> handleAddedEvent(MailboxSession session,
+ MailboxEvents.Added addedEvent,
+ MailboxId mailboxId) {
+ if (addedEvent.isMoved()) {
+ return Mono.empty();
+ } else {
+ return factory.getMailboxMapper(session)
+ .findMailboxById(mailboxId)
+ .flatMap(mailbox -> handleAdded(session, mailbox, addedEvent));
+ }
+ }
+
@Override
protected Flux<MessageUid> doSearch(MailboxSession session, Mailbox mailbox, SearchQuery searchQuery) {
Preconditions.checkArgument(session != null, "'session' is mandatory");
@@ -163,11 +285,14 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
session.getUser().asString(),
message.getUid().asLong());
- RoutingKey from = routingKeyFactory.from(mailbox.getMailboxId());
- DocumentId id = indexIdFor(mailbox.getMailboxId(), message.getUid());
-
return generateIndexedJson(mailbox, message, session)
- .flatMap(jsonContent -> openSearchIndexer.index(id, jsonContent, from))
+ .flatMap(jsonContent -> add(mailbox.getMailboxId(), message.getUid(), jsonContent));
+ }
+
+ private Mono<Void> add(MailboxId mailboxId, MessageUid messageUid, String jsonContent) {
+ RoutingKey from = routingKeyFactory.from(mailboxId);
+ DocumentId id = indexIdFor(mailboxId, messageUid);
+ return openSearchIndexer.index(id, jsonContent, from)
.then();
}
diff --git a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/json/IndexableMessage.java b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/json/IndexableMessage.java
index be7e22b6d8..00a8b23164 100644
--- a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/json/IndexableMessage.java
+++ b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/json/IndexableMessage.java
@@ -44,7 +44,7 @@ import reactor.core.publisher.Mono;
public class IndexableMessage {
- private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ssZ");
+ public static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ssZ");
public static class Builder {
private static ZonedDateTime getSanitizedInternalDate(MailboxMessage message, ZoneId zoneId) {
diff --git a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/json/MessageToOpenSearchJson.java b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/json/MessageToOpenSearchJson.java
index 4dfed28820..4a8f337924 100644
--- a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/json/MessageToOpenSearchJson.java
+++ b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/json/MessageToOpenSearchJson.java
@@ -19,19 +19,26 @@
package org.apache.james.mailbox.opensearch.json;
+import static org.apache.james.mailbox.opensearch.json.IndexableMessage.DATE_TIME_FORMATTER;
+
import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.Date;
import javax.inject.Inject;
import javax.mail.Flags;
+import org.apache.james.mailbox.MessageUid;
import org.apache.james.mailbox.ModSeq;
import org.apache.james.mailbox.extractor.TextExtractor;
+import org.apache.james.mailbox.model.MailboxId;
import org.apache.james.mailbox.opensearch.IndexAttachments;
import org.apache.james.mailbox.opensearch.IndexHeaders;
import org.apache.james.mailbox.store.mail.model.MailboxMessage;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import com.fasterxml.jackson.datatype.guava.GuavaModule;
import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
import com.github.fge.lambdas.Throwing;
@@ -90,4 +97,23 @@ public class MessageToOpenSearchJson {
Preconditions.checkNotNull(flags);
return mapper.writeValueAsString(new MessageUpdateJson(flags, modSeq));
}
+
+ public ObjectNode updateMailboxId(ObjectNode origin, MailboxId newMailboxId) {
+ origin.put(JsonMessageConstants.MAILBOX_ID, newMailboxId.serialize());
+ return origin;
+ }
+
+ public ObjectNode updateSaveDate(ObjectNode origin, Date newSaveDate) {
+ origin.put(JsonMessageConstants.SAVE_DATE, DATE_TIME_FORMATTER.format(ZonedDateTime.ofInstant(newSaveDate.toInstant(), zoneId)));
+ return origin;
+ }
+
+ public ObjectNode updateMessageUid(ObjectNode origin, MessageUid newMessageUid) {
+ origin.put(JsonMessageConstants.UID, newMessageUid.asLong());
+ return origin;
+ }
+
+ public String toString(ObjectNode objectNode) {
+ return Throwing.supplier(() -> mapper.writeValueAsString(objectNode)).get();
+ }
}
diff --git a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java
index 66c25e9f46..43414d7d67 100644
--- a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java
+++ b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java
@@ -20,6 +20,7 @@
package org.apache.james.mailbox.opensearch;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.SoftAssertions.assertSoftly;
import static org.awaitility.Durations.ONE_HUNDRED_MILLISECONDS;
import java.io.IOException;
@@ -34,12 +35,14 @@ import org.apache.james.backends.opensearch.WriteAliasName;
import org.apache.james.mailbox.MailboxSession;
import org.apache.james.mailbox.MailboxSessionUtil;
import org.apache.james.mailbox.MessageManager;
+import org.apache.james.mailbox.MessageUid;
import org.apache.james.mailbox.inmemory.InMemoryMessageId;
import org.apache.james.mailbox.inmemory.manager.InMemoryIntegrationResources;
import org.apache.james.mailbox.model.ComposedMessageId;
import org.apache.james.mailbox.model.MailboxId;
import org.apache.james.mailbox.model.MailboxPath;
import org.apache.james.mailbox.model.MessageId;
+import org.apache.james.mailbox.model.MessageRange;
import org.apache.james.mailbox.model.SearchQuery;
import org.apache.james.mailbox.opensearch.events.OpenSearchListeningMessageSearchIndex;
import org.apache.james.mailbox.opensearch.json.MessageToOpenSearchJson;
@@ -63,10 +66,15 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import org.opensearch.client.opensearch._types.query_dsl.MatchAllQuery;
import org.opensearch.client.opensearch._types.query_dsl.Query;
import org.opensearch.client.opensearch._types.query_dsl.QueryBuilders;
+import org.opensearch.client.opensearch.core.DeleteByQueryRequest;
+import org.opensearch.client.opensearch.core.GetRequest;
+import org.opensearch.client.opensearch.core.GetResponse;
import org.opensearch.client.opensearch.core.SearchRequest;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
@@ -157,6 +165,82 @@ class OpenSearchIntegrationTest extends AbstractMessageSearchIndexTest {
return InMemoryMessageId.of(1000);
}
+ @Test
+ void theDocumentShouldBeReindexWithNewMailboxWhenMoveMessages() throws Exception {
+ // Given mailboxA, mailboxB. Add message in mailboxA
+ MailboxPath mailboxA = MailboxPath.forUser(USERNAME, "mailboxA");
+ MailboxPath mailboxB = MailboxPath.forUser(USERNAME, "mailboxB");
+ MailboxId mailboxAId = storeMailboxManager.createMailbox(mailboxA, session).get();
+ MailboxId mailboxBId = storeMailboxManager.createMailbox(mailboxB, session).get();
+
+ ComposedMessageId composedMessageId = storeMailboxManager.getMailbox(mailboxAId, session)
+ .appendMessage(MessageManager.AppendCommand.from(
+ Message.Builder.of()
+ .setTo("benwa@linagora.com")
+ .setBody(Strings.repeat("append to inbox A", 5000), StandardCharsets.UTF_8)),
+ session).getId();
+
+ awaitUntilAsserted(mailboxAId, 1);
+
+ // When moving the message from mailboxA to mailboxB
+ storeMailboxManager.moveMessages(MessageRange.from(composedMessageId.getUid()), mailboxAId, mailboxBId, session);
+
+ // Then the message is not anymore when searching with mailboxA, but is in mailboxB
+ awaitUntilAsserted(mailboxAId, 0);
+
+ // verify the messageDocumentWasUpdated
+ MessageUid bMessageUid = Flux.from(storeMailboxManager.getMailbox(mailboxBId, session).search(SearchQuery.matchAll(), session))
+ .next().block();
+
+ ObjectNode updatedDocument = client.get(
+ new GetRequest.Builder()
+ .index("mailboxWriteAlias")
+ .id(mailboxBId.serialize() + ":" + bMessageUid.asLong())
+ .routing(mailboxBId.serialize())
+ .build())
+ .filter(GetResponse::found)
+ .map(GetResponse::source)
+ .block();
+
+ assertThat(updatedDocument).isNotNull();
+ assertSoftly(softly -> {
+ softly.assertThat(updatedDocument.get("mailboxId").asText()).isEqualTo(mailboxBId.serialize());
+ softly.assertThat(updatedDocument.get("uid").asLong()).isEqualTo(bMessageUid.asLong());
+ });
+ }
+
+ @Test
+ void theMessageShouldBeIndexedWhenMoveMessagesButIndexedDocumentNotFound() throws Exception {
+ // Given mailboxA, mailboxB. Add message in mailboxA
+ MailboxPath mailboxA = MailboxPath.forUser(USERNAME, "mailboxA");
+ MailboxPath mailboxB = MailboxPath.forUser(USERNAME, "mailboxB");
+ MailboxId mailboxAId = storeMailboxManager.createMailbox(mailboxA, session).get();
+ MailboxId mailboxBId = storeMailboxManager.createMailbox(mailboxB, session).get();
+
+ ComposedMessageId composedMessageId = storeMailboxManager.getMailbox(mailboxAId, session)
+ .appendMessage(MessageManager.AppendCommand.from(
+ Message.Builder.of()
+ .setTo("benwa@linagora.com")
+ .setBody(Strings.repeat("append to inbox A", 5000), StandardCharsets.UTF_8)),
+ session).getId();
+
+ awaitUntilAsserted(mailboxAId, 1);
+
+ // Try to delete the document manually to simulate a not found document.
+ client.deleteByQuery(new DeleteByQueryRequest.Builder()
+ .index("mailboxWriteAlias")
+ .query(new MatchAllQuery.Builder().build()._toQuery())
+ .build())
+ .block();
+ awaitUntilAsserted(mailboxAId, 0);
+
+ // When moving the message from mailboxA to mailboxB
+ storeMailboxManager.moveMessages(MessageRange.from(composedMessageId.getUid()), mailboxAId, mailboxBId, session);
+
+ // Then the message should be indexed in mailboxB
+ awaitUntilAsserted(mailboxBId, 1);
+ }
+
@Test
void termsBetweenOpenSearchAndLuceneLimitDueTuNonAsciiCharsShouldBeTruncated() throws Exception {
MailboxPath mailboxPath = MailboxPath.forUser(USERNAME, INBOX);
@@ -479,4 +563,10 @@ class OpenSearchIntegrationTest extends AbstractMessageSearchIndexTest {
.block()
.hits().total().value()).isEqualTo(totalHits));
}
+
+ private void awaitUntilAsserted(MailboxId mailboxId, long expectedCountResult) {
+ CALMLY_AWAIT.atMost(Durations.TEN_SECONDS)
+ .untilAsserted(() -> assertThat(messageSearchIndex.search(session, List.of(mailboxId), SearchQuery.matchAll(), 100L).toStream().count())
+ .isEqualTo(expectedCountResult));
+ }
}
\ No newline at end of file
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/search/ListeningMessageSearchIndex.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/search/ListeningMessageSearchIndex.java
index 5fb5019d56..43dde2fcfe 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/search/ListeningMessageSearchIndex.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/search/ListeningMessageSearchIndex.java
@@ -122,7 +122,7 @@ public abstract class ListeningMessageSearchIndex implements MessageSearchIndex,
}
}
- private Mono<Void> handleAdded(MailboxSession session, Mailbox mailbox, Added added) {
+ protected Mono<Void> handleAdded(MailboxSession session, Mailbox mailbox, Added added) {
return Flux.fromIterable(MessageRange.toRanges(added.getUids()))
.concatMap(range -> retrieveMailboxMessages(session, mailbox, range))
.publishOn(Schedulers.parallel())
diff --git a/mailbox/store/src/test/java/org/apache/james/mailbox/store/search/AbstractMessageSearchIndexTest.java b/mailbox/store/src/test/java/org/apache/james/mailbox/store/search/AbstractMessageSearchIndexTest.java
index f3f65979e2..c997ab3633 100644
--- a/mailbox/store/src/test/java/org/apache/james/mailbox/store/search/AbstractMessageSearchIndexTest.java
+++ b/mailbox/store/src/test/java/org/apache/james/mailbox/store/search/AbstractMessageSearchIndexTest.java
@@ -110,7 +110,7 @@ public abstract class AbstractMessageSearchIndexTest {
private Mailbox mailbox2;
private Mailbox otherMailbox;
private Mailbox quanMailbox;
- private MailboxSession session;
+ protected MailboxSession session;
private MailboxSession otherSession;
private MailboxSession quanSession;
@@ -131,7 +131,7 @@ public abstract class AbstractMessageSearchIndexTest {
private MailboxPath inboxPath;
private MailboxPath otherInboxPath;
private MailboxPath quanInboxPath;
- private StoreMessageManager inboxMessageManager;
+ protected StoreMessageManager inboxMessageManager;
private StoreMessageManager quanInboxMessageManager;
private MessageMapper messageMapper;
private MessageId newBasedMessageId;
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 01/03: JAMES-3728 add moved mailboxId field to Expunged, Added Event
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 243e0498c3ce867765f0efc30953de64b02094e7
Author: Tung Tran <vt...@linagora.com>
AuthorDate: Tue Sep 26 11:59:16 2023 +0700
JAMES-3728 add moved mailboxId field to Expunged, Added Event
---
.../apache/james/mailbox/events/MailboxEvents.java | 35 +++++-
.../java/org/apache/james/mailbox/EventTest.java | 8 +-
.../apache/james/mailbox/MailboxListenerTest.java | 9 +-
.../james/event/json/MailboxEventSerializer.scala | 18 ++-
.../james/event/json/AddedSerializationTest.java | 43 +++++++-
.../event/json/ExpungedSerializationTest.java | 43 +++++++-
.../OpenSearchListeningMessageSearchIndexTest.java | 3 +-
.../james/mailbox/store/StoreMessageIdManager.java | 71 ++++++++----
.../james/mailbox/store/StoreMessageManager.java | 2 +
.../james/mailbox/store/event/EventFactory.java | 122 +++++++++++++++++----
.../event/PopulateEmailQueryViewListenerTest.java | 3 +-
11 files changed, 288 insertions(+), 69 deletions(-)
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/events/MailboxEvents.java b/mailbox/api/src/main/java/org/apache/james/mailbox/events/MailboxEvents.java
index 4d16a85be1..bafa6d55a8 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/events/MailboxEvents.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/events/MailboxEvents.java
@@ -415,10 +415,14 @@ public interface MailboxEvents {
class Expunged extends MetaDataHoldingEvent {
private final Map<MessageUid, MessageMetaData> expunged;
+ private final Optional<MailboxId> movedToMailboxId;
- public Expunged(MailboxSession.SessionId sessionId, Username username, MailboxPath path, MailboxId mailboxId, Map<MessageUid, MessageMetaData> uids, EventId eventId) {
+ public Expunged(MailboxSession.SessionId sessionId, Username username,
+ MailboxPath path, MailboxId mailboxId, Map<MessageUid, MessageMetaData> uids,
+ EventId eventId, Optional<MailboxId> movedToMailboxId) {
super(sessionId, username, path, mailboxId, eventId);
this.expunged = ImmutableMap.copyOf(uids);
+ this.movedToMailboxId = movedToMailboxId;
}
@Override
@@ -445,6 +449,14 @@ public interface MailboxEvents {
return expunged.isEmpty();
}
+ public boolean isMoved() {
+ return movedToMailboxId.isPresent();
+ }
+
+ public Optional<MailboxId> movedToMailboxId() {
+ return movedToMailboxId;
+ }
+
@Override
public final boolean equals(Object o) {
if (o instanceof Expunged) {
@@ -455,6 +467,7 @@ public interface MailboxEvents {
&& Objects.equals(this.username, that.username)
&& Objects.equals(this.path, that.path)
&& Objects.equals(this.mailboxId, that.mailboxId)
+ && Objects.equals(this.movedToMailboxId, that.movedToMailboxId)
&& Objects.equals(this.expunged, that.expunged);
}
return false;
@@ -462,7 +475,8 @@ public interface MailboxEvents {
@Override
public final int hashCode() {
- return Objects.hash(eventId, sessionId, username, path, mailboxId, expunged);
+ return Objects.hash(eventId, sessionId, username,
+ path, mailboxId, expunged, movedToMailboxId);
}
}
@@ -533,12 +547,16 @@ public interface MailboxEvents {
private final boolean isDelivery;
private final boolean isAppended;
+ private final Optional<MailboxId> movedFromMailboxId;
+
public Added(MailboxSession.SessionId sessionId, Username username, MailboxPath path, MailboxId mailboxId,
- SortedMap<MessageUid, MessageMetaData> uids, EventId eventId, boolean isDelivery, boolean isAppended) {
+ SortedMap<MessageUid, MessageMetaData> uids, EventId eventId,
+ boolean isDelivery, boolean isAppended, Optional<MailboxId> movedFromMailboxId) {
super(sessionId, username, path, mailboxId, eventId);
this.added = ImmutableMap.copyOf(uids);
this.isDelivery = isDelivery;
this.isAppended = isAppended;
+ this.movedFromMailboxId = movedFromMailboxId;
}
/**
@@ -567,6 +585,14 @@ public interface MailboxEvents {
return isAppended;
}
+ public boolean isMoved() {
+ return movedFromMailboxId.isPresent();
+ }
+
+ public Optional<MailboxId> movedFromMailboxId() {
+ return movedFromMailboxId;
+ }
+
@Override
public boolean isNoop() {
return added.isEmpty();
@@ -584,6 +610,7 @@ public interface MailboxEvents {
&& Objects.equals(this.mailboxId, that.mailboxId)
&& Objects.equals(this.added, that.added)
&& Objects.equals(this.isAppended, that.isAppended)
+ && Objects.equals(this.movedFromMailboxId, that.movedFromMailboxId)
&& Objects.equals(this.isDelivery, that.isDelivery);
}
return false;
@@ -591,7 +618,7 @@ public interface MailboxEvents {
@Override
public final int hashCode() {
- return Objects.hash(eventId, sessionId, username, path, mailboxId, added, isDelivery, isAppended);
+ return Objects.hash(eventId, sessionId, username, path, mailboxId, added, isDelivery, isAppended, movedFromMailboxId);
}
}
diff --git a/mailbox/api/src/test/java/org/apache/james/mailbox/EventTest.java b/mailbox/api/src/test/java/org/apache/james/mailbox/EventTest.java
index 9af7cfa969..7849cd0beb 100644
--- a/mailbox/api/src/test/java/org/apache/james/mailbox/EventTest.java
+++ b/mailbox/api/src/test/java/org/apache/james/mailbox/EventTest.java
@@ -60,7 +60,9 @@ class EventTest {
@Test
void getMessageIdsShouldReturnEmptyWhenAddedEmpty() {
- Added added = new Added(MailboxSession.SessionId.of(36), BOB, MailboxPath.inbox(BOB), TestId.of(48), ImmutableSortedMap.of(), Event.EventId.of(UUID_1), !IS_DELIVERY, IS_APPENDED);
+ Added added = new Added(MailboxSession.SessionId.of(36), BOB, MailboxPath.inbox(BOB),
+ TestId.of(48), ImmutableSortedMap.of(), Event.EventId.of(UUID_1),
+ !IS_DELIVERY, IS_APPENDED, Optional.empty());
assertThat(added.getMessageIds()).isEmpty();
}
@@ -78,7 +80,7 @@ class EventTest {
ImmutableSortedMap.of(
uid1, metaData1,
uid2, metaData2),
- Event.EventId.of(UUID_1), !IS_DELIVERY, IS_APPENDED);
+ Event.EventId.of(UUID_1), !IS_DELIVERY, IS_APPENDED, Optional.empty());
assertThat(added.getMessageIds()).containsOnly(messageId1, messageId2);
}
@@ -95,7 +97,7 @@ class EventTest {
ImmutableSortedMap.of(
uid1, metaData1,
uid2, metaData2),
- Event.EventId.of(UUID_1), !IS_DELIVERY, IS_APPENDED);
+ Event.EventId.of(UUID_1), !IS_DELIVERY, IS_APPENDED, Optional.empty());
assertThat(added.getMessageIds()).containsExactly(messageId);
}
diff --git a/mailbox/api/src/test/java/org/apache/james/mailbox/MailboxListenerTest.java b/mailbox/api/src/test/java/org/apache/james/mailbox/MailboxListenerTest.java
index f4597194ba..3adb767418 100644
--- a/mailbox/api/src/test/java/org/apache/james/mailbox/MailboxListenerTest.java
+++ b/mailbox/api/src/test/java/org/apache/james/mailbox/MailboxListenerTest.java
@@ -67,6 +67,7 @@ class MailboxListenerTest {
private static final Username BOB = Username.of("bob");
private static final MailboxSession.SessionId SESSION_ID = MailboxSession.SessionId.of(42);
private static final TestId MAILBOX_ID = TestId.of(18);
+ private static final TestId MOVED_FROM_MAILBOX_ID = TestId.of(28);
private static final QuotaRoot QUOTA_ROOT = QuotaRoot.quotaRoot("bob", Optional.empty());
private static final QuotaCountUsage QUOTA_COUNT = QuotaCountUsage.count(34);
private static final QuotaSizeUsage QUOTA_SIZE = QuotaSizeUsage.size(48);
@@ -163,7 +164,7 @@ class MailboxListenerTest {
@Test
void addedShouldBeNoopWhenEmpty() {
Added added = new Added(SESSION_ID, BOB, PATH, MAILBOX_ID, ImmutableSortedMap.of(),
- Event.EventId.random(), !IS_DELIVERY, IS_APPENDED);
+ Event.EventId.random(), !IS_DELIVERY, IS_APPENDED, Optional.empty());
assertThat(added.isNoop()).isTrue();
}
@@ -171,7 +172,7 @@ class MailboxListenerTest {
@Test
void addedShouldNotBeNoopWhenNotEmpty() {
Added added = new Added(SESSION_ID, BOB, PATH, MAILBOX_ID, ImmutableSortedMap.of(UID, META_DATA),
- Event.EventId.random(), !IS_DELIVERY, IS_APPENDED);
+ Event.EventId.random(), !IS_DELIVERY, IS_APPENDED, Optional.empty());
assertThat(added.isNoop()).isFalse();
}
@@ -179,7 +180,7 @@ class MailboxListenerTest {
@Test
void expungedShouldBeNoopWhenEmpty() {
Expunged expunged = new Expunged(SESSION_ID, BOB, PATH, MAILBOX_ID, ImmutableSortedMap.of(),
- Event.EventId.random());
+ Event.EventId.random(), Optional.empty());
assertThat(expunged.isNoop()).isTrue();
}
@@ -187,7 +188,7 @@ class MailboxListenerTest {
@Test
void expungedShouldNotBeNoopWhenNotEmpty() {
Expunged expunged = new Expunged(SESSION_ID, BOB, PATH, MAILBOX_ID, ImmutableSortedMap.of(UID, META_DATA),
- Event.EventId.random());
+ Event.EventId.random(), Optional.empty());
assertThat(expunged.isNoop()).isFalse();
}
diff --git a/mailbox/event/json/src/main/scala/org/apache/james/event/json/MailboxEventSerializer.scala b/mailbox/event/json/src/main/scala/org/apache/james/event/json/MailboxEventSerializer.scala
index 53d19fff0d..a14789492e 100644
--- a/mailbox/event/json/src/main/scala/org/apache/james/event/json/MailboxEventSerializer.scala
+++ b/mailbox/event/json/src/main/scala/org/apache/james/event/json/MailboxEventSerializer.scala
@@ -40,6 +40,7 @@ import org.apache.james.mailbox.{MessageUid, ModSeq}
import play.api.libs.json._
import scala.jdk.CollectionConverters._
+import scala.jdk.OptionConverters._
private sealed trait Event {
def toJava: JavaEvent
@@ -70,7 +71,8 @@ private object DTO {
}
case class Added(eventId: EventId, sessionId: SessionId, user: Username, path: MailboxPath, mailboxId: MailboxId,
- added: Map[MessageUid, DTOs.MessageMetaData], isDelivery: Option[IsDelivery], isAppended: Option[IsAppended]) extends Event {
+ added: Map[MessageUid, DTOs.MessageMetaData], isDelivery: Option[IsDelivery], isAppended: Option[IsAppended],
+ movedFromMailboxId: Option[MailboxId]) extends Event {
override def toJava: JavaEvent = new JavaAdded(
sessionId,
user,
@@ -79,7 +81,8 @@ private object DTO {
new JavaTreeMap[MessageUid, JavaMessageMetaData](added.view.mapValues(_.toJava).toMap.asJava),
eventId,
fallbackIsDelivery(),
- fallbackIsAppended())
+ fallbackIsAppended(),
+ movedFromMailboxId.toJava)
def fallbackIsDelivery(): Boolean = isDelivery.exists(_.value)
@@ -87,14 +90,15 @@ private object DTO {
}
case class Expunged(eventId: EventId, sessionId: SessionId, user: Username, path: MailboxPath, mailboxId: MailboxId,
- expunged: Map[MessageUid, DTOs.MessageMetaData]) extends Event {
+ expunged: Map[MessageUid, DTOs.MessageMetaData], movedToMailboxId: Option[MailboxId]) extends Event {
override def toJava: JavaEvent = new JavaExpunged(
sessionId,
user,
path.toJava,
mailboxId,
expunged.view.mapValues(_.toJava).toMap.asJava,
- eventId)
+ eventId,
+ movedToMailboxId.toJava)
}
case class MessageMoveEvent(eventId: EventId, user: Username, previousMailboxIds: Iterable[MailboxId], targetMailboxIds: Iterable[MailboxId],
@@ -181,7 +185,8 @@ private object ScalaConverter {
mailboxId = event.getMailboxId,
added = event.getAdded.asScala.view.mapValues(DTOs.MessageMetaData.fromJava).toMap,
isDelivery = Option(IsDelivery(event.isDelivery)),
- isAppended = Option(IsAppended(event.isAppended)))
+ isAppended = Option(IsAppended(event.isAppended)),
+ movedFromMailboxId = event.movedFromMailboxId().toScala)
private def toScala(event: JavaExpunged): DTO.Expunged = DTO.Expunged(
eventId = event.getEventId,
@@ -189,7 +194,8 @@ private object ScalaConverter {
user = event.getUsername,
path = MailboxPath.fromJava(event.getMailboxPath),
mailboxId = event.getMailboxId,
- expunged = event.getExpunged.asScala.view.mapValues(DTOs.MessageMetaData.fromJava).toMap)
+ expunged = event.getExpunged.asScala.view.mapValues(DTOs.MessageMetaData.fromJava).toMap,
+ movedToMailboxId = event.movedToMailboxId.toScala)
private def toScala(event: JavaMessageMoveEvent): DTO.MessageMoveEvent = DTO.MessageMoveEvent(
eventId = event.getEventId,
diff --git a/mailbox/event/json/src/test/java/org/apache/james/event/json/AddedSerializationTest.java b/mailbox/event/json/src/test/java/org/apache/james/event/json/AddedSerializationTest.java
index a71cc20da9..b77328f692 100644
--- a/mailbox/event/json/src/test/java/org/apache/james/event/json/AddedSerializationTest.java
+++ b/mailbox/event/json/src/test/java/org/apache/james/event/json/AddedSerializationTest.java
@@ -59,6 +59,7 @@ class AddedSerializationTest {
private static final Username USERNAME = Username.of("user");
private static final MailboxSession.SessionId SESSION_ID = MailboxSession.SessionId.of(42);
private static final MailboxId MAILBOX_ID = TestId.of(18);
+ private static final MailboxId MOVED_FROM_MAILBOX_ID = TestId.of(28);
private static final String MAILBOX_NAME = "mailboxName";
private static final MailboxPath MAILBOX_PATH = new MailboxPath(MailboxConstants.USER_NAMESPACE, Username.of("user"), MAILBOX_NAME);
private static final MessageUid MESSAGE_UID = MessageUid.of(123456);
@@ -77,10 +78,10 @@ class AddedSerializationTest {
private static final SortedMap<MessageUid, MessageMetaData> ADDED_WITH_DISTINCT_MESSAGE_ID_AND_THREAD_ID = ImmutableSortedMap.of(
MESSAGE_UID, new MessageMetaData(MESSAGE_UID, MOD_SEQ, FLAGS, SIZE, Date.from(INSTANT), Optional.of(Date.from(INSTANT)), MESSAGE_ID, ThreadId.fromBaseMessageId(TestMessageId.of(100))));
- private static final Added DEFAULT_ADDED_EVENT = new Added(SESSION_ID, USERNAME, MAILBOX_PATH, MAILBOX_ID, ADDED, EVENT_ID, !IS_DELIVERY, IS_APPENDED);
- private static final Added BACKWARD_ADDED_EVENT = new Added(SESSION_ID, USERNAME, MAILBOX_PATH, MAILBOX_ID, BACKWARD_ADDED, EVENT_ID, !IS_DELIVERY, IS_APPENDED);
+ private static final Added DEFAULT_ADDED_EVENT = new Added(SESSION_ID, USERNAME, MAILBOX_PATH, MAILBOX_ID, ADDED, EVENT_ID, !IS_DELIVERY, IS_APPENDED, Optional.empty());
+ private static final Added BACKWARD_ADDED_EVENT = new Added(SESSION_ID, USERNAME, MAILBOX_PATH, MAILBOX_ID, BACKWARD_ADDED, EVENT_ID, !IS_DELIVERY, IS_APPENDED, Optional.empty());
private static final Added ADDED_WITH_DISTINCT_MESSAGE_ID_AND_THREAD_ID_EVENT = new Added(
- SESSION_ID, USERNAME, MAILBOX_PATH, MAILBOX_ID, ADDED_WITH_DISTINCT_MESSAGE_ID_AND_THREAD_ID, EVENT_ID, !IS_DELIVERY, IS_APPENDED);
+ SESSION_ID, USERNAME, MAILBOX_PATH, MAILBOX_ID, ADDED_WITH_DISTINCT_MESSAGE_ID_AND_THREAD_ID, EVENT_ID, !IS_DELIVERY, IS_APPENDED, Optional.of(MOVED_FROM_MAILBOX_ID));
private static final String DEFAULT_ADDED_EVENT_JSON =
"{" +
" \"Added\": {" +
@@ -138,6 +139,7 @@ class AddedSerializationTest {
" \"sessionId\": 42," +
" \"isDelivery\": false," +
" \"isAppended\": true," +
+ " \"movedFromMailboxId\": \"28\"," +
" \"user\": \"user\"" +
" }" +
"}";
@@ -192,11 +194,44 @@ class AddedSerializationTest {
.isEqualTo(BACKWARD_ADDED_EVENT);
}
+ @Test
+ void addedWithMovedFromMailboxIdShouldBeWellDeserialized() {
+ assertThat(EVENT_SERIALIZER.fromJson("{" +
+ " \"Added\": {" +
+ " \"eventId\":\"6e0dd59d-660e-4d9b-b22f-0354479f47b4\"," +
+ " \"path\": {" +
+ " \"namespace\": \"#private\"," +
+ " \"user\": \"user\"," +
+ " \"name\": \"mailboxName\"" +
+ " }," +
+ " \"mailboxId\": \"18\"," +
+ " \"added\": {" +
+ " \"123456\": {" +
+ " \"uid\": 123456," +
+ " \"modSeq\": 35," +
+ " \"flags\": {" +
+ " \"systemFlags\":[\"Answered\",\"Draft\"], " +
+ " \"userFlags\":[\"User Custom Flag\"]}," +
+ " \"size\": 45, " +
+ " \"internalDate\": \"2018-12-14T09:41:51.541Z\"," +
+ " \"messageId\": \"42\"" +
+ " }" +
+ " }," +
+ " \"sessionId\": 42," +
+ " \"user\": \"user\"," +
+ " \"movedFromMailboxId\": \"28\"" +
+ " }" +
+ "}").get())
+ .isEqualTo(new Added(SESSION_ID, USERNAME, MAILBOX_PATH, MAILBOX_ID, BACKWARD_ADDED, EVENT_ID, !IS_DELIVERY, IS_APPENDED, Optional.of(MOVED_FROM_MAILBOX_ID)));
+ }
+
+
@Nested
class WithEmptyAddedMap {
private final Added emptyAddedEvent = new Added(SESSION_ID, USERNAME, MAILBOX_PATH,
- MAILBOX_ID, ImmutableSortedMap.of(), EVENT_ID, !IS_DELIVERY, IS_APPENDED);
+ MAILBOX_ID, ImmutableSortedMap.of(), EVENT_ID, !IS_DELIVERY,
+ IS_APPENDED, Optional.empty());
private final String emptyAddedEventJson =
"{" +
" \"Added\": {" +
diff --git a/mailbox/event/json/src/test/java/org/apache/james/event/json/ExpungedSerializationTest.java b/mailbox/event/json/src/test/java/org/apache/james/event/json/ExpungedSerializationTest.java
index 4a98466031..ddb84be81c 100644
--- a/mailbox/event/json/src/test/java/org/apache/james/event/json/ExpungedSerializationTest.java
+++ b/mailbox/event/json/src/test/java/org/apache/james/event/json/ExpungedSerializationTest.java
@@ -57,6 +57,7 @@ class ExpungedSerializationTest {
private static final Username USERNAME = Username.of("user");
private static final MailboxSession.SessionId SESSION_ID = MailboxSession.SessionId.of(42);
private static final MailboxId MAILBOX_ID = TestId.of(18);
+ private static final MailboxId MOVED_MAILBOX_ID = TestId.of(28);
private static final String MAILBOX_NAME = "mailboxName";
private static final MailboxPath MAILBOX_PATH = new MailboxPath(MailboxConstants.USER_NAMESPACE, Username.of("user"), MAILBOX_NAME);
private static final MessageUid MESSAGE_UID = MessageUid.of(123456);
@@ -75,10 +76,10 @@ class ExpungedSerializationTest {
private static final Map<MessageUid, MessageMetaData> EXPUNGED_WITH_DISTINCT_MESSAGE_ID_AND_THREAD_ID = ImmutableMap.of(
MESSAGE_UID, new MessageMetaData(MESSAGE_UID, MOD_SEQ, FLAGS, SIZE, Date.from(INSTANT), Optional.of(Date.from(INSTANT)), MESSAGE_ID, ThreadId.fromBaseMessageId(TestMessageId.of(100))));
- private static final Expunged DEFAULT_EXPUNGED_EVENT = new Expunged(SESSION_ID, USERNAME, MAILBOX_PATH, MAILBOX_ID, EXPUNGED, EVENT_ID);
- private static final Expunged BACKWARD_EXPUNGED_EVENT = new Expunged(SESSION_ID, USERNAME, MAILBOX_PATH, MAILBOX_ID, BACKWARD_EXPUNGED, EVENT_ID);
+ private static final Expunged DEFAULT_EXPUNGED_EVENT = new Expunged(SESSION_ID, USERNAME, MAILBOX_PATH, MAILBOX_ID, EXPUNGED, EVENT_ID, Optional.empty());
+ private static final Expunged BACKWARD_EXPUNGED_EVENT = new Expunged(SESSION_ID, USERNAME, MAILBOX_PATH, MAILBOX_ID, BACKWARD_EXPUNGED, EVENT_ID, Optional.empty());
private static final Expunged EXPUNGED_WITH_DISTINCT_MESSAGE_ID_AND_THREAD_ID_EVENT = new Expunged(
- SESSION_ID, USERNAME, MAILBOX_PATH, MAILBOX_ID, EXPUNGED_WITH_DISTINCT_MESSAGE_ID_AND_THREAD_ID, EVENT_ID);
+ SESSION_ID, USERNAME, MAILBOX_PATH, MAILBOX_ID, EXPUNGED_WITH_DISTINCT_MESSAGE_ID_AND_THREAD_ID, EVENT_ID, Optional.of(MOVED_MAILBOX_ID));
private static final String DEFAULT_EXPUNGED_EVENT_JSON =
"{" +
" \"Expunged\": {" +
@@ -132,7 +133,8 @@ class ExpungedSerializationTest {
" }" +
" }," +
" \"sessionId\": 42," +
- " \"user\": \"user\"" +
+ " \"user\": \"user\"," +
+ " \"movedToMailboxId\": \"28\"" +
" }" +
"}";
private static final String DEFAULT_BACKWARD_EXPUNGED_EVENT_JSON =
@@ -186,11 +188,42 @@ class ExpungedSerializationTest {
.isEqualTo(BACKWARD_EXPUNGED_EVENT);
}
+ @Test
+ void expungedWithMovedToMailboxIdShouldBeWellDeserialized() {
+ assertThat(EVENT_SERIALIZER.fromJson("{" +
+ " \"Expunged\": {" +
+ " \"eventId\":\"6e0dd59d-660e-4d9b-b22f-0354479f47b4\"," +
+ " \"path\": {" +
+ " \"namespace\": \"#private\"," +
+ " \"user\": \"user\"," +
+ " \"name\": \"mailboxName\"" +
+ " }," +
+ " \"mailboxId\": \"18\"," +
+ " \"expunged\": {" +
+ " \"123456\": {" +
+ " \"uid\": 123456," +
+ " \"modSeq\": 35," +
+ " \"flags\": {" +
+ " \"systemFlags\":[\"Answered\",\"Draft\"], " +
+ " \"userFlags\":[\"User Custom Flag\"]}," +
+ " \"size\": 45, " +
+ " \"internalDate\": \"2018-12-14T09:41:51.541Z\"," +
+ " \"messageId\": \"42\"" +
+ " }" +
+ " }," +
+ " \"sessionId\": 42," +
+ " \"user\": \"user\"," +
+ " \"movedToMailboxId\": \"28\"" +
+ " }" +
+ "}").get())
+ .isEqualTo(new Expunged(SESSION_ID, USERNAME, MAILBOX_PATH, MAILBOX_ID, BACKWARD_EXPUNGED, EVENT_ID, Optional.of(MOVED_MAILBOX_ID)));
+ }
+
@Nested
class WithEmptyExpungedMap {
private final Expunged emptyExpungedEvent = new Expunged(SESSION_ID, USERNAME,
- MAILBOX_PATH, MAILBOX_ID, ImmutableMap.of(), EVENT_ID);
+ MAILBOX_PATH, MAILBOX_ID, ImmutableMap.of(), EVENT_ID, Optional.empty());
private final String emptyExpungedEventJson =
"{" +
" \"Expunged\": {" +
diff --git a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndexTest.java b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndexTest.java
index 1fa8355431..5575b3b51d 100644
--- a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndexTest.java
+++ b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndexTest.java
@@ -32,6 +32,7 @@ import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.ZoneId;
import java.util.Date;
+import java.util.Optional;
import javax.mail.Flags;
@@ -339,7 +340,7 @@ class OpenSearchListeningMessageSearchIndexTest {
outbox.getMailboxId(),
ImmutableSortedMap.of(MESSAGE_UID_1, outdatedMessageMetaData),
Event.EventId.random(),
- !IS_DELIVERY, IS_APPENDED);
+ !IS_DELIVERY, IS_APPENDED, Optional.empty());
Mono.from(testee.reactiveEvent(addedOutDatedEvent)).block();
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageIdManager.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageIdManager.java
index 83fe4fd65b..591650215d 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageIdManager.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageIdManager.java
@@ -368,8 +368,8 @@ public class StoreMessageIdManager implements MessageIdManager {
.collect(ImmutableList.toImmutableList());
return validateQuota(messageMoves, mailboxMessage.get())
- .then(addMessageToMailboxes(mailboxMessage.get(), messageMoves.addedMailboxes(), mailboxSession))
- .then(removeMessageFromMailboxes(mailboxMessage.get().getMessageId(), messagesToRemove, mailboxSession))
+ .then(addMessageToMailboxes(mailboxMessage.get(), messageMoves, mailboxSession))
+ .then(expungeMessageFromMailboxes(mailboxMessage.get().getMessageId(), messagesToRemove, mailboxSession, messageMoves))
.then(eventBus.dispatch(EventFactory.moved()
.session(mailboxSession)
.messageMoves(messageMoves.asMessageMoves())
@@ -381,7 +381,7 @@ public class StoreMessageIdManager implements MessageIdManager {
.collect(ImmutableSet.toImmutableSet())));
}
- private Mono<Void> removeMessageFromMailboxes(MessageId messageId, List<Pair<MailboxMessage, Mailbox>> messages, MailboxSession mailboxSession) {
+ private Mono<Void> expungeMessageFromMailboxes(MessageId messageId, List<Pair<MailboxMessage, Mailbox>> messages, MailboxSession mailboxSession, MessageMovesWithMailbox messageMoves) {
if (messages.isEmpty()) {
return Mono.empty();
}
@@ -394,16 +394,28 @@ public class StoreMessageIdManager implements MessageIdManager {
return Mono.from(messageIdMapper.deleteReactive(messageId, mailboxIds))
.then(Flux.fromIterable(messages)
- .flatMap(message -> eventBus.dispatch(EventFactory.expunged()
- .randomEventId()
- .mailboxSession(mailboxSession)
- .mailbox(message.getRight())
- .addMetaData(message.getLeft().metaData())
- .build(),
- new MailboxIdRegistrationKey(message.getRight().getMailboxId())), DEFAULT_CONCURRENCY)
+ .flatMap(message -> dispatchExpungedEvent(message, mailboxSession, messageMoves), DEFAULT_CONCURRENCY)
.then());
}
-
+
+ private Mono<Void> dispatchExpungedEvent(Pair<MailboxMessage, Mailbox> message, MailboxSession mailboxSession, MessageMovesWithMailbox messageMoves) {
+ return Mono.just(EventFactory.expunged()
+ .randomEventId()
+ .mailboxSession(mailboxSession)
+ .mailbox(message.getRight())
+ .addMetaData(message.getLeft().metaData()))
+ .map(eventBuilder -> {
+ if (isSingleMove(messageMoves)) {
+ return eventBuilder
+ .movedTo(messageMoves.addedMailboxes().iterator().next().getMailboxId())
+ .build();
+ } else {
+ return eventBuilder.build();
+ }
+ })
+ .flatMap(event -> eventBus.dispatch(event, new MailboxIdRegistrationKey(message.getRight().getMailboxId())));
+ }
+
private Mono<Void> dispatchFlagsChange(MailboxSession mailboxSession, MailboxId mailboxId, ImmutableList<UpdatedFlags> updatedFlags, List<Mailbox> knownMailboxes) {
if (updatedFlags.stream().anyMatch(UpdatedFlags::flagsChanged)) {
return knownMailboxes.stream()
@@ -463,10 +475,10 @@ public class StoreMessageIdManager implements MessageIdManager {
}
}
- private Mono<Void> addMessageToMailboxes(MailboxMessage mailboxMessage, Set<Mailbox> mailboxes, MailboxSession mailboxSession) {
+ private Mono<Void> addMessageToMailboxes(MailboxMessage mailboxMessage, MessageMovesWithMailbox messageMoves, MailboxSession mailboxSession) {
MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession);
- return Flux.fromIterable(mailboxes)
+ return Flux.fromIterable(messageMoves.addedMailboxes())
.flatMap(Throwing.<Mailbox, Mono<Void>>function(mailbox -> {
MailboxACL.Rfc4314Rights myRights = rightManager.myRights(mailbox, mailboxSession);
boolean shouldPreserveFlags = myRights.contains(Right.Write);
@@ -482,19 +494,34 @@ public class StoreMessageIdManager implements MessageIdManager {
.build());
return save(messageIdMapper, copy, mailbox)
- .flatMap(metadata -> eventBus.dispatch(EventFactory.added()
- .randomEventId()
- .mailboxSession(mailboxSession)
- .mailbox(mailbox)
- .addMetaData(metadata)
- .isDelivery(!IS_DELIVERY)
- .isAppended(!IS_APPENDED)
- .build(),
- new MailboxIdRegistrationKey(mailbox.getMailboxId())));
+ .flatMap(metadata -> dispatchAddedEvent(mailboxSession, mailbox, metadata, messageMoves));
}).sneakyThrow())
.then();
}
+ private Mono<Void> dispatchAddedEvent(MailboxSession mailboxSession, Mailbox mailbox, MessageMetaData messageMetaData, MessageMovesWithMailbox messageMoves) {
+ return Mono.just(EventFactory.added()
+ .randomEventId()
+ .mailboxSession(mailboxSession)
+ .mailbox(mailbox)
+ .addMetaData(messageMetaData)
+ .isDelivery(!IS_DELIVERY)
+ .isAppended(!IS_APPENDED))
+ .map(eventBuilder -> {
+ if (isSingleMove(messageMoves)) {
+ return eventBuilder
+ .movedFrom(messageMoves.removedMailboxes().iterator().next().getMailboxId())
+ .build();
+ } else {
+ return eventBuilder.build();
+ }
+ }).flatMap(event -> eventBus.dispatch(event, new MailboxIdRegistrationKey(mailbox.getMailboxId())));
+ }
+
+ private boolean isSingleMove(MessageMovesWithMailbox messageMoves) {
+ return messageMoves.addedMailboxes().size() == 1 && messageMoves.removedMailboxes().size() == 1;
+ }
+
private Mono<MessageMetaData> save(MessageIdMapper messageIdMapper, MailboxMessage mailboxMessage, Mailbox mailbox) {
return Mono.zip(
mailboxSessionMapperFactory.getModSeqProvider().nextModSeqReactive(mailbox.getMailboxId()),
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java
index 2cea1f2a43..d2f5ac73f3 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java
@@ -932,6 +932,7 @@ public class StoreMessageManager implements MessageManager {
.metaData(moveUids)
.isDelivery(!IS_DELIVERY)
.isAppended(!IS_APPENDED)
+ .movedFrom(getId())
.build(),
new MailboxIdRegistrationKey(to.getMailboxEntity().getMailboxId())),
eventBus.dispatch(EventFactory.expunged()
@@ -939,6 +940,7 @@ public class StoreMessageManager implements MessageManager {
.mailboxSession(session)
.mailbox(getMailboxEntity())
.addMetaData(moveResult.getOriginalMessages())
+ .movedTo(to.getId())
.build(),
new MailboxIdRegistrationKey(mailbox.getMailboxId())),
eventBus.dispatch(EventFactory.moved()
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/event/EventFactory.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/event/EventFactory.java
index 9ebf759679..0a248d0cc0 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/event/EventFactory.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/event/EventFactory.java
@@ -19,9 +19,13 @@
package org.apache.james.mailbox.store.event;
+import static org.apache.james.mailbox.events.MailboxEvents.Added.IS_APPENDED;
+import static org.apache.james.mailbox.events.MailboxEvents.Added.IS_DELIVERY;
+
import java.time.Instant;
import java.util.Iterator;
import java.util.Map;
+import java.util.Optional;
import java.util.SortedMap;
import java.util.function.Function;
@@ -145,16 +149,6 @@ public class EventFactory {
}
}
- @FunctionalInterface
- public interface RequireIsDelivery<T> {
- T isDelivery(boolean isDelivery);
- }
-
- @FunctionalInterface
- public interface RequireIsAppended<T> {
- T isAppended(boolean isAppended);
- }
-
@FunctionalInterface
public interface RequireAclDiff<T> {
T aclDiff(ACLDiff aclDiff);
@@ -232,6 +226,55 @@ public class EventFactory {
}
public static class AddedFinalStage {
+
+ public static class Builder {
+ static final boolean IS_DELIVERY_DEFAULT = IS_DELIVERY;
+ static final boolean IS_APPENDED_DEFAULT = IS_APPENDED;
+
+ private final Event.EventId eventId;
+ private final MailboxPath path;
+ private final MailboxId mailboxId;
+ private final Username username;
+ private final MailboxSession.SessionId sessionId;
+ private final Map<MessageUid, MessageMetaData> metaData;
+ private boolean isDelivery;
+
+ private boolean isAppended;
+ private Optional<MailboxId> movedFromMailboxId;
+
+ public Builder(Event.EventId eventId, MailboxPath path, MailboxId mailboxId, Username username,
+ MailboxSession.SessionId sessionId, Map<MessageUid, MessageMetaData> metaData) {
+ this.eventId = eventId;
+ this.path = path;
+ this.mailboxId = mailboxId;
+ this.username = username;
+ this.sessionId = sessionId;
+ this.metaData = metaData;
+ this.isDelivery = IS_DELIVERY_DEFAULT;
+ this.isAppended = IS_APPENDED_DEFAULT;
+ this.movedFromMailboxId = Optional.empty();
+ }
+
+ public Builder isDelivery(boolean isDelivery) {
+ this.isDelivery = isDelivery;
+ return this;
+ }
+
+ public Builder isAppended(boolean isAppended) {
+ this.isAppended = isAppended;
+ return this;
+ }
+
+ public Builder movedFrom(MailboxId mailboxId) {
+ this.movedFromMailboxId = Optional.of(mailboxId);
+ return this;
+ }
+
+ public Added build() {
+ return new AddedFinalStage(eventId, path, mailboxId, username, sessionId, metaData, isDelivery, isAppended, movedFromMailboxId).build();
+ }
+ }
+
private final Event.EventId eventId;
private final MailboxPath path;
private final MailboxId mailboxId;
@@ -241,10 +284,11 @@ public class EventFactory {
private final boolean isDelivery;
private final boolean isAppended;
+ private final Optional<MailboxId> movedFromMailboxId;
AddedFinalStage(Event.EventId eventId, MailboxPath path, MailboxId mailboxId, Username username,
MailboxSession.SessionId sessionId, Map<MessageUid, MessageMetaData> metaData,
- boolean isDelivery, boolean isAppended) {
+ boolean isDelivery, boolean isAppended, Optional<MailboxId> movedFromMailboxId) {
this.eventId = eventId;
this.path = path;
this.mailboxId = mailboxId;
@@ -253,6 +297,7 @@ public class EventFactory {
this.metaData = ImmutableSortedMap.copyOf(metaData);
this.isDelivery = isDelivery;
this.isAppended = isAppended;
+ this.movedFromMailboxId = movedFromMailboxId;
}
public Added build() {
@@ -262,25 +307,61 @@ public class EventFactory {
Preconditions.checkNotNull(sessionId);
Preconditions.checkNotNull(metaData);
- return new Added(sessionId, username, path, mailboxId, metaData, eventId, isDelivery, isAppended);
+ return new Added(sessionId, username, path, mailboxId, metaData, eventId, isDelivery, isAppended, movedFromMailboxId);
}
}
public static class ExpungedFinalStage {
+
+ public static class Builder {
+ private final Event.EventId eventId;
+ private final MailboxPath path;
+ private final MailboxId mailboxId;
+ private final Username username;
+ private final MailboxSession.SessionId sessionId;
+ private final Map<MessageUid, MessageMetaData> metaData;
+
+ private Optional<MailboxId> movedToMailboxId;
+
+ public Builder(Event.EventId eventId, MailboxPath path, MailboxId mailboxId,
+ Username username, MailboxSession.SessionId sessionId, Map<MessageUid, MessageMetaData> metaData) {
+ this.eventId = eventId;
+ this.path = path;
+ this.mailboxId = mailboxId;
+ this.username = username;
+ this.sessionId = sessionId;
+ this.metaData = metaData;
+ this.movedToMailboxId = Optional.empty();
+ }
+
+ public Builder movedTo(MailboxId mailboxId) {
+ this.movedToMailboxId = Optional.of(mailboxId);
+ return this;
+ }
+
+ public Expunged build() {
+ return new ExpungedFinalStage(eventId, path, mailboxId, username, sessionId, metaData, movedToMailboxId).build();
+ }
+ }
+
private final Event.EventId eventId;
private final MailboxPath path;
private final MailboxId mailboxId;
private final Username username;
private final MailboxSession.SessionId sessionId;
private final ImmutableSortedMap<MessageUid, MessageMetaData> metaData;
+ private final Optional<MailboxId> movedToMailboxId;
- ExpungedFinalStage(Event.EventId eventId, MailboxPath path, MailboxId mailboxId, Username username, MailboxSession.SessionId sessionId, Map<MessageUid, MessageMetaData> metaData) {
+ ExpungedFinalStage(Event.EventId eventId, MailboxPath path, MailboxId mailboxId, Username username,
+ MailboxSession.SessionId sessionId, Map<MessageUid, MessageMetaData> metaData,
+ Optional<MailboxId> movedToMailboxId) {
this.eventId = eventId;
this.path = path;
this.mailboxId = mailboxId;
this.username = username;
this.sessionId = sessionId;
this.metaData = ImmutableSortedMap.copyOf(metaData);
+ this.movedToMailboxId = movedToMailboxId;
}
public Expunged build() {
@@ -289,8 +370,9 @@ public class EventFactory {
Preconditions.checkNotNull(username);
Preconditions.checkNotNull(sessionId);
Preconditions.checkNotNull(metaData);
+ Preconditions.checkNotNull(movedToMailboxId);
- return new Expunged(sessionId, username, path, mailboxId, metaData, eventId);
+ return new Expunged(sessionId, username, path, mailboxId, metaData, eventId, movedToMailboxId);
}
}
@@ -488,13 +570,15 @@ public class EventFactory {
}
}
- public static RequireMailboxEvent<RequireMetadata<RequireIsDelivery<RequireIsAppended<AddedFinalStage>>>> added() {
- return eventId -> user -> sessionId -> mailboxId -> path -> metaData -> isDelivery -> isAppended
- -> new AddedFinalStage(eventId, path, mailboxId, user, sessionId, metaData, isDelivery, isAppended);
+
+ public static RequireMailboxEvent<RequireMetadata<AddedFinalStage.Builder>> added() {
+ return eventId -> user -> sessionId -> mailboxId -> path -> metaData
+ -> new AddedFinalStage.Builder(eventId, path, mailboxId, user, sessionId, metaData);
}
- public static RequireMailboxEvent<RequireMetadata<ExpungedFinalStage>> expunged() {
- return eventId -> user -> sessionId -> mailboxId -> path -> metaData -> new ExpungedFinalStage(eventId, path, mailboxId, user, sessionId, metaData);
+ public static RequireMailboxEvent<RequireMetadata<ExpungedFinalStage.Builder>> expunged() {
+ return eventId -> user -> sessionId -> mailboxId -> path -> metaData
+ -> new ExpungedFinalStage.Builder(eventId, path, mailboxId, user, sessionId, metaData);
}
public static RequireMailboxEvent<RequireUpdatedFlags<FlagsUpdatedFinalStage>> flagsUpdated() {
diff --git a/server/protocols/jmap-draft/src/test/java/org/apache/james/jmap/event/PopulateEmailQueryViewListenerTest.java b/server/protocols/jmap-draft/src/test/java/org/apache/james/jmap/event/PopulateEmailQueryViewListenerTest.java
index e51becedf4..4d9a073051 100644
--- a/server/protocols/jmap-draft/src/test/java/org/apache/james/jmap/event/PopulateEmailQueryViewListenerTest.java
+++ b/server/protocols/jmap-draft/src/test/java/org/apache/james/jmap/event/PopulateEmailQueryViewListenerTest.java
@@ -188,7 +188,8 @@ public class PopulateEmailQueryViewListenerTest {
ImmutableSortedMap.of(MessageUid.of(1), outdatedMessageMetaData),
Event.EventId.random(),
!IS_DELIVERY,
- !IS_APPENDED);
+ !IS_APPENDED,
+ Optional.empty());
Mono.from(queryViewListener.reactiveEvent(addedOutDatedEvent)).block();
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 03/03: JAMES-3728 OpenSearch index message support Index strategy
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 05adac3392c16cc6c2e9b39b3f1b038728e02bb8
Author: Tung Tran <vt...@linagora.com>
AuthorDate: Fri Oct 6 14:37:42 2023 +0700
JAMES-3728 OpenSearch index message support Index strategy
---
.../opensearch/OpenSearchMailboxConfiguration.java | 39 +++-
.../OpenSearchListeningMessageSearchIndex.java | 219 ++++++++++++---------
.../opensearch/OpenSearchIntegrationTest.java | 9 +-
.../OpenSearchOptimizeMoveIntegrationTest.java | 30 +++
.../OpenSearchListeningMessageSearchIndexTest.java | 8 +-
.../opensearch/search/OpenSearchSearcherTest.java | 4 +-
.../elasticsearch/host/OpenSearchHostSystem.java | 5 +-
.../modules/ROOT/pages/configure/opensearch.adoc | 5 +
.../sample-configuration/opensearch.properties | 4 +
.../james/webadmin/routes/MailboxesRoutesTest.java | 4 +-
.../webadmin/routes/UserMailboxesRoutesTest.java | 4 +-
src/site/xdoc/server/config-opensearch.xml | 4 +
12 files changed, 229 insertions(+), 106 deletions(-)
diff --git a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/OpenSearchMailboxConfiguration.java b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/OpenSearchMailboxConfiguration.java
index cfa20857f2..094409581e 100644
--- a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/OpenSearchMailboxConfiguration.java
+++ b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/OpenSearchMailboxConfiguration.java
@@ -36,46 +36,55 @@ public class OpenSearchMailboxConfiguration {
private Optional<IndexAttachments> indexAttachment;
private Optional<IndexHeaders> indexHeaders;
+ private Optional<Boolean> optimiseMoves;
+
Builder() {
indexMailboxName = Optional.empty();
readAliasMailboxName = Optional.empty();
writeAliasMailboxName = Optional.empty();
indexAttachment = Optional.empty();
indexHeaders = Optional.empty();
+ optimiseMoves = Optional.empty();
}
- Builder indexMailboxName(Optional<IndexName> indexMailboxName) {
+ public Builder indexMailboxName(Optional<IndexName> indexMailboxName) {
this.indexMailboxName = indexMailboxName;
return this;
}
- Builder readAliasMailboxName(Optional<ReadAliasName> readAliasMailboxName) {
+ public Builder readAliasMailboxName(Optional<ReadAliasName> readAliasMailboxName) {
this.readAliasMailboxName = readAliasMailboxName;
return this;
}
- Builder writeAliasMailboxName(Optional<WriteAliasName> writeAliasMailboxName) {
+ public Builder writeAliasMailboxName(Optional<WriteAliasName> writeAliasMailboxName) {
this.writeAliasMailboxName = writeAliasMailboxName;
return this;
}
- Builder indexAttachment(IndexAttachments indexAttachment) {
+ public Builder indexAttachment(IndexAttachments indexAttachment) {
this.indexAttachment = Optional.of(indexAttachment);
return this;
}
- Builder indexHeaders(IndexHeaders indexHeaders) {
+ public Builder indexHeaders(IndexHeaders indexHeaders) {
this.indexHeaders = Optional.of(indexHeaders);
return this;
}
+ public Builder optimiseMoves(Boolean optimiseMoves) {
+ this.optimiseMoves = Optional.ofNullable(optimiseMoves);
+ return this;
+ }
+
public OpenSearchMailboxConfiguration build() {
return new OpenSearchMailboxConfiguration(
indexMailboxName.orElse(MailboxOpenSearchConstants.DEFAULT_MAILBOX_INDEX),
readAliasMailboxName.orElse(MailboxOpenSearchConstants.DEFAULT_MAILBOX_READ_ALIAS),
writeAliasMailboxName.orElse(MailboxOpenSearchConstants.DEFAULT_MAILBOX_WRITE_ALIAS),
indexAttachment.orElse(IndexAttachments.YES),
- indexHeaders.orElse(IndexHeaders.YES));
+ indexHeaders.orElse(IndexHeaders.YES),
+ optimiseMoves.orElse(DEFAULT_OPTIMIZE_MOVES));
}
}
@@ -91,9 +100,10 @@ public class OpenSearchMailboxConfiguration {
private static final String OPENSEARCH_ALIAS_WRITE_MAILBOX_NAME = "opensearch.alias.write.mailbox.name";
private static final String OPENSEARCH_INDEX_ATTACHMENTS = "opensearch.indexAttachments";
private static final String OPENSEARCH_INDEX_HEADERS = "opensearch.indexHeaders";
+ private static final String OPENSEARCH_MESSAGE_INDEX_OPTIMIZE_MOVE = "opensearch.message.index.optimize.move";
private static final boolean DEFAULT_INDEX_ATTACHMENTS = true;
private static final boolean DEFAULT_INDEX_HEADERS = true;
-
+ public static final boolean DEFAULT_OPTIMIZE_MOVES = false;
public static final OpenSearchMailboxConfiguration DEFAULT_CONFIGURATION = builder().build();
public static OpenSearchMailboxConfiguration fromProperties(Configuration configuration) {
@@ -103,6 +113,7 @@ public class OpenSearchMailboxConfiguration {
.writeAliasMailboxName(computeMailboxWriteAlias(configuration))
.indexAttachment(provideIndexAttachments(configuration))
.indexHeaders(provideIndexHeaders(configuration))
+ .optimiseMoves(configuration.getBoolean(OPENSEARCH_MESSAGE_INDEX_OPTIMIZE_MOVE, null))
.build();
}
@@ -140,6 +151,7 @@ public class OpenSearchMailboxConfiguration {
}
return IndexHeaders.NO;
}
+
private final IndexName indexMailboxName;
private final ReadAliasName readAliasMailboxName;
@@ -147,13 +159,17 @@ public class OpenSearchMailboxConfiguration {
private final IndexAttachments indexAttachment;
private final IndexHeaders indexHeaders;
+ private final boolean optimiseMoves;
+
private OpenSearchMailboxConfiguration(IndexName indexMailboxName, ReadAliasName readAliasMailboxName,
- WriteAliasName writeAliasMailboxName, IndexAttachments indexAttachment, IndexHeaders indexHeaders) {
+ WriteAliasName writeAliasMailboxName, IndexAttachments indexAttachment,
+ IndexHeaders indexHeaders, boolean optimiseMoves) {
this.indexMailboxName = indexMailboxName;
this.readAliasMailboxName = readAliasMailboxName;
this.writeAliasMailboxName = writeAliasMailboxName;
this.indexAttachment = indexAttachment;
this.indexHeaders = indexHeaders;
+ this.optimiseMoves = optimiseMoves;
}
public IndexName getIndexMailboxName() {
@@ -176,6 +192,10 @@ public class OpenSearchMailboxConfiguration {
return indexHeaders;
}
+ public boolean isOptimiseMoves() {
+ return optimiseMoves;
+ }
+
@Override
public final boolean equals(Object o) {
if (o instanceof OpenSearchMailboxConfiguration) {
@@ -185,6 +205,7 @@ public class OpenSearchMailboxConfiguration {
&& Objects.equals(this.indexHeaders, that.indexHeaders)
&& Objects.equals(this.indexMailboxName, that.indexMailboxName)
&& Objects.equals(this.readAliasMailboxName, that.readAliasMailboxName)
+ && Objects.equals(this.optimiseMoves, that.optimiseMoves)
&& Objects.equals(this.writeAliasMailboxName, that.writeAliasMailboxName);
}
return false;
@@ -192,6 +213,6 @@ public class OpenSearchMailboxConfiguration {
@Override
public final int hashCode() {
- return Objects.hash(indexMailboxName, readAliasMailboxName, writeAliasMailboxName, indexAttachment, indexHeaders, writeAliasMailboxName);
+ return Objects.hash(indexMailboxName, readAliasMailboxName, writeAliasMailboxName, indexAttachment, indexHeaders, writeAliasMailboxName, optimiseMoves);
}
}
diff --git a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java
index 3aaa627e04..9cc37b9933 100644
--- a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java
+++ b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java
@@ -59,12 +59,15 @@ import org.apache.james.mailbox.model.MessageMetaData;
import org.apache.james.mailbox.model.SearchQuery;
import org.apache.james.mailbox.model.UpdatedFlags;
import org.apache.james.mailbox.opensearch.MailboxOpenSearchConstants;
+import org.apache.james.mailbox.opensearch.OpenSearchMailboxConfiguration;
import org.apache.james.mailbox.opensearch.json.MessageToOpenSearchJson;
import org.apache.james.mailbox.opensearch.search.OpenSearchSearcher;
import org.apache.james.mailbox.store.MailboxSessionMapperFactory;
import org.apache.james.mailbox.store.mail.MessageMapper.FetchType;
import org.apache.james.mailbox.store.mail.model.MailboxMessage;
import org.apache.james.mailbox.store.search.ListeningMessageSearchIndex;
+import org.apache.james.metrics.api.Metric;
+import org.apache.james.metrics.api.MetricFactory;
import org.opensearch.client.json.JsonData;
import org.opensearch.client.opensearch._types.FieldValue;
import org.opensearch.client.opensearch._types.query_dsl.Query;
@@ -95,6 +98,112 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
}
}
+ interface IndexingStrategy {
+ Mono<Void> handleExpungedEvent(MailboxEvents.Expunged expunged, MailboxSession session, MailboxId mailboxId);
+
+ Mono<Void> handleAddedEvent(MailboxSession session, MailboxEvents.Added addedEvent, MailboxId mailboxId);
+ }
+
+ class NaiveIndexingStrategy implements IndexingStrategy {
+
+ @Override
+ public Mono<Void> handleExpungedEvent(MailboxEvents.Expunged expunged, MailboxSession session, MailboxId mailboxId) {
+ return delete(session, expunged.getMailboxId(), expunged.getUids());
+ }
+
+ @Override
+ public Mono<Void> handleAddedEvent(MailboxSession session, MailboxEvents.Added addedEvent, MailboxId mailboxId) {
+ return processAddedEvent(session, addedEvent, mailboxId);
+ }
+ }
+
+
+ class OptimizedIndexingStrategy implements IndexingStrategy {
+
+ @Override
+ public Mono<Void> handleExpungedEvent(MailboxEvents.Expunged expunged, MailboxSession session, MailboxId mailboxId) {
+ return Mono.just(expunged)
+ .filter(MailboxEvents.Expunged::isMoved)
+ .flatMap(expungedEvent -> processExpunged(session, mailboxId, expungedEvent).thenReturn(expunged))
+ .switchIfEmpty(Mono.defer(() -> delete(session, mailboxId, expunged.getUids())).thenReturn(expunged))
+ .then();
+ }
+
+ @Override
+ public Mono<Void> handleAddedEvent(MailboxSession session, MailboxEvents.Added addedEvent, MailboxId mailboxId) {
+ return Mono.just(addedEvent)
+ .filter(event -> !event.isMoved())
+ .flatMap(added -> processAddedEvent(session, added, mailboxId));
+ }
+
+ private Mono<Void> processExpunged(MailboxSession session, MailboxId mailboxId, MailboxEvents.Expunged expungedEvent) {
+ return Mono.justOrEmpty(expungedEvent.movedToMailboxId())
+ .flatMap(movedToMailboxId -> Flux.fromIterable(expungedEvent.getExpunged().values())
+ .concatMap(oldMessageData -> handleExpungedMessage(oldMessageData, mailboxId, movedToMailboxId, session))
+ .then(Mono.defer(() -> delete(session, mailboxId, expungedEvent.getUids()))));
+ }
+
+ private Mono<Void> handleExpungedMessage(MessageMetaData expungedMessageMetaData,
+ MailboxId movedFromMailboxId,
+ MailboxId movedToMailboxId,
+ MailboxSession session) {
+ return newIndexByIndexedDocument(expungedMessageMetaData.getUid(), movedFromMailboxId, movedToMailboxId, session, expungedMessageMetaData.getMessageId())
+ .onErrorResume(OpenSearchNotFoundException.class,
+ notFoundException -> {
+ LOGGER.warn("Can not find message {} in mailbox {} for reindexing",
+ expungedMessageMetaData.getUid(),
+ movedFromMailboxId.serialize());
+ return notFoundHandleFallBack(expungedMessageMetaData, movedToMailboxId, session);
+ });
+ }
+
+ private Mono<Void> newIndexByIndexedDocument(MessageUid oldMessageUid,
+ MailboxId movedFromMailboxId,
+ MailboxId movedToMailboxId,
+ MailboxSession session,
+ MessageId messageId) {
+ RoutingKey movedFromRoutingKey = routingKeyFactory.from(movedFromMailboxId);
+ return openSearchIndexer.get(indexIdFor(movedFromMailboxId, oldMessageUid), movedFromRoutingKey)
+ .filter(GetResponse::found)
+ .switchIfEmpty(Mono.error(() -> new OpenSearchNotFoundException("Can not find message " + oldMessageUid + " in mailbox " + movedFromMailboxId.serialize())))
+ .mapNotNull(GetResponse::source)
+ .flatMap(document -> updateDocumentThenIndex(messageId, movedToMailboxId, document, session));
+ }
+
+ private Mono<Void> notFoundHandleFallBack(MessageMetaData expungedMessageMetaData, MailboxId movedToMailboxId, MailboxSession session) {
+ return retrieveMailboxMessage(session, expungedMessageMetaData.getMessageId(), movedToMailboxId, FetchType.FULL)
+ .publishOn(Schedulers.parallel())
+ .flatMap(mailboxMessage -> factory.getMailboxMapper(session)
+ .findMailboxById(movedToMailboxId)
+ .flatMap(mailbox -> add(session, mailbox, mailboxMessage))
+ .doOnSuccess(any -> reIndexNotFoundMetric.increment()));
+ }
+
+ private Mono<Void> updateDocumentThenIndex(MessageId messageId,
+ MailboxId movedToMailboxId,
+ ObjectNode origin,
+ MailboxSession session) {
+ return retrieveMailboxMessage(session, messageId, movedToMailboxId, FetchType.METADATA)
+ .flatMap(mailboxMessage -> Mono.fromCallable(() -> {
+ messageToOpenSearchJson.updateMessageUid(origin, mailboxMessage.getUid());
+ messageToOpenSearchJson.updateMailboxId(origin, movedToMailboxId);
+ mailboxMessage.getSaveDate().ifPresent(newSaveDate -> messageToOpenSearchJson.updateSaveDate(origin, newSaveDate));
+ return origin;
+ }).map(messageToOpenSearchJson::toString)
+ .flatMap(jsonContent -> add(movedToMailboxId, mailboxMessage.getUid(), jsonContent)));
+ }
+
+ private Mono<MailboxMessage> retrieveMailboxMessage(MailboxSession session,
+ MessageId messageId,
+ MailboxId movedToMailboxId,
+ FetchType fetchType) {
+ return factory.getMessageIdMapper(session)
+ .findReactive(List.of(messageId), fetchType)
+ .filter(mailboxMessage -> mailboxMessage.getMailboxId().equals(movedToMailboxId))
+ .next();
+ }
+ }
+
private static final int FLAGS_UPDATE_PROCESSING_WINDOW_SIZE = 32;
public static class OpenSearchListeningMessageSearchIndexGroup extends Group {
@@ -115,13 +224,17 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
private final MessageId.Factory messageIdFactory;
private final SessionProvider sessionProvider;
private final MailboxSessionMapperFactory factory;
+
+ private final Metric reIndexNotFoundMetric;
+ private final IndexingStrategy indexingStrategy;
@Inject
public OpenSearchListeningMessageSearchIndex(MailboxSessionMapperFactory factory,
Set<SearchOverride> searchOverrides,
@Named(MailboxOpenSearchConstants.InjectionNames.MAILBOX) OpenSearchIndexer indexer,
OpenSearchSearcher searcher, MessageToOpenSearchJson messageToOpenSearchJson,
- SessionProvider sessionProvider, RoutingKey.Factory<MailboxId> routingKeyFactory, MessageId.Factory messageIdFactory) {
+ SessionProvider sessionProvider, RoutingKey.Factory<MailboxId> routingKeyFactory, MessageId.Factory messageIdFactory,
+ OpenSearchMailboxConfiguration configuration, MetricFactory metricFactory) {
super(factory, searchOverrides, sessionProvider);
this.sessionProvider = sessionProvider;
this.factory = factory;
@@ -130,6 +243,14 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
this.searcher = searcher;
this.routingKeyFactory = routingKeyFactory;
this.messageIdFactory = messageIdFactory;
+ if (configuration.isOptimiseMoves()) {
+ this.indexingStrategy = new OptimizedIndexingStrategy();
+ } else {
+ this.indexingStrategy = new NaiveIndexingStrategy();
+ }
+ this.reIndexNotFoundMetric = metricFactory.generate("opensearch_reindex_not_found");
+
+ LOGGER.info("OpenSearchMessageSearchIndex activated with index strategy: {}", indexingStrategy.getClass().getSimpleName());
}
@Override
@@ -159,9 +280,9 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
MailboxId mailboxId = mailboxEvent.getMailboxId();
if (event instanceof MailboxEvents.Added) {
- return handleAddedEvent(session, (MailboxEvents.Added) event, mailboxId);
+ return indexingStrategy.handleAddedEvent(session, (MailboxEvents.Added) event, mailboxId);
} else if (event instanceof MailboxEvents.Expunged) {
- return handleExpungedEvent((MailboxEvents.Expunged) event, session, mailboxId);
+ return indexingStrategy.handleExpungedEvent((MailboxEvents.Expunged) event, session, mailboxId);
} else if (event instanceof MailboxEvents.FlagsUpdated) {
MailboxEvents.FlagsUpdated flagsUpdated = (MailboxEvents.FlagsUpdated) event;
return update(session, mailboxId, flagsUpdated.getUpdatedFlags());
@@ -172,86 +293,10 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
}
}
- private Mono<Void> handleExpungedEvent(MailboxEvents.Expunged expunged, MailboxSession session, MailboxId mailboxId) {
- if (expunged.isMoved()) {
- return Mono.justOrEmpty(expunged.movedToMailboxId())
- .flatMap(movedToMailboxId -> Flux.fromIterable(expunged.getExpunged().values())
- .concatMap(oldMessageData -> handleExpungedMessage(oldMessageData, mailboxId, movedToMailboxId, session))
- .then(Mono.defer(() -> delete(session, mailboxId, expunged.getUids()))));
- } else {
- return delete(session, mailboxId, expunged.getUids());
- }
- }
-
- private Mono<Void> handleExpungedMessage(MessageMetaData expungedMessageMetaData,
- MailboxId movedFromMailboxId,
- MailboxId movedToMailboxId,
- MailboxSession session) {
- return newIndexByIndexedDocument(expungedMessageMetaData.getUid(), movedFromMailboxId, movedToMailboxId, session, expungedMessageMetaData.getMessageId())
- .onErrorResume(OpenSearchNotFoundException.class,
- notFoundException -> {
- LOGGER.warn("Can not find message {} in mailbox {} for reindexing",
- expungedMessageMetaData.getUid(),
- movedFromMailboxId.serialize());
- return notFoundHandleFallBack(expungedMessageMetaData, movedToMailboxId, session);
- });
- }
-
- private Mono<Void> notFoundHandleFallBack(MessageMetaData expungedMessageMetaData, MailboxId movedToMailboxId, MailboxSession session) {
- return retrieveMailboxMessage(session, expungedMessageMetaData.getMessageId(), movedToMailboxId, FetchType.FULL)
- .publishOn(Schedulers.parallel())
- .flatMap(mailboxMessage -> factory.getMailboxMapper(session)
- .findMailboxById(movedToMailboxId)
- .flatMap(mailbox -> add(session, mailbox, mailboxMessage)));
- }
-
- private Mono<Void> newIndexByIndexedDocument(MessageUid oldMessageUid,
- MailboxId movedFromMailboxId,
- MailboxId movedToMailboxId,
- MailboxSession session,
- MessageId messageId) {
- RoutingKey movedFromRoutingKey = routingKeyFactory.from(movedFromMailboxId);
- return openSearchIndexer.get(indexIdFor(movedFromMailboxId, oldMessageUid), movedFromRoutingKey)
- .filter(GetResponse::found)
- .switchIfEmpty(Mono.error(() -> new OpenSearchNotFoundException("Can not find message " + oldMessageUid + " in mailbox " + movedFromMailboxId.serialize())))
- .mapNotNull(GetResponse::source)
- .flatMap(document -> updateDocumentThenIndex(messageId, movedToMailboxId, document, session));
- }
-
- private Mono<Void> updateDocumentThenIndex(MessageId messageId,
- MailboxId movedToMailboxId,
- ObjectNode origin,
- MailboxSession session) {
- return retrieveMailboxMessage(session, messageId, movedToMailboxId, FetchType.METADATA)
- .flatMap(mailboxMessage -> Mono.fromCallable(() -> {
- messageToOpenSearchJson.updateMessageUid(origin, mailboxMessage.getUid());
- messageToOpenSearchJson.updateMailboxId(origin, movedToMailboxId);
- mailboxMessage.getSaveDate().ifPresent(newSaveDate -> messageToOpenSearchJson.updateSaveDate(origin, newSaveDate));
- return origin;
- }).map(messageToOpenSearchJson::toString)
- .flatMap(jsonContent -> add(movedToMailboxId, mailboxMessage.getUid(), jsonContent)));
- }
-
- private Mono<MailboxMessage> retrieveMailboxMessage(MailboxSession session,
- MessageId messageId,
- MailboxId movedToMailboxId,
- FetchType fetchType) {
- return factory.getMessageIdMapper(session)
- .findReactive(List.of(messageId), fetchType)
- .filter(mailboxMessage -> mailboxMessage.getMailboxId().equals(movedToMailboxId))
- .next();
- }
-
- private Mono<Void> handleAddedEvent(MailboxSession session,
- MailboxEvents.Added addedEvent,
- MailboxId mailboxId) {
- if (addedEvent.isMoved()) {
- return Mono.empty();
- } else {
- return factory.getMailboxMapper(session)
- .findMailboxById(mailboxId)
- .flatMap(mailbox -> handleAdded(session, mailbox, addedEvent));
- }
+ private Mono<Void> processAddedEvent(MailboxSession session, MailboxEvents.Added addedEvent, MailboxId mailboxId) {
+ return factory.getMailboxMapper(session)
+ .findMailboxById(mailboxId)
+ .flatMap(mailbox -> handleAdded(session, mailbox, addedEvent));
}
@Override
@@ -354,19 +399,13 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
return DocumentId.fromString(mailboxId.serialize() + ID_SEPARATOR + uid.asLong());
}
- private void logIfNoMessageId(SearchResult searchResult) {
- if (!searchResult.getMessageId().isPresent()) {
- LOGGER.error("No messageUid for {} in mailbox {}", searchResult.getMessageUid(), searchResult.getMailboxId());
- }
- }
-
@Override
public Mono<Flags> retrieveIndexedFlags(Mailbox mailbox, MessageUid uid) {
RoutingKey routingKey = routingKeyFactory.from(mailbox.getMailboxId());
return openSearchIndexer.get(indexIdFor(mailbox.getMailboxId(), uid), routingKey)
.filter(GetResponse::found)
- .map(GetResponse::source)
+ .mapNotNull(GetResponse::source)
.map(this::extractFlags);
}
diff --git a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java
index 43414d7d67..8eece0cced 100644
--- a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java
+++ b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java
@@ -124,6 +124,12 @@ class OpenSearchIntegrationTest extends AbstractMessageSearchIndexTest {
awaitForOpenSearch(queryConverter.from(mailboxIds, query), messageCount);
}
+ protected OpenSearchMailboxConfiguration openSearchMailboxConfiguration() {
+ return OpenSearchMailboxConfiguration.builder()
+ .optimiseMoves(false)
+ .build();
+ }
+
@Override
protected void initializeMailboxManager() {
messageIdFactory = new InMemoryMessageId.Factory();
@@ -144,7 +150,8 @@ class OpenSearchIntegrationTest extends AbstractMessageSearchIndexTest {
new OpenSearchSearcher(client, new QueryConverter(new CriterionConverter()), SEARCH_SIZE,
MailboxOpenSearchConstants.DEFAULT_MAILBOX_READ_ALIAS, routingKeyFactory),
new MessageToOpenSearchJson(textExtractor, ZoneId.of("Europe/Paris"), IndexAttachments.YES, IndexHeaders.YES),
- preInstanciationStage.getSessionProvider(), routingKeyFactory, messageIdFactory))
+ preInstanciationStage.getSessionProvider(), routingKeyFactory, messageIdFactory,
+ openSearchMailboxConfiguration(), new RecordingMetricFactory()))
.noPreDeletionHooks()
.storeQuotaManager()
.build();
diff --git a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchOptimizeMoveIntegrationTest.java b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchOptimizeMoveIntegrationTest.java
new file mode 100644
index 0000000000..0fce784109
--- /dev/null
+++ b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchOptimizeMoveIntegrationTest.java
@@ -0,0 +1,30 @@
+/****************************************************************
+ * 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.opensearch;
+
+class OpenSearchOptimizeMoveIntegrationTest extends OpenSearchIntegrationTest {
+
+ @Override
+ protected OpenSearchMailboxConfiguration openSearchMailboxConfiguration() {
+ return OpenSearchMailboxConfiguration.builder()
+ .optimiseMoves(true)
+ .build();
+ }
+}
\ No newline at end of file
diff --git a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndexTest.java b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndexTest.java
index 5575b3b51d..7d346c0bc0 100644
--- a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndexTest.java
+++ b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndexTest.java
@@ -73,6 +73,7 @@ import org.apache.james.mailbox.opensearch.IndexHeaders;
import org.apache.james.mailbox.opensearch.MailboxIdRoutingKeyFactory;
import org.apache.james.mailbox.opensearch.MailboxIndexCreationUtil;
import org.apache.james.mailbox.opensearch.MailboxOpenSearchConstants;
+import org.apache.james.mailbox.opensearch.OpenSearchMailboxConfiguration;
import org.apache.james.mailbox.opensearch.json.MessageToOpenSearchJson;
import org.apache.james.mailbox.opensearch.query.CriterionConverter;
import org.apache.james.mailbox.opensearch.query.QueryConverter;
@@ -86,6 +87,7 @@ import org.apache.james.mailbox.store.mail.model.impl.PropertyBuilder;
import org.apache.james.mailbox.store.mail.model.impl.SimpleMailboxMessage;
import org.apache.james.mailbox.store.search.ListeningMessageSearchIndex;
import org.apache.james.mailbox.store.search.ListeningMessageSearchIndexContract;
+import org.apache.james.metrics.tests.RecordingMetricFactory;
import org.apache.james.utils.UpdatableTickingClock;
import org.awaitility.Awaitility;
import org.awaitility.Durations;
@@ -215,7 +217,8 @@ class OpenSearchListeningMessageSearchIndexTest {
testee = new OpenSearchListeningMessageSearchIndex(mapperFactory,
ImmutableSet.of(), openSearchIndexer, openSearchSearcher,
- messageToOpenSearchJson, sessionProvider, new MailboxIdRoutingKeyFactory(), messageIdFactory);
+ messageToOpenSearchJson, sessionProvider, new MailboxIdRoutingKeyFactory(), messageIdFactory,
+ OpenSearchMailboxConfiguration.builder().build(), new RecordingMetricFactory());
session = sessionProvider.createSystemSession(USERNAME);
mailbox = mapperFactory.getMailboxMapper(session).create(MailboxPath.forUser(USERNAME, DefaultMailboxes.INBOX), UidValidity.generate()).block();
@@ -281,7 +284,8 @@ class OpenSearchListeningMessageSearchIndexTest {
testee = new OpenSearchListeningMessageSearchIndex(mapperFactory,
ImmutableSet.of(), openSearchIndexer, openSearchSearcher,
- messageToOpenSearchJson, sessionProvider, new MailboxIdRoutingKeyFactory(), new InMemoryMessageId.Factory());
+ messageToOpenSearchJson, sessionProvider, new MailboxIdRoutingKeyFactory(), new InMemoryMessageId.Factory(),
+ OpenSearchMailboxConfiguration.builder().build(), new RecordingMetricFactory());
testee.add(session, mailbox, MESSAGE_WITH_ATTACHMENT).block();
awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 1L);
diff --git a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/search/OpenSearchSearcherTest.java b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/search/OpenSearchSearcherTest.java
index 6e6686df78..ec7401fad6 100644
--- a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/search/OpenSearchSearcherTest.java
+++ b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/search/OpenSearchSearcherTest.java
@@ -50,6 +50,7 @@ import org.apache.james.mailbox.opensearch.IndexHeaders;
import org.apache.james.mailbox.opensearch.MailboxIdRoutingKeyFactory;
import org.apache.james.mailbox.opensearch.MailboxIndexCreationUtil;
import org.apache.james.mailbox.opensearch.MailboxOpenSearchConstants;
+import org.apache.james.mailbox.opensearch.OpenSearchMailboxConfiguration;
import org.apache.james.mailbox.opensearch.events.OpenSearchListeningMessageSearchIndex;
import org.apache.james.mailbox.opensearch.json.MessageToOpenSearchJson;
import org.apache.james.mailbox.opensearch.query.CriterionConverter;
@@ -125,7 +126,8 @@ class OpenSearchSearcherTest {
new OpenSearchSearcher(client, new QueryConverter(new CriterionConverter()), SEARCH_SIZE,
MailboxOpenSearchConstants.DEFAULT_MAILBOX_READ_ALIAS, routingKeyFactory),
new MessageToOpenSearchJson(textExtractor, ZoneId.of("Europe/Paris"), IndexAttachments.YES, IndexHeaders.YES),
- preInstanciationStage.getSessionProvider(), routingKeyFactory, messageIdFactory))
+ preInstanciationStage.getSessionProvider(), routingKeyFactory, messageIdFactory,
+ OpenSearchMailboxConfiguration.builder().build(), new RecordingMetricFactory()))
.noPreDeletionHooks()
.storeQuotaManager()
.build();
diff --git a/mpt/impl/imap-mailbox/opensearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/OpenSearchHostSystem.java b/mpt/impl/imap-mailbox/opensearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/OpenSearchHostSystem.java
index 9808814988..8ce82167ac 100644
--- a/mpt/impl/imap-mailbox/opensearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/OpenSearchHostSystem.java
+++ b/mpt/impl/imap-mailbox/opensearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/OpenSearchHostSystem.java
@@ -42,6 +42,7 @@ import org.apache.james.mailbox.opensearch.IndexHeaders;
import org.apache.james.mailbox.opensearch.MailboxIdRoutingKeyFactory;
import org.apache.james.mailbox.opensearch.MailboxIndexCreationUtil;
import org.apache.james.mailbox.opensearch.MailboxOpenSearchConstants;
+import org.apache.james.mailbox.opensearch.OpenSearchMailboxConfiguration;
import org.apache.james.mailbox.opensearch.events.OpenSearchListeningMessageSearchIndex;
import org.apache.james.mailbox.opensearch.json.MessageToOpenSearchJson;
import org.apache.james.mailbox.opensearch.query.CriterionConverter;
@@ -52,6 +53,7 @@ import org.apache.james.mailbox.store.StoreSubscriptionManager;
import org.apache.james.mailbox.store.extractor.DefaultTextExtractor;
import org.apache.james.mailbox.store.quota.NoQuotaManager;
import org.apache.james.metrics.logger.DefaultMetricFactory;
+import org.apache.james.metrics.tests.RecordingMetricFactory;
import org.apache.james.mpt.api.ImapFeatures;
import org.apache.james.mpt.api.ImapFeatures.Feature;
import org.apache.james.mpt.host.JamesImapHostSystem;
@@ -105,7 +107,8 @@ public class OpenSearchHostSystem extends JamesImapHostSystem {
new OpenSearchSearcher(client, new QueryConverter(new CriterionConverter()), OpenSearchSearcher.DEFAULT_SEARCH_SIZE,
MailboxOpenSearchConstants.DEFAULT_MAILBOX_READ_ALIAS, routingKeyFactory),
new MessageToOpenSearchJson(new DefaultTextExtractor(), ZoneId.of("Europe/Paris"), IndexAttachments.YES, IndexHeaders.YES),
- preInstanciationStage.getSessionProvider(), routingKeyFactory, messageIdFactory))
+ preInstanciationStage.getSessionProvider(), routingKeyFactory, messageIdFactory,
+ OpenSearchMailboxConfiguration.builder().build(), new RecordingMetricFactory()))
.noPreDeletionHooks()
.storeQuotaManager()
.build();
diff --git a/server/apps/distributed-app/docs/modules/ROOT/pages/configure/opensearch.adoc b/server/apps/distributed-app/docs/modules/ROOT/pages/configure/opensearch.adoc
index 85b289e8c8..e4b7d32611 100644
--- a/server/apps/distributed-app/docs/modules/ROOT/pages/configure/opensearch.adoc
+++ b/server/apps/distributed-app/docs/modules/ROOT/pages/configure/opensearch.adoc
@@ -91,6 +91,11 @@ The target of the alias is the index name configured above.
(From, To, Cc, Bcc, Subject, Message-Id, Date, Content-Type) are still indexed in their dedicated type.
Header indexing is expensive as each header currently need to be stored as a nested document but
turning off headers indexing result in non-strict compliance with the IMAP / JMAP standards.
+
+| opensearch.message.index.optimize.move
+| When set to true, James will attempt to reindex from the indexed message when moved.
+If the message is not found, it will fall back to the old behavior (The message will be indexed from the blobStore source)
+Default to false.
|===
=== Quota search
diff --git a/server/apps/distributed-app/sample-configuration/opensearch.properties b/server/apps/distributed-app/sample-configuration/opensearch.properties
index 87011f9f5c..209c08be1e 100644
--- a/server/apps/distributed-app/sample-configuration/opensearch.properties
+++ b/server/apps/distributed-app/sample-configuration/opensearch.properties
@@ -94,3 +94,7 @@ opensearch.indexAttachments=true
# Please note that custom overrides can be defined here.
#
# opensearch.search.overrides=org.apache.james.mailbox.cassandra.search.AllSearchOverride,org.apache.james.mailbox.cassandra.search.DeletedSearchOverride, org.apache.james.mailbox.cassandra.search.DeletedWithRangeSearchOverride,org.apache.james.mailbox.cassandra.search.NotDeletedWithRangeSearchOverride,org.apache.james.mailbox.cassandra.search.UidSearchOverride,org.apache.james.mailbox.cassandra.search.UnseenSearchOverride
+
+# Optional. Default is `false`
+# When set to true, James will attempt to reindex from the indexed message when moved. If the message is not found, it will fall back to the old behavior (The message will be indexed from the blobStore source)
+# opensearch.message.index.optimize.move=false
\ No newline at end of file
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/MailboxesRoutesTest.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/MailboxesRoutesTest.java
index 376271045d..cea305d951 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/MailboxesRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/MailboxesRoutesTest.java
@@ -71,6 +71,7 @@ import org.apache.james.mailbox.opensearch.IndexHeaders;
import org.apache.james.mailbox.opensearch.MailboxIdRoutingKeyFactory;
import org.apache.james.mailbox.opensearch.MailboxIndexCreationUtil;
import org.apache.james.mailbox.opensearch.MailboxOpenSearchConstants;
+import org.apache.james.mailbox.opensearch.OpenSearchMailboxConfiguration;
import org.apache.james.mailbox.opensearch.events.OpenSearchListeningMessageSearchIndex;
import org.apache.james.mailbox.opensearch.json.MessageToOpenSearchJson;
import org.apache.james.mailbox.opensearch.query.CriterionConverter;
@@ -81,6 +82,7 @@ import org.apache.james.mailbox.store.mail.model.MailboxMessage;
import org.apache.james.mailbox.store.mail.model.impl.PropertyBuilder;
import org.apache.james.mailbox.store.mail.model.impl.SimpleMailboxMessage;
import org.apache.james.mailbox.store.search.ListeningMessageSearchIndex;
+import org.apache.james.metrics.tests.RecordingMetricFactory;
import org.apache.james.task.Hostname;
import org.apache.james.task.MemoryTaskManager;
import org.apache.james.webadmin.WebAdminServer;
@@ -163,7 +165,7 @@ class MailboxesRoutesTest {
new OpenSearchSearcher(client, new QueryConverter(new CriterionConverter()), SEARCH_SIZE,
MailboxOpenSearchConstants.DEFAULT_MAILBOX_READ_ALIAS, routingKeyFactory),
new MessageToOpenSearchJson(new DefaultTextExtractor(), ZoneId.of("Europe/Paris"), IndexAttachments.YES, IndexHeaders.YES),
- preInstanciationStage.getSessionProvider(), routingKeyFactory, messageIdFactory))
+ preInstanciationStage.getSessionProvider(), routingKeyFactory, messageIdFactory, OpenSearchMailboxConfiguration.builder().build(), new RecordingMetricFactory()))
.noPreDeletionHooks()
.storeQuotaManager()
.build();
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/UserMailboxesRoutesTest.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/UserMailboxesRoutesTest.java
index 6640703eb8..4a1aecd0a7 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/UserMailboxesRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/UserMailboxesRoutesTest.java
@@ -95,6 +95,7 @@ import org.apache.james.mailbox.opensearch.IndexHeaders;
import org.apache.james.mailbox.opensearch.MailboxIdRoutingKeyFactory;
import org.apache.james.mailbox.opensearch.MailboxIndexCreationUtil;
import org.apache.james.mailbox.opensearch.MailboxOpenSearchConstants;
+import org.apache.james.mailbox.opensearch.OpenSearchMailboxConfiguration;
import org.apache.james.mailbox.opensearch.events.OpenSearchListeningMessageSearchIndex;
import org.apache.james.mailbox.opensearch.json.MessageToOpenSearchJson;
import org.apache.james.mailbox.opensearch.query.CriterionConverter;
@@ -106,6 +107,7 @@ import org.apache.james.mailbox.store.mail.model.MailboxMessage;
import org.apache.james.mailbox.store.mail.model.impl.PropertyBuilder;
import org.apache.james.mailbox.store.mail.model.impl.SimpleMailboxMessage;
import org.apache.james.mailbox.store.search.ListeningMessageSearchIndex;
+import org.apache.james.metrics.tests.RecordingMetricFactory;
import org.apache.james.task.Hostname;
import org.apache.james.task.MemoryTaskManager;
import org.apache.james.user.api.UsersRepository;
@@ -1519,7 +1521,7 @@ class UserMailboxesRoutesTest {
new OpenSearchSearcher(client, new QueryConverter(new CriterionConverter()), SEARCH_SIZE,
MailboxOpenSearchConstants.DEFAULT_MAILBOX_READ_ALIAS, routingKeyFactory),
new MessageToOpenSearchJson(new DefaultTextExtractor(), ZoneId.of("Europe/Paris"), IndexAttachments.YES, IndexHeaders.YES),
- preInstanciationStage.getSessionProvider(), routingKeyFactory, messageIdFactory))
+ preInstanciationStage.getSessionProvider(), routingKeyFactory, messageIdFactory, OpenSearchMailboxConfiguration.builder().build(), new RecordingMetricFactory()))
.noPreDeletionHooks()
.storeQuotaManager()
.build();
diff --git a/src/site/xdoc/server/config-opensearch.xml b/src/site/xdoc/server/config-opensearch.xml
index bea64dc242..5983902f1c 100644
--- a/src/site/xdoc/server/config-opensearch.xml
+++ b/src/site/xdoc/server/config-opensearch.xml
@@ -118,6 +118,10 @@
(From, To, Cc, Bcc, Subject, Message-Id, Date, Content-Type) are still indexed in their dedicated type.
Header indexing is expensive as each header currently need to be stored as a nested document but
turning off headers indexing result in non-strict compliance with the IMAP / JMAP standards.</dd>
+ <dt><strong>opensearch.message.index.optimize.move</strong></dt>
+ <dd>When set to true, James will attempt to reindex from the indexed message when moved.
+ If the message is not found, it will fall back to the old behavior (The message will be indexed from the blobStore source).
+ Default to false.</dd>
<dt><strong>opensearch.index.quota.ratio.name</strong></dt>
<dd>Specify the OpenSearch alias name used for quotas</dd>
<dt><strong>opensearch.alias.read.quota.ratio.name</strong></dt>
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org