BUG 2185 : Disable all internal switching of behavior

[controller.git] / opendaylight / md-sal / sal-akka-raft / src / test / java / org / opendaylight / controller / cluster / raft / utils / InMemoryJournal.java

diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/utils/InMemoryJournal.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/utils/InMemoryJournal.java
index 0737d75a7f2d679ae240c996c87cbd882052da28..7142c478295ad5869873253059bffbfa8f2a7466 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/utils/InMemoryJournal.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/utils/InMemoryJournal.java
@@ -36,13 +36,23 @@ import scala.concurrent.Future;
  */
 public class InMemoryJournal extends AsyncWriteJournal {
 
+    private static class WriteMessagesComplete {
+        final CountDownLatch latch;
+        final Class<?> ofType;
+
+        public WriteMessagesComplete(int count, Class<?> ofType) {
+            this.latch = new CountDownLatch(count);
+            this.ofType = ofType;
+        }
+    }
+
     static final Logger LOG = LoggerFactory.getLogger(InMemoryJournal.class);
 
     private static final Map<String, Map<Long, Object>> journals = new ConcurrentHashMap<>();
 
     private static final Map<String, CountDownLatch> deleteMessagesCompleteLatches = new ConcurrentHashMap<>();
 
-    private static final Map<String, CountDownLatch> writeMessagesCompleteLatches = new ConcurrentHashMap<>();
+    private static final Map<String, WriteMessagesComplete> writeMessagesComplete = new ConcurrentHashMap<>();
 
     private static final Map<String, CountDownLatch> blockReadMessagesLatches = new ConcurrentHashMap<>();
 
@@ -107,7 +117,7 @@ public class InMemoryJournal extends AsyncWriteJournal {
     }
 
     public static void waitForWriteMessagesComplete(String persistenceId) {
-        if(!Uninterruptibles.awaitUninterruptibly(writeMessagesCompleteLatches.get(persistenceId), 5, TimeUnit.SECONDS)) {
+        if(!Uninterruptibles.awaitUninterruptibly(writeMessagesComplete.get(persistenceId).latch, 5, TimeUnit.SECONDS)) {
             throw new AssertionError("Journal write messages did not complete");
         }
     }
@@ -117,7 +127,11 @@ public class InMemoryJournal extends AsyncWriteJournal {
     }
 
     public static void addWriteMessagesCompleteLatch(String persistenceId, int count) {
-        writeMessagesCompleteLatches.put(persistenceId, new CountDownLatch(count));
+        writeMessagesComplete.put(persistenceId, new WriteMessagesComplete(count, null));
+    }
+
+    public static void addWriteMessagesCompleteLatch(String persistenceId, int count, Class<?> ofType) {
+        writeMessagesComplete.put(persistenceId, new WriteMessagesComplete(count, ofType));
     }
 
     public static void addBlockReadMessagesLatch(String persistenceId, CountDownLatch latch) {
@@ -125,8 +139,8 @@ public class InMemoryJournal extends AsyncWriteJournal {
     }
 
     @Override
-    public Future<Void> doAsyncReplayMessages(final String persistenceId, long fromSequenceNr,
-            long toSequenceNr, long max, final Procedure<PersistentRepr> replayCallback) {
+    public Future<Void> doAsyncReplayMessages(final String persistenceId, final long fromSequenceNr,
+            final long toSequenceNr, long max, final Procedure<PersistentRepr> replayCallback) {
         return Futures.future(new Callable<Void>() {
             @Override
             public Void call() throws Exception {
@@ -136,16 +150,18 @@ public class InMemoryJournal extends AsyncWriteJournal {
                 }
 
                 Map<Long, Object> journal = journals.get(persistenceId);
-                if(journal == null) {
+                if (journal == null) {
                     return null;
                 }
 
                 synchronized (journal) {
                     for (Map.Entry<Long,Object> entry : journal.entrySet()) {
-                        PersistentRepr persistentMessage =
-                                new PersistentImpl(entry.getValue(), entry.getKey(), persistenceId,
-                                        false, null, null);
-                        replayCallback.apply(persistentMessage);
+                        if (entry.getKey() >= fromSequenceNr && entry.getKey() <= toSequenceNr) {
+                            PersistentRepr persistentMessage =
+                                    new PersistentImpl(entry.getValue(), entry.getKey(), persistenceId,
+                                            false, null, null);
+                            replayCallback.apply(persistentMessage);
+                        }
                     }
                 }
 
@@ -193,9 +209,11 @@ public class InMemoryJournal extends AsyncWriteJournal {
                         journal.put(repr.sequenceNr(), repr.payload());
                     }
 
-                    CountDownLatch latch = writeMessagesCompleteLatches.get(repr.persistenceId());
-                    if(latch != null) {
-                        latch.countDown();
+                    WriteMessagesComplete complete = writeMessagesComplete.get(repr.persistenceId());
+                    if(complete != null) {
+                        if(complete.ofType == null || complete.ofType.equals(repr.payload().getClass())) {
+                            complete.latch.countDown();
+                        }
                     }
                 }
 
@@ -216,6 +234,7 @@ public class InMemoryJournal extends AsyncWriteJournal {
 
     @Override
     public Future<Void> doAsyncDeleteMessagesTo(String persistenceId, long toSequenceNr, boolean permanent) {
+        LOG.trace("doAsyncDeleteMessagesTo: {}", toSequenceNr);
         Map<Long, Object> journal = journals.get(persistenceId);
         if(journal != null) {
             synchronized (journal) {