Code Review
/
controller.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
review
|
tree
raw
|
inline
| side by side
Change InMemoryJournal to ser/der the data
[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 e1af7db7ee401351191d98ac0e7f778a3baf9ab5..31894991e81224ad782fc2df3aca19b6eeb75c17 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
@@
-16,6
+16,7
@@
import akka.persistence.PersistentRepr;
import akka.persistence.journal.japi.AsyncWriteJournal;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.Uninterruptibles;
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.ArrayList;
import java.util.Collections;
import java.util.Iterator;
@@
-25,6
+26,7
@@
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang.SerializationUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;
@@
-56,6
+58,10
@@
public class InMemoryJournal extends AsyncWriteJournal {
private static final Map<String, CountDownLatch> blockReadMessagesLatches = new ConcurrentHashMap<>();
private static final Map<String, CountDownLatch> blockReadMessagesLatches = 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) {
public static void addEntry(String persistenceId, long sequenceNr, Object data) {
Map<Long, Object> journal = journals.get(persistenceId);
if(journal == null) {
@@
-64,7
+70,8
@@
public class InMemoryJournal extends AsyncWriteJournal {
}
synchronized (journal) {
}
synchronized (journal) {
- journal.put(sequenceNr, data);
+ journal.put(sequenceNr, data instanceof Serializable ?
+ SerializationUtils.serialize((Serializable) data) : data);
}
}
}
}
@@
-82,8
+89,9
@@
public class InMemoryJournal extends AsyncWriteJournal {
synchronized (journalMap) {
List<T> journal = new ArrayList<>(journalMap.size());
for(Object entry: journalMap.values()) {
synchronized (journalMap) {
List<T> journal = new ArrayList<>(journalMap.size());
for(Object entry: journalMap.values()) {
- if(type.isInstance(entry)) {
- journal.add((T) entry);
+ Object data = deserialize(entry);
+ if(type.isInstance(data)) {
+ journal.add((T) data);
}
}
}
}
@@
-140,7
+148,7
@@
public class InMemoryJournal extends AsyncWriteJournal {
@Override
public Future<Void> doAsyncReplayMessages(final String persistenceId, final long fromSequenceNr,
@Override
public Future<Void> doAsyncReplayMessages(final String persistenceId, final long fromSequenceNr,
- final long toSequenceNr, long max, final Procedure<PersistentRepr> replayCallback) {
+ final long toSequenceNr,
final
long max, final Procedure<PersistentRepr> replayCallback) {
return Futures.future(new Callable<Void>() {
@Override
public Void call() throws Exception {
return Futures.future(new Callable<Void>() {
@Override
public Void call() throws Exception {
@@
-155,10
+163,11
@@
public class InMemoryJournal extends AsyncWriteJournal {
}
synchronized (journal) {
}
synchronized (journal) {
+ int count = 0;
for (Map.Entry<Long,Object> entry : journal.entrySet()) {
for (Map.Entry<Long,Object> entry : journal.entrySet()) {
- if (entry.getKey() >= fromSequenceNr && entry.getKey() <= toSequenceNr) {
+ if (
++count <= max &&
entry.getKey() >= fromSequenceNr && entry.getKey() <= toSequenceNr) {
PersistentRepr persistentMessage =
PersistentRepr persistentMessage =
- new PersistentImpl(
entry.getValue(
), entry.getKey(), persistenceId,
+ new PersistentImpl(
deserialize(entry.getValue()
), entry.getKey(), persistenceId,
false, null, null);
replayCallback.apply(persistentMessage);
}
false, null, null);
replayCallback.apply(persistentMessage);
}
@@
-196,17
+205,10
@@
public class InMemoryJournal extends AsyncWriteJournal {
@Override
public Void call() throws Exception {
for (PersistentRepr repr : messages) {
@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);
- }
+ LOG.trace("doAsyncWriteMessages: id: {}: seqNr: {}, payload: {}", repr.persistenceId(),
+ repr.sequenceNr(), repr.payload());
- synchronized (journal) {
- LOG.trace("doAsyncWriteMessages: id: {}: seqNr: {}, payload: {}", repr.persistenceId(),
- repr.sequenceNr(), repr.payload());
- journal.put(repr.sequenceNr(), repr.payload());
- }
+ addEntry(repr.persistenceId(), repr.sequenceNr(), repr.payload());
WriteMessagesComplete complete = writeMessagesComplete.get(repr.persistenceId());
if(complete != null) {
WriteMessagesComplete complete = writeMessagesComplete.get(repr.persistenceId());
if(complete != null) {