Fix warnings in sal-akka-raft test classes
[controller.git] / opendaylight / md-sal / sal-akka-raft / src / test / java / org / opendaylight / controller / cluster / raft / utils / InMemoryJournal.java
index 0e2449e..f2a216c 100644 (file)
@@ -8,23 +8,24 @@
 package org.opendaylight.controller.cluster.raft.utils;
 
 import akka.dispatch.Futures;
-import akka.japi.Procedure;
-import akka.persistence.PersistentConfirmation;
-import akka.persistence.PersistentId;
+import akka.persistence.AtomicWrite;
 import akka.persistence.PersistentImpl;
 import akka.persistence.PersistentRepr;
 import akka.persistence.journal.japi.AsyncWriteJournal;
 import com.google.common.collect.Maps;
 import com.google.common.util.concurrent.Uninterruptibles;
+import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.Callable;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import org.apache.commons.lang.SerializationUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import scala.concurrent.Future;
@@ -40,7 +41,7 @@ public class InMemoryJournal extends AsyncWriteJournal {
         final CountDownLatch latch;
         final Class<?> ofType;
 
-        public WriteMessagesComplete(int count, Class<?> ofType) {
+        WriteMessagesComplete(int count, Class<?> ofType) {
             this.latch = new CountDownLatch(count);
             this.ofType = ofType;
         }
@@ -48,42 +49,48 @@ public class InMemoryJournal extends AsyncWriteJournal {
 
     static final Logger LOG = LoggerFactory.getLogger(InMemoryJournal.class);
 
-    private static final Map<String, Map<Long, Object>> journals = new ConcurrentHashMap<>();
+    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> DELETE_MESSAGES_COMPLETE_LATCHES = new ConcurrentHashMap<>();
 
-    private static final Map<String, WriteMessagesComplete> writeMessagesComplete = new ConcurrentHashMap<>();
+    private static final Map<String, WriteMessagesComplete> WRITE_MESSAGES_COMPLETE = new ConcurrentHashMap<>();
 
-    private static final Map<String, CountDownLatch> blockReadMessagesLatches = new ConcurrentHashMap<>();
+    private static final Map<String, CountDownLatch> BLOCK_READ_MESSAGES_LATCHES = new ConcurrentHashMap<>();
+
+    private static Object deserialize(Object data) {
+        return data instanceof byte[] ? SerializationUtils.deserialize((byte[])data) : data;
+    }
 
     public static void addEntry(String persistenceId, long sequenceNr, Object data) {
-        Map<Long, Object> journal = journals.get(persistenceId);
-        if(journal == null) {
+        Map<Long, Object> journal = JOURNALS.get(persistenceId);
+        if (journal == null) {
             journal = Maps.newLinkedHashMap();
-            journals.put(persistenceId, journal);
+            JOURNALS.put(persistenceId, journal);
         }
 
         synchronized (journal) {
-            journal.put(sequenceNr, data);
+            journal.put(sequenceNr, data instanceof Serializable
+                    ? SerializationUtils.serialize((Serializable) data) : data);
         }
     }
 
     public static void clear() {
-        journals.clear();
+        JOURNALS.clear();
     }
 
     @SuppressWarnings("unchecked")
     public static <T> List<T> get(String persistenceId, Class<T> type) {
-        Map<Long, Object> journalMap = journals.get(persistenceId);
-        if(journalMap == null) {
+        Map<Long, Object> journalMap = JOURNALS.get(persistenceId);
+        if (journalMap == null) {
             return Collections.<T>emptyList();
         }
 
         synchronized (journalMap) {
             List<T> journal = new ArrayList<>(journalMap.size());
-            for(Object entry: journalMap.values()) {
-                if(type.isInstance(entry)) {
-                    journal.add((T) entry);
+            for (Object entry: journalMap.values()) {
+                Object data = deserialize(entry);
+                if (type.isInstance(data)) {
+                    journal.add((T) data);
                 }
             }
 
@@ -92,16 +99,16 @@ public class InMemoryJournal extends AsyncWriteJournal {
     }
 
     public static Map<Long, Object> get(String persistenceId) {
-        Map<Long, Object> journalMap = journals.get(persistenceId);
+        Map<Long, Object> journalMap = JOURNALS.get(persistenceId);
         return journalMap != null ? journalMap : Collections.<Long, Object>emptyMap();
     }
 
     public static void dumpJournal(String persistenceId) {
         StringBuilder builder = new StringBuilder(String.format("Journal log for %s:", persistenceId));
-        Map<Long, Object> journalMap = journals.get(persistenceId);
-        if(journalMap != null) {
+        Map<Long, Object> journalMap = JOURNALS.get(persistenceId);
+        if (journalMap != null) {
             synchronized (journalMap) {
-                for(Map.Entry<Long, Object> e: journalMap.entrySet()) {
+                for (Map.Entry<Long, Object> e: journalMap.entrySet()) {
                     builder.append("\n    ").append(e.getKey()).append(" = ").append(e.getValue());
                 }
             }
@@ -111,76 +118,81 @@ public class InMemoryJournal extends AsyncWriteJournal {
     }
 
     public static void waitForDeleteMessagesComplete(String persistenceId) {
-        if(!Uninterruptibles.awaitUninterruptibly(deleteMessagesCompleteLatches.get(persistenceId), 5, TimeUnit.SECONDS)) {
+        if (!Uninterruptibles.awaitUninterruptibly(DELETE_MESSAGES_COMPLETE_LATCHES.get(persistenceId),
+                5, TimeUnit.SECONDS)) {
             throw new AssertionError("Delete messages did not complete");
         }
     }
 
     public static void waitForWriteMessagesComplete(String persistenceId) {
-        if(!Uninterruptibles.awaitUninterruptibly(writeMessagesComplete.get(persistenceId).latch, 5, TimeUnit.SECONDS)) {
+        if (!Uninterruptibles.awaitUninterruptibly(WRITE_MESSAGES_COMPLETE.get(persistenceId).latch,
+                5, TimeUnit.SECONDS)) {
             throw new AssertionError("Journal write messages did not complete");
         }
     }
 
     public static void addDeleteMessagesCompleteLatch(String persistenceId) {
-        deleteMessagesCompleteLatches.put(persistenceId, new CountDownLatch(1));
+        DELETE_MESSAGES_COMPLETE_LATCHES.put(persistenceId, new CountDownLatch(1));
     }
 
     public static void addWriteMessagesCompleteLatch(String persistenceId, int count) {
-        writeMessagesComplete.put(persistenceId, new WriteMessagesComplete(count, null));
+        WRITE_MESSAGES_COMPLETE.put(persistenceId, new WriteMessagesComplete(count, null));
     }
 
     public static void addWriteMessagesCompleteLatch(String persistenceId, int count, Class<?> ofType) {
-        writeMessagesComplete.put(persistenceId, new WriteMessagesComplete(count, ofType));
+        WRITE_MESSAGES_COMPLETE.put(persistenceId, new WriteMessagesComplete(count, ofType));
     }
 
     public static void addBlockReadMessagesLatch(String persistenceId, CountDownLatch latch) {
-        blockReadMessagesLatches.put(persistenceId, latch);
+        BLOCK_READ_MESSAGES_LATCHES.put(persistenceId, latch);
     }
 
     @Override
-    public Future<Void> doAsyncReplayMessages(final String persistenceId, long fromSequenceNr,
-            long toSequenceNr, long max, final Procedure<PersistentRepr> replayCallback) {
-        return Futures.future(new Callable<Void>() {
-            @Override
-            public Void call() throws Exception {
-                CountDownLatch blockLatch = blockReadMessagesLatches.remove(persistenceId);
-                if(blockLatch != null) {
-                    Uninterruptibles.awaitUninterruptibly(blockLatch);
-                }
+    public Future<Void> doAsyncReplayMessages(final String persistenceId, final long fromSequenceNr,
+            final long toSequenceNr, final long max, final Consumer<PersistentRepr> replayCallback) {
+        LOG.trace("doAsyncReplayMessages for {}: fromSequenceNr: {}, toSequenceNr: {}", persistenceId,
+                fromSequenceNr,toSequenceNr);
+        return Futures.future(() -> {
+            CountDownLatch blockLatch = BLOCK_READ_MESSAGES_LATCHES.remove(persistenceId);
+            if (blockLatch != null) {
+                Uninterruptibles.awaitUninterruptibly(blockLatch);
+            }
 
-                Map<Long, Object> journal = journals.get(persistenceId);
-                if(journal == null) {
-                    return null;
-                }
+            Map<Long, Object> journal = JOURNALS.get(persistenceId);
+            if (journal == null) {
+                return null;
+            }
 
-                synchronized (journal) {
-                    for (Map.Entry<Long,Object> entry : journal.entrySet()) {
+            synchronized (journal) {
+                int count = 0;
+                for (Map.Entry<Long,Object> entry : journal.entrySet()) {
+                    if (++count <= max && entry.getKey() >= fromSequenceNr && entry.getKey() <= toSequenceNr) {
                         PersistentRepr persistentMessage =
-                                new PersistentImpl(entry.getValue(), entry.getKey(), persistenceId,
-                                        false, null, null);
-                        replayCallback.apply(persistentMessage);
+                                new PersistentImpl(deserialize(entry.getValue()), entry.getKey(), persistenceId,
+                                        null, false, null, null);
+                        replayCallback.accept(persistentMessage);
                     }
                 }
-
-                return null;
             }
+
+            return null;
         }, context().dispatcher());
     }
 
     @Override
     public Future<Long> doAsyncReadHighestSequenceNr(String persistenceId, long fromSequenceNr) {
-        // Akka calls this during recovery.
+        LOG.trace("doAsyncReadHighestSequenceNr for {}: fromSequenceNr: {}", persistenceId, fromSequenceNr);
 
-        Map<Long, Object> journal = journals.get(persistenceId);
-        if(journal == null) {
-            return Futures.successful(-1L);
+        // Akka calls this during recovery.
+        Map<Long, Object> journal = JOURNALS.get(persistenceId);
+        if (journal == null) {
+            return Futures.successful(fromSequenceNr);
         }
 
         synchronized (journal) {
             long highest = -1;
             for (Long seqNr : journal.keySet()) {
-                if(seqNr.longValue() >= fromSequenceNr && seqNr.longValue() > highest) {
+                if (seqNr.longValue() >= fromSequenceNr && seqNr.longValue() > highest) {
                     highest = seqNr.longValue();
                 }
             }
@@ -190,64 +202,49 @@ public class InMemoryJournal extends AsyncWriteJournal {
     }
 
     @Override
-    public Future<Void> doAsyncWriteMessages(final Iterable<PersistentRepr> messages) {
-        return Futures.future(new Callable<Void>() {
-            @Override
-            public Void call() throws Exception {
-                for (PersistentRepr repr : messages) {
-                    Map<Long, Object> journal = journals.get(repr.persistenceId());
-                    if(journal == null) {
-                        journal = Maps.newLinkedHashMap();
-                        journals.put(repr.persistenceId(), journal);
-                    }
-
-                    synchronized (journal) {
-                        LOG.trace("doAsyncWriteMessages: id: {}: seqNr: {}, payload: {}", repr.persistenceId(),
-                                repr.sequenceNr(), repr.payload());
-                        journal.put(repr.sequenceNr(), repr.payload());
-                    }
-
-                    WriteMessagesComplete complete = writeMessagesComplete.get(repr.persistenceId());
-                    if(complete != null) {
-                        if(complete.ofType == null || complete.ofType.equals(repr.payload().getClass())) {
+    public Future<Iterable<Optional<Exception>>> doAsyncWriteMessages(final Iterable<AtomicWrite> messages) {
+        return Futures.future(() -> {
+            for (AtomicWrite write : messages) {
+                // Copy to array - workaround for eclipse "ambiguous method" errors for toIterator, toIterable etc
+                PersistentRepr[] array = new PersistentRepr[write.payload().size()];
+                write.payload().copyToArray(array);
+                for (PersistentRepr repr: array) {
+                    LOG.trace("doAsyncWriteMessages: id: {}: seqNr: {}, payload: {}", repr.persistenceId(),
+                        repr.sequenceNr(), repr.payload());
+
+                    addEntry(repr.persistenceId(), repr.sequenceNr(), repr.payload());
+
+                    WriteMessagesComplete complete = WRITE_MESSAGES_COMPLETE.get(repr.persistenceId());
+                    if (complete != null) {
+                        if (complete.ofType == null || complete.ofType.equals(repr.payload().getClass())) {
                             complete.latch.countDown();
                         }
                     }
                 }
-
-                return null;
             }
-        }, context().dispatcher());
-    }
 
-    @Override
-    public Future<Void> doAsyncWriteConfirmations(Iterable<PersistentConfirmation> confirmations) {
-        return Futures.successful(null);
-    }
-
-    @Override
-    public Future<Void> doAsyncDeleteMessages(Iterable<PersistentId> messageIds, boolean permanent) {
-        return Futures.successful(null);
+            return Collections.emptyList();
+        }, context().dispatcher());
     }
 
     @Override
-    public Future<Void> doAsyncDeleteMessagesTo(String persistenceId, long toSequenceNr, boolean permanent) {
+    public Future<Void> doAsyncDeleteMessagesTo(String persistenceId, long toSequenceNr) {
         LOG.trace("doAsyncDeleteMessagesTo: {}", toSequenceNr);
-        Map<Long, Object> journal = journals.get(persistenceId);
-        if(journal != null) {
+        Map<Long, Object> journal = JOURNALS.get(persistenceId);
+        if (journal != null) {
             synchronized (journal) {
                 Iterator<Long> iter = journal.keySet().iterator();
-                while(iter.hasNext()) {
-                    Long n = iter.next();
-                    if(n <= toSequenceNr) {
+                while (iter.hasNext()) {
+                    Long num = iter.next();
+                    if (num <= toSequenceNr) {
                         iter.remove();
                     }
                 }
             }
         }
 
-        CountDownLatch latch = deleteMessagesCompleteLatches.get(persistenceId);
-        if(latch != null) {
+        CountDownLatch latch = DELETE_MESSAGES_COMPLETE_LATCHES.get(persistenceId);
+        if (latch != null) {
             latch.countDown();
         }
 

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.