james-server-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From adup...@apache.org
Subject [20/41] james-project git commit: JAMES-2082 Migration from V1 to V2 should be run asynchronously on separated thread
Date Mon, 10 Jul 2017 17:54:32 GMT
JAMES-2082 Migration from V1 to V2 should be run asynchronously on separated thread


Project: http://git-wip-us.apache.org/repos/asf/james-project/repo
Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/eb55762a
Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/eb55762a
Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/eb55762a

Branch: refs/heads/master
Commit: eb55762a3c91998c11720c1f8535913230641215
Parents: 17ed1d4
Author: benwa <btellier@linagora.com>
Authored: Fri Jul 7 09:39:29 2017 +0700
Committer: Antoine Duprat <aduprat@linagora.com>
Committed: Mon Jul 10 14:23:57 2017 +0200

----------------------------------------------------------------------
 mailbox/cassandra/pom.xml                       |   6 +
 .../CassandraMailboxSessionMapperFactory.java   |  12 +-
 .../mail/CassandraMessageIdMapper.java          |   4 +-
 .../cassandra/mail/CassandraMessageMapper.java  |  14 +--
 .../mail/migration/V1ToV2Migration.java         |  67 +++++------
 .../mail/migration/V1ToV2MigrationThread.java   | 118 +++++++++++++++++++
 .../mail/migration/V1ToV2MigrationTest.java     |  61 +++++++---
 7 files changed, 216 insertions(+), 66 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/james-project/blob/eb55762a/mailbox/cassandra/pom.xml
----------------------------------------------------------------------
diff --git a/mailbox/cassandra/pom.xml b/mailbox/cassandra/pom.xml
index 91bbd95..1e39812 100644
--- a/mailbox/cassandra/pom.xml
+++ b/mailbox/cassandra/pom.xml
@@ -260,6 +260,12 @@
                     <artifactId>throwing-lambdas</artifactId>
                     <version>0.5.0</version>
                 </dependency>
+                <dependency>
+                    <groupId>com.jayway.awaitility</groupId>
+                    <artifactId>awaitility</artifactId>
+                    <version>1.6.5</version>
+                    <scope>test</scope>
+                </dependency>
             </dependencies>
             <build>
                 <plugins>

http://git-wip-us.apache.org/repos/asf/james-project/blob/eb55762a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
----------------------------------------------------------------------
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
index 4ddd2db..cabd043 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
@@ -42,6 +42,7 @@ import org.apache.james.mailbox.cassandra.mail.CassandraMessageMapper;
 import org.apache.james.mailbox.cassandra.mail.CassandraModSeqProvider;
 import org.apache.james.mailbox.cassandra.mail.CassandraUidProvider;
 import org.apache.james.mailbox.cassandra.mail.*;
+import org.apache.james.mailbox.cassandra.mail.migration.V1ToV2Migration;
 import org.apache.james.mailbox.cassandra.user.CassandraSubscriptionMapper;
 import org.apache.james.mailbox.exception.MailboxException;
 import org.apache.james.mailbox.store.MailboxSessionMapperFactory;
@@ -74,6 +75,7 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
     private final CassandraMailboxPathDAO mailboxPathDAO;
     private final CassandraFirstUnseenDAO firstUnseenDAO;
     private final CassandraApplicableFlagDAO applicableFlagDAO;
+    private final V1ToV2Migration v1ToV2Migration;
     private CassandraUtils cassandraUtils;
     private CassandraConfiguration cassandraConfiguration;
     private final CassandraDeletedMessageDAO deletedMessageDAO;
@@ -84,7 +86,7 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
                                                 CassandraMessageIdDAO messageIdDAO, CassandraMessageIdToImapUidDAO
imapUidDAO,
                                                 CassandraMailboxCounterDAO mailboxCounterDAO,
CassandraMailboxRecentsDAO mailboxRecentsDAO, CassandraMailboxDAO mailboxDAO,
                                                 CassandraMailboxPathDAO mailboxPathDAO, CassandraFirstUnseenDAO
firstUnseenDAO, CassandraApplicableFlagDAO applicableFlagDAO,
-                                                CassandraDeletedMessageDAO deletedMessageDAO,
CassandraUtils cassandraUtils, CassandraConfiguration cassandraConfiguration) {
+                                                CassandraDeletedMessageDAO deletedMessageDAO,
V1ToV2Migration v1ToV2Migration, CassandraUtils cassandraUtils, CassandraConfiguration cassandraConfiguration)
{
         this.uidProvider = uidProvider;
         this.modSeqProvider = modSeqProvider;
         this.session = session;
@@ -107,6 +109,7 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
             firstUnseenDAO,
             applicableFlagDAO,
             deletedMessageDAO);
+        this.v1ToV2Migration = v1ToV2Migration;
     }
 
     public CassandraMailboxSessionMapperFactory(
@@ -126,7 +129,8 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
         CassandraDeletedMessageDAO deletedMesageDAO) {
 
         this(uidProvider, modSeqProvider, session, messageDAO, messageDAOV2, messageIdDAO,
imapUidDAO, mailboxCounterDAO,
-             mailboxRecentsDAO, mailboxDAO, mailboxPathDAO, firstUnseenDAO, applicableFlagDAO,
deletedMesageDAO,
+            mailboxRecentsDAO, mailboxDAO, mailboxPathDAO, firstUnseenDAO, applicableFlagDAO,
deletedMesageDAO,
+            new V1ToV2Migration(messageDAO, messageDAOV2, new CassandraAttachmentMapper(session),
CassandraConfiguration.DEFAULT_CONFIGURATION),
             CassandraUtils.WITH_DEFAULT_CONFIGURATION, CassandraConfiguration.DEFAULT_CONFIGURATION);
     }
 
@@ -137,7 +141,6 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
                                           modSeqProvider,
                                           null,
                                           (CassandraAttachmentMapper) createAttachmentMapper(mailboxSession),
-                                          messageDAO,
                                           messageDAOV2,
                                           messageIdDAO,
                                           imapUidDAO,
@@ -147,6 +150,7 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
                                           indexTableHandler,
                                           firstUnseenDAO,
                                           deletedMessageDAO,
+                                          v1ToV2Migration,
                                           cassandraConfiguration);
     }
 
@@ -155,7 +159,7 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
         return new CassandraMessageIdMapper(getMailboxMapper(mailboxSession), mailboxDAO,
                 (CassandraAttachmentMapper) getAttachmentMapper(mailboxSession),
                 imapUidDAO, messageIdDAO, messageDAO, messageDAOV2, indexTableHandler, modSeqProvider,
mailboxSession,
-                cassandraConfiguration);
+                v1ToV2Migration, cassandraConfiguration);
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/james-project/blob/eb55762a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
----------------------------------------------------------------------
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
index 479f43f..bf43a76 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
@@ -77,7 +77,7 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
                                     CassandraMessageIdToImapUidDAO imapUidDAO, CassandraMessageIdDAO
messageIdDAO,
                                     CassandraMessageDAO messageDAOV1, CassandraMessageDAOV2
messageDAOV2,
                                     CassandraIndexTableHandler indexTableHandler, ModSeqProvider
modSeqProvider, MailboxSession mailboxSession,
-                                    CassandraConfiguration cassandraConfiguration) {
+                                    V1ToV2Migration v1ToV2Migration, CassandraConfiguration
cassandraConfiguration) {
 
         this.mailboxMapper = mailboxMapper;
         this.mailboxDAO = mailboxDAO;
@@ -89,7 +89,7 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
         this.mailboxSession = mailboxSession;
         this.attachmentLoader = new AttachmentLoader(attachmentMapper);
         this.cassandraConfiguration = cassandraConfiguration;
-        this.v1ToV2Migration = new V1ToV2Migration(messageDAOV1, messageDAOV2, attachmentMapper,
cassandraConfiguration);
+        this.v1ToV2Migration = v1ToV2Migration;
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/james-project/blob/eb55762a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
----------------------------------------------------------------------
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
index 3ed7842..b34f03c 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
@@ -87,14 +87,14 @@ public class CassandraMessageMapper implements MessageMapper {
     private final V1ToV2Migration v1ToV2Migration;
     private final CassandraConfiguration cassandraConfiguration;
 
-
-
     public CassandraMessageMapper(CassandraUidProvider uidProvider, CassandraModSeqProvider
modSeqProvider,
                                   MailboxSession mailboxSession, CassandraAttachmentMapper
attachmentMapper,
-                                  CassandraMessageDAO messageDAO, CassandraMessageDAOV2 messageDAOV2,
-                                  CassandraMessageIdDAO messageIdDAO, CassandraMessageIdToImapUidDAO
imapUidDAO,
-                                  CassandraMailboxCounterDAO mailboxCounterDAO, CassandraMailboxRecentsDAO
mailboxRecentDAO, CassandraApplicableFlagDAO applicableFlagDAO,
-                                  CassandraIndexTableHandler indexTableHandler, CassandraFirstUnseenDAO
firstUnseenDAO, CassandraDeletedMessageDAO deletedMessageDAO, CassandraConfiguration cassandraConfiguration)
{
+                                  CassandraMessageDAOV2 messageDAOV2, CassandraMessageIdDAO
messageIdDAO,
+                                  CassandraMessageIdToImapUidDAO imapUidDAO, CassandraMailboxCounterDAO
mailboxCounterDAO,
+                                  CassandraMailboxRecentsDAO mailboxRecentDAO, CassandraApplicableFlagDAO
applicableFlagDAO,
+                                  CassandraIndexTableHandler indexTableHandler, CassandraFirstUnseenDAO
firstUnseenDAO,
+                                  CassandraDeletedMessageDAO deletedMessageDAO, V1ToV2Migration
v1ToV2Migration,
+                                  CassandraConfiguration cassandraConfiguration) {
         this.uidProvider = uidProvider;
         this.modSeqProvider = modSeqProvider;
         this.mailboxSession = mailboxSession;
@@ -108,7 +108,7 @@ public class CassandraMessageMapper implements MessageMapper {
         this.attachmentLoader = new AttachmentLoader(attachmentMapper);
         this.applicableFlagDAO = applicableFlagDAO;
         this.deletedMessageDAO = deletedMessageDAO;
-        this.v1ToV2Migration = new V1ToV2Migration(messageDAO, messageDAOV2, attachmentMapper,
cassandraConfiguration);
+        this.v1ToV2Migration = v1ToV2Migration;
         this.cassandraConfiguration = cassandraConfiguration;
     }
 

http://git-wip-us.apache.org/repos/asf/james-project/blob/eb55762a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2Migration.java
----------------------------------------------------------------------
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2Migration.java
b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2Migration.java
index 0a4463e..131addb 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2Migration.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2Migration.java
@@ -19,13 +19,17 @@
 
 package org.apache.james.mailbox.cassandra.mail.migration;
 
-import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
+import javax.annotation.PreDestroy;
+import javax.inject.Inject;
+
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.james.backends.cassandra.CassandraConfiguration;
-import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
 import org.apache.james.mailbox.cassandra.mail.AttachmentLoader;
 import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentMapper;
 import org.apache.james.mailbox.cassandra.mail.CassandraMessageDAO;
@@ -33,28 +37,37 @@ import org.apache.james.mailbox.cassandra.mail.CassandraMessageDAOV2;
 import org.apache.james.mailbox.cassandra.mail.MessageAttachmentRepresentation;
 import org.apache.james.mailbox.cassandra.mail.MessageWithoutAttachment;
 import org.apache.james.mailbox.cassandra.mail.utils.Limit;
-import org.apache.james.mailbox.exception.MailboxException;
 import org.apache.james.mailbox.store.mail.MessageMapper;
-import org.apache.james.mailbox.store.mail.model.impl.SimpleMailboxMessage;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
+import com.google.common.collect.EvictingQueue;
 import com.google.common.collect.ImmutableList;
 
 public class V1ToV2Migration {
-    private static final Logger LOGGER = LoggerFactory.getLogger(V1ToV2Migration.class);
+    private static final int MIGRATION_THREAD_COUNT = 2;
+    private static final int MIGRATION_QUEUE_LENGTH = 1000;
 
     private final CassandraMessageDAO messageDAOV1;
-    private final CassandraMessageDAOV2 messageDAOV2;
     private final AttachmentLoader attachmentLoader;
     private final CassandraConfiguration cassandraConfiguration;
+    private final ExecutorService migrationExecutor;
+    private final EvictingQueue<Pair<MessageWithoutAttachment, Stream<MessageAttachmentRepresentation>>>
messagesToBeMigrated;
 
+    @Inject
     public V1ToV2Migration(CassandraMessageDAO messageDAOV1, CassandraMessageDAOV2 messageDAOV2,
                            CassandraAttachmentMapper attachmentMapper, CassandraConfiguration
cassandraConfiguration) {
         this.messageDAOV1 = messageDAOV1;
-        this.messageDAOV2 = messageDAOV2;
         this.attachmentLoader = new AttachmentLoader(attachmentMapper);
         this.cassandraConfiguration = cassandraConfiguration;
+        this.migrationExecutor = Executors.newFixedThreadPool(MIGRATION_THREAD_COUNT);
+        this.messagesToBeMigrated = EvictingQueue.create(MIGRATION_QUEUE_LENGTH);
+        IntStream.range(0, MIGRATION_THREAD_COUNT)
+            .mapToObj(i -> new V1ToV2MigrationThread(messagesToBeMigrated, messageDAOV1,
messageDAOV2, attachmentLoader))
+            .forEach(migrationExecutor::execute);
+    }
+
+    @PreDestroy
+    public void stop() {
+        migrationExecutor.shutdownNow();
     }
 
     public CompletableFuture<Pair<MessageWithoutAttachment, Stream<MessageAttachmentRepresentation>>>
@@ -67,37 +80,15 @@ public class V1ToV2Migration {
         return messageDAOV1.retrieveMessages(ImmutableList.of(result.getMetadata()), MessageMapper.FetchType.Full,
Limit.unlimited())
             .thenApply(results -> results.findAny()
                 .orElseThrow(() -> new IllegalArgumentException("Message not found in
DAO V1" + result.getMetadata())))
-            .thenCompose(this::performV1ToV2Migration);
-    }
-
-    private CompletableFuture<Pair<MessageWithoutAttachment, Stream<MessageAttachmentRepresentation>>>
performV1ToV2Migration(Pair<MessageWithoutAttachment, Stream<MessageAttachmentRepresentation>>
messageV1) {
-        return attachmentLoader.addAttachmentToMessages(Stream.of(messageV1), MessageMapper.FetchType.Full)
-            .thenApply(stream -> stream.findAny().get())
-            .thenCompose(this::performV1ToV2Migration)
-            .thenApply(any -> messageV1);
-    }
-
-    private CompletableFuture<Void> performV1ToV2Migration(SimpleMailboxMessage message)
{
-        if (!cassandraConfiguration.isOnTheFlyV1ToV2Migration()) {
-            return CompletableFuture.completedFuture(null);
-        }
-        return saveInV2FromV1(message)
-            .thenCompose(this::deleteInV1);
-    }
-
-    private CompletableFuture<Void> deleteInV1(Optional<SimpleMailboxMessage>
optional) {
-        return optional.map(SimpleMailboxMessage::getMessageId)
-            .map(messageId -> (CassandraMessageId) messageId)
-            .map(messageDAOV1::delete)
-            .orElse(CompletableFuture.completedFuture(null));
+            .thenApply(this::submitMigration);
     }
 
-    private CompletableFuture<Optional<SimpleMailboxMessage>> saveInV2FromV1(SimpleMailboxMessage
message) {
-        try {
-            return messageDAOV2.save(message).thenApply(any -> Optional.of(message));
-        } catch (MailboxException e) {
-            LOGGER.error("Exception while saving message during migration", e);
-            return CompletableFuture.completedFuture(Optional.<SimpleMailboxMessage>empty());
+    private Pair<MessageWithoutAttachment, Stream<MessageAttachmentRepresentation>>
submitMigration(Pair<MessageWithoutAttachment, Stream<MessageAttachmentRepresentation>>
messageV1) {
+        if (cassandraConfiguration.isOnTheFlyV1ToV2Migration()) {
+            synchronized (messagesToBeMigrated) {
+                messagesToBeMigrated.add(messageV1);
+            }
         }
+        return messageV1;
     }
 }

http://git-wip-us.apache.org/repos/asf/james-project/blob/eb55762a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2MigrationThread.java
----------------------------------------------------------------------
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2MigrationThread.java
b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2MigrationThread.java
new file mode 100644
index 0000000..acac3a3
--- /dev/null
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2MigrationThread.java
@@ -0,0 +1,118 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one   *
+ * or more contributor license agreements.  See the NOTICE file *
+ * distributed with this work for additional information        *
+ * regarding copyright ownership.  The ASF licenses this file   *
+ * to you under the Apache License, Version 2.0 (the            *
+ * "License"); you may not use this file except in compliance   *
+ * with the License.  You may obtain a copy of the License at   *
+ *                                                              *
+ *   http://www.apache.org/licenses/LICENSE-2.0                 *
+ *                                                              *
+ * Unless required by applicable law or agreed to in writing,   *
+ * software distributed under the License is distributed on an  *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY       *
+ * KIND, either express or implied.  See the License for the    *
+ * specific language governing permissions and limitations      *
+ * under the License.                                           *
+ ****************************************************************/
+
+package org.apache.james.mailbox.cassandra.mail.migration;
+
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Stream;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
+import org.apache.james.mailbox.cassandra.mail.AttachmentLoader;
+import org.apache.james.mailbox.cassandra.mail.CassandraMessageDAO;
+import org.apache.james.mailbox.cassandra.mail.CassandraMessageDAOV2;
+import org.apache.james.mailbox.cassandra.mail.MessageAttachmentRepresentation;
+import org.apache.james.mailbox.cassandra.mail.MessageWithoutAttachment;
+import org.apache.james.mailbox.exception.MailboxException;
+import org.apache.james.mailbox.store.mail.MessageMapper;
+import org.apache.james.mailbox.store.mail.model.impl.SimpleMailboxMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Throwables;
+import com.google.common.collect.EvictingQueue;
+
+public class V1ToV2MigrationThread implements Runnable {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(V1ToV2MigrationThread.class);
+    public static final int POLL_INTERVAL_IN_MS = 10;
+
+    private final EvictingQueue<Pair<MessageWithoutAttachment, Stream<MessageAttachmentRepresentation>>>
messagesToBeMigrated;
+    private final CassandraMessageDAO messageDAOV1;
+    private final CassandraMessageDAOV2 messageDAOV2;
+    private final AttachmentLoader attachmentLoader;
+
+    public V1ToV2MigrationThread(EvictingQueue<Pair<MessageWithoutAttachment, Stream<MessageAttachmentRepresentation>>>
messagesToBeMigrated,
+                                 CassandraMessageDAO messageDAOV1, CassandraMessageDAOV2
messageDAOV2, AttachmentLoader attachmentLoader) {
+        this.messagesToBeMigrated = messagesToBeMigrated;
+        this.messageDAOV1 = messageDAOV1;
+        this.messageDAOV2 = messageDAOV2;
+        this.attachmentLoader = attachmentLoader;
+    }
+
+    @Override
+    public void run() {
+        while (true) {
+            try {
+                Pair<MessageWithoutAttachment, Stream<MessageAttachmentRepresentation>>
message = dequeue();
+                performV1ToV2Migration(message).join();
+            } catch (Exception e) {
+                LOGGER.error("Error occured in migration thread", e);
+            }
+        }
+    }
+
+    private Pair<MessageWithoutAttachment, Stream<MessageAttachmentRepresentation>>
dequeue() {
+        while (true) {
+            Optional<Pair<MessageWithoutAttachment, Stream<MessageAttachmentRepresentation>>>
poll = poll();
+            if (poll.isPresent()) {
+                return poll.get();
+            }
+            try {
+                Thread.sleep(POLL_INTERVAL_IN_MS);
+            } catch (InterruptedException e) {
+                Throwables.propagate(e);
+            }
+        }
+    }
+
+    private Optional<Pair<MessageWithoutAttachment, Stream<MessageAttachmentRepresentation>>>
poll() {
+        synchronized (messagesToBeMigrated) {
+            return Optional.ofNullable(messagesToBeMigrated.poll());
+        }
+    }
+
+    private CompletableFuture<Void> performV1ToV2Migration(Pair<MessageWithoutAttachment,
Stream<MessageAttachmentRepresentation>> messageV1) {
+        return attachmentLoader.addAttachmentToMessages(Stream.of(messageV1), MessageMapper.FetchType.Full)
+            .thenApply(stream -> stream.findAny().get())
+            .thenCompose(this::performV1ToV2Migration);
+    }
+
+    private CompletableFuture<Void> performV1ToV2Migration(SimpleMailboxMessage message)
{
+        return saveInV2FromV1(message)
+            .thenCompose(this::deleteInV1);
+    }
+
+    private CompletableFuture<Void> deleteInV1(Optional<SimpleMailboxMessage>
optional) {
+        return optional.map(SimpleMailboxMessage::getMessageId)
+            .map(messageId -> (CassandraMessageId) messageId)
+            .map(messageDAOV1::delete)
+            .orElse(CompletableFuture.completedFuture(null));
+    }
+
+    private CompletableFuture<Optional<SimpleMailboxMessage>> saveInV2FromV1(SimpleMailboxMessage
message) {
+        try {
+            return messageDAOV2.save(message).thenApply(any -> Optional.of(message));
+        } catch (MailboxException e) {
+            LOGGER.error("Exception while saving message during migration", e);
+            return CompletableFuture.completedFuture(Optional.<SimpleMailboxMessage>empty());
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/james-project/blob/eb55762a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2MigrationTest.java
----------------------------------------------------------------------
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2MigrationTest.java
b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2MigrationTest.java
index 96b2f95..57ec014 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2MigrationTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2MigrationTest.java
@@ -22,6 +22,8 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.Date;
 import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
 
 import javax.mail.Flags;
 import javax.mail.util.SharedByteArrayInputStream;
@@ -33,12 +35,12 @@ import org.apache.james.backends.cassandra.init.CassandraModuleComposite;
 import org.apache.james.mailbox.MessageUid;
 import org.apache.james.mailbox.cassandra.ids.CassandraId;
 import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
-import org.apache.james.mailbox.cassandra.mail.utils.Limit;
 import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentMapper;
 import org.apache.james.mailbox.cassandra.mail.CassandraBlobsDAO;
 import org.apache.james.mailbox.cassandra.mail.CassandraMessageDAO;
 import org.apache.james.mailbox.cassandra.mail.CassandraMessageDAOV2;
 import org.apache.james.mailbox.cassandra.mail.MessageAttachmentRepresentation;
+import org.apache.james.mailbox.cassandra.mail.utils.Limit;
 import org.apache.james.mailbox.cassandra.modules.CassandraAttachmentModule;
 import org.apache.james.mailbox.cassandra.modules.CassandraBlobModule;
 import org.apache.james.mailbox.cassandra.modules.CassandraMessageModule;
@@ -61,6 +63,9 @@ import org.junit.Test;
 
 import com.google.common.base.Charsets;
 import com.google.common.collect.ImmutableList;
+import com.jayway.awaitility.Awaitility;
+import com.jayway.awaitility.Duration;
+import com.jayway.awaitility.core.ConditionFactory;
 
 public class V1ToV2MigrationTest {
     private static final int BODY_START = 16;
@@ -85,6 +90,7 @@ public class V1ToV2MigrationTest {
     
     @Rule
     public final JUnitSoftAssertions softly = new JUnitSoftAssertions();
+    private ConditionFactory awaitability;
 
     @Before
     public void setUp() {
@@ -126,10 +132,18 @@ public class V1ToV2MigrationTest {
             .isInline(true)
             .name("toto.png")
             .build();
+
+        Duration slowPacedPollInterval = Duration.FIVE_HUNDRED_MILLISECONDS;
+        awaitability = Awaitility
+            .with()
+            .pollInterval(slowPacedPollInterval)
+            .and()
+            .pollDelay(slowPacedPollInterval).await();
     }
 
     @After
     public void tearDown() {
+        testee.stop();
         cassandra.clearAllTables();
         cassandra.close();
     }
@@ -142,15 +156,13 @@ public class V1ToV2MigrationTest {
 
         testee.getFromV2orElseFromV1AfterMigration(CassandraMessageDAOV2.notFound(metaData)).join();
 
-        Optional<CassandraMessageDAOV2.MessageResult> messageResult = messageDAOV2.retrieveMessages(metaDataList,
MessageMapper.FetchType.Full, Limit.unlimited())
-            .join()
-            .findAny();
+        awaitMigration();
 
-        assertThat(messageResult.isPresent()).isTrue();
-        softly.assertThat(messageResult.get().message().getLeft().getMessageId()).isEqualTo(messageId);
-        softly.assertThat(IOUtils.toString(messageResult.get().message().getLeft().getContent(),
Charsets.UTF_8))
+        CassandraMessageDAOV2.MessageResult messageResult = retrieveMessageOnV2().get();
+        softly.assertThat(messageResult.message().getLeft().getMessageId()).isEqualTo(messageId);
+        softly.assertThat(IOUtils.toString(messageResult.message().getLeft().getContent(),
Charsets.UTF_8))
             .isEqualTo(CONTENT);
-        softly.assertThat(messageResult.get().message().getRight().findAny().isPresent()).isFalse();
+        softly.assertThat(messageResult.message().getRight().findAny().isPresent()).isFalse();
     }
 
     @Test
@@ -164,15 +176,13 @@ public class V1ToV2MigrationTest {
 
         testee.getFromV2orElseFromV1AfterMigration(CassandraMessageDAOV2.notFound(metaData)).join();
 
-        Optional<CassandraMessageDAOV2.MessageResult> messageResult = messageDAOV2.retrieveMessages(metaDataList,
MessageMapper.FetchType.Full, Limit.unlimited())
-            .join()
-            .findAny();
+        awaitMigration();
 
-        assertThat(messageResult.isPresent()).isTrue();
-        softly.assertThat(messageResult.get().message().getLeft().getMessageId()).isEqualTo(messageId);
-        softly.assertThat(IOUtils.toString(messageResult.get().message().getLeft().getContent(),
Charsets.UTF_8))
+        CassandraMessageDAOV2.MessageResult messageResult = retrieveMessageOnV2().get();
+        softly.assertThat(messageResult.message().getLeft().getMessageId()).isEqualTo(messageId);
+        softly.assertThat(IOUtils.toString(messageResult.message().getLeft().getContent(),
Charsets.UTF_8))
             .isEqualTo(CONTENT);
-        softly.assertThat(messageResult.get().message().getRight().findAny().get()).isEqualTo(MessageAttachmentRepresentation.builder()
+        softly.assertThat(messageResult.message().getRight().findAny().get()).isEqualTo(MessageAttachmentRepresentation.builder()
             .attachmentId(attachment.getAttachmentId())
             .cid(OptionalConverter.fromGuava(messageAttachment.getCid()))
             .isInline(messageAttachment.isInline())
@@ -180,6 +190,27 @@ public class V1ToV2MigrationTest {
             .build());
     }
 
+    private void awaitMigration() {
+        awaitability.atMost(1, TimeUnit.MINUTES)
+            .until(() -> {
+                try {
+                    retrieveMessageOnV2();
+                    return true;
+                } catch(AssertionError e) {
+                    return false;
+                }
+            });
+    }
+
+    private Optional<CassandraMessageDAOV2.MessageResult> retrieveMessageOnV2() {
+        Optional<CassandraMessageDAOV2.MessageResult> messageResult = messageDAOV2.retrieveMessages(metaDataList,
MessageMapper.FetchType.Full, Limit.unlimited())
+            .join()
+            .findAny();
+
+        assertThat(messageResult.isPresent()).isTrue();
+        return messageResult;
+    }
+
     private SimpleMailboxMessage createMessage(MessageId messageId, String content, int bodyStart,
PropertyBuilder propertyBuilder, List<MessageAttachment> attachments) {
         return new SimpleMailboxMessage(messageId, new Date(), content.length(), bodyStart,
new SharedByteArrayInputStream(content.getBytes()), new Flags(), propertyBuilder, MAILBOX_ID,
attachments);
     }


---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org


Mime
View raw message