import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
-import static org.opendaylight.controller.cluster.datastore.messages.CreateTransaction.CURRENT_CLIENT_VERSION;
+import static org.opendaylight.controller.cluster.datastore.messages.CreateTransaction.CURRENT_VERSION;
+import akka.actor.ActorRef;
+import akka.actor.PoisonPill;
+import akka.actor.Props;
+import akka.dispatch.Dispatchers;
+import akka.dispatch.OnComplete;
+import akka.japi.Creator;
+import akka.pattern.Patterns;
+import akka.testkit.TestActorRef;
+import akka.util.Timeout;
+import com.google.common.base.Function;
+import com.google.common.base.Optional;
+import com.google.common.util.concurrent.CheckedFuture;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.Uninterruptibles;
import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
-import akka.actor.ActorRef;
-import akka.actor.PoisonPill;
-import akka.actor.Props;
-import akka.dispatch.Dispatchers;
-import akka.dispatch.OnComplete;
-import akka.japi.Creator;
-import akka.pattern.Patterns;
-import akka.testkit.TestActorRef;
-import akka.util.Timeout;
-import com.google.common.base.Function;
-import com.google.common.base.Optional;
-import com.google.common.util.concurrent.CheckedFuture;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.MoreExecutors;
-import com.google.common.util.concurrent.Uninterruptibles;
-
public class ShardTest extends AbstractActorTest {
}};
}
+ @SuppressWarnings("serial")
@Test
public void testPeerAddressResolved() throws Exception {
new ShardTestKit(getSystem()) {{
shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
}
- private CompositeModificationPayload newPayload(Modification... mods) {
+ private CompositeModificationPayload newPayload(final Modification... mods) {
MutableCompositeModification compMod = new MutableCompositeModification();
for(Modification mod: mods) {
compMod.addModification(mod);
return new CompositeModificationPayload(compMod.toSerializable());
}
- private DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(String cohortName,
- InMemoryDOMDataStore dataStore, YangInstanceIdentifier path, NormalizedNode data,
- MutableCompositeModification modification) {
+ private DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(final String cohortName,
+ final InMemoryDOMDataStore dataStore, final YangInstanceIdentifier path, final NormalizedNode<?, ?> data,
+ final MutableCompositeModification modification) {
return setupMockWriteTransaction(cohortName, dataStore, path, data, modification, null);
}
- private DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(String cohortName,
- InMemoryDOMDataStore dataStore, YangInstanceIdentifier path, NormalizedNode data,
- MutableCompositeModification modification,
+ private DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(final String cohortName,
+ final InMemoryDOMDataStore dataStore, final YangInstanceIdentifier path, final NormalizedNode<?, ?> data,
+ final MutableCompositeModification modification,
final Function<DOMStoreThreePhaseCommitCohort,ListenableFuture<Void>> preCommit) {
DOMStoreWriteTransaction tx = dataStore.newWriteOnlyTransaction();
doAnswer(new Answer<ListenableFuture<Boolean>>() {
@Override
- public ListenableFuture<Boolean> answer(InvocationOnMock invocation) {
+ public ListenableFuture<Boolean> answer(final InvocationOnMock invocation) {
return realCohort.canCommit();
}
}).when(cohort).canCommit();
doAnswer(new Answer<ListenableFuture<Void>>() {
@Override
- public ListenableFuture<Void> answer(InvocationOnMock invocation) throws Throwable {
+ public ListenableFuture<Void> answer(final InvocationOnMock invocation) throws Throwable {
if(preCommit != null) {
return preCommit.apply(realCohort);
} else {
doAnswer(new Answer<ListenableFuture<Void>>() {
@Override
- public ListenableFuture<Void> answer(InvocationOnMock invocation) throws Throwable {
+ public ListenableFuture<Void> answer(final InvocationOnMock invocation) throws Throwable {
return realCohort.commit();
}
}).when(cohort).commit();
doAnswer(new Answer<ListenableFuture<Void>>() {
@Override
- public ListenableFuture<Void> answer(InvocationOnMock invocation) throws Throwable {
+ public ListenableFuture<Void> answer(final InvocationOnMock invocation) throws Throwable {
return realCohort.abort();
}
}).when(cohort).abort();
// Simulate the ForwardedReadyTransaction message for the first Tx that would be sent
// by the ShardTransaction.
- shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
cohort1, modification1, true), getRef());
ReadyTransactionReply readyReply = ReadyTransactionReply.fromSerializable(
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS));
// Send the ForwardedReadyTransaction for the next 2 Tx's.
- shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
cohort2, modification2, true), getRef());
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS);
- shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_VERSION,
cohort3, modification3, true), getRef());
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS);
class OnFutureComplete extends OnComplete<Object> {
private final Class<?> expRespType;
- OnFutureComplete(Class<?> expRespType) {
+ OnFutureComplete(final Class<?> expRespType) {
this.expRespType = expRespType;
}
@Override
- public void onComplete(Throwable error, Object resp) {
+ public void onComplete(final Throwable error, final Object resp) {
if(error != null) {
caughtEx.set(new AssertionError(getClass().getSimpleName() + " failure", error));
} else {
}
}
- void onSuccess(Object resp) throws Exception {
+ void onSuccess(final Object resp) throws Exception {
}
}
}
@Override
- public void onComplete(Throwable error, Object resp) {
+ public void onComplete(final Throwable error, final Object resp) {
super.onComplete(error, resp);
commitLatch.countDown();
}
class OnCanCommitFutureComplete extends OnFutureComplete {
private final String transactionID;
- OnCanCommitFutureComplete(String transactionID) {
+ OnCanCommitFutureComplete(final String transactionID) {
super(CanCommitTransactionReply.SERIALIZABLE_CLASS);
this.transactionID = transactionID;
}
@Override
- void onSuccess(Object resp) throws Exception {
+ void onSuccess(final Object resp) throws Exception {
CanCommitTransactionReply canCommitReply =
CanCommitTransactionReply.fromSerializable(resp);
assertEquals("Can commit", true, canCommitReply.getCanCommit());
// Simulate the ForwardedReadyTransaction messages that would be sent
// by the ShardTransaction.
- shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
cohort1, modification1, true), getRef());
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS);
- shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
cohort2, modification2, true), getRef());
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS);
final CountDownLatch latch = new CountDownLatch(1);
canCommitFuture.onComplete(new OnComplete<Object>() {
@Override
- public void onComplete(Throwable t, Object resp) {
+ public void onComplete(final Throwable t, final Object resp) {
latch.countDown();
}
}, getSystem().dispatcher());
// Simulate the ForwardedReadyTransaction messages that would be sent
// by the ShardTransaction.
- shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
cohort, modification, true), getRef());
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS);
// Simulate the ForwardedReadyTransaction messages that would be sent
// by the ShardTransaction.
- shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
cohort, modification, true), getRef());
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS);
new AbortTransaction(transactionID).toSerializable(), timeout);
abortFuture.onComplete(new OnComplete<Object>() {
@Override
- public void onComplete(Throwable e, Object resp) {
+ public void onComplete(final Throwable e, final Object resp) {
abortComplete.countDown();
}
}, getSystem().dispatcher());
TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME),
modification, preCommit);
- shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
cohort, modification, true), getRef());
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS);
// Ready the Tx's
- shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
cohort1, modification1, true), getRef());
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS);
- shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
cohort2, modification2, true), getRef());
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS);
// Ready the Tx's
- shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
cohort1, modification1, true), getRef());
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS);
- shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
cohort2, modification2, true), getRef());
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS);
- shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_VERSION,
cohort3, modification3, true), getRef());
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS);
// Simulate the ForwardedReadyTransaction messages that would be sent
// by the ShardTransaction.
- shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
cohort1, modification1, true), getRef());
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS);
- shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_CLIENT_VERSION,
+ shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
cohort2, modification2, true), getRef());
expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS);
final CountDownLatch latch = new CountDownLatch(1);
canCommitFuture.onComplete(new OnComplete<Object>() {
@Override
- public void onComplete(Throwable t, Object resp) {
+ public void onComplete(final Throwable t, final Object resp) {
latch.countDown();
}
}, getSystem().dispatcher());
testCreateSnapshot(false, "testCreateSnapshotWithNonPersistentData");
}
- public void testCreateSnapshot(boolean persistent, final String shardActorName) throws IOException, InterruptedException {
+ @SuppressWarnings("serial")
+ public void testCreateSnapshot(final boolean persistent, final String shardActorName) throws IOException, InterruptedException {
final DatastoreContext dataStoreContext = DatastoreContext.newBuilder().
shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(persistent).build();
return new Shard(shardID, Collections.<ShardIdentifier,String>emptyMap(),
dataStoreContext, SCHEMA_CONTEXT) {
@Override
- protected void commitSnapshot(long sequenceNumber) {
+ protected void commitSnapshot(final long sequenceNumber) {
super.commitSnapshot(sequenceNumber);
latch.get().countDown();
}
*/
@Test
public void testInMemoryDataStoreRestore() throws ReadFailedException {
- InMemoryDOMDataStore store = new InMemoryDOMDataStore("test", MoreExecutors.listeningDecorator(
- MoreExecutors.sameThreadExecutor()), MoreExecutors.sameThreadExecutor());
+ InMemoryDOMDataStore store = new InMemoryDOMDataStore("test", MoreExecutors.sameThreadExecutor());
store.onGlobalContextUpdated(SCHEMA_CONTEXT);
commitTransaction(putTransaction);
- NormalizedNode expected = readStore(store);
+ NormalizedNode<?, ?> expected = readStore(store);
DOMStoreWriteTransaction writeTransaction = store.newWriteOnlyTransaction();
commitTransaction(writeTransaction);
- NormalizedNode actual = readStore(store);
+ NormalizedNode<?, ?> actual = readStore(store);
assertEquals(expected, actual);
-
}
@Test
}
- private NormalizedNode readStore(InMemoryDOMDataStore store) throws ReadFailedException {
+ private NormalizedNode<?, ?> readStore(final InMemoryDOMDataStore store) throws ReadFailedException {
DOMStoreReadTransaction transaction = store.newReadOnlyTransaction();
CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read =
transaction.read(YangInstanceIdentifier.builder().build());
return normalizedNode;
}
- private void commitTransaction(DOMStoreWriteTransaction transaction) {
+ private void commitTransaction(final DOMStoreWriteTransaction transaction) {
DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready();
ListenableFuture<Void> future =
commitCohort.preCommit();
return new AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>() {
@Override
public void onDataChanged(
- AsyncDataChangeEvent<YangInstanceIdentifier, NormalizedNode<?, ?>> change) {
+ final AsyncDataChangeEvent<YangInstanceIdentifier, NormalizedNode<?, ?>> change) {
}
};
}
- static NormalizedNode<?,?> readStore(TestActorRef<Shard> shard, YangInstanceIdentifier id)
+ static NormalizedNode<?,?> readStore(final TestActorRef<Shard> shard, final YangInstanceIdentifier id)
throws ExecutionException, InterruptedException {
DOMStoreReadTransaction transaction = shard.underlyingActor().getDataStore().newReadOnlyTransaction();
return node;
}
- private void writeToStore(TestActorRef<Shard> shard, YangInstanceIdentifier id, NormalizedNode<?,?> node)
+ private void writeToStore(final TestActorRef<Shard> shard, final YangInstanceIdentifier id, final NormalizedNode<?,?> node)
throws ExecutionException, InterruptedException {
DOMStoreWriteTransaction transaction = shard.underlyingActor().getDataStore().newWriteOnlyTransaction();
commitCohort.commit().get();
}
+ @SuppressWarnings("serial")
private static final class DelegatingShardCreator implements Creator<Shard> {
private final Creator<Shard> delegate;
- DelegatingShardCreator(Creator<Shard> delegate) {
+ DelegatingShardCreator(final Creator<Shard> delegate) {
this.delegate = delegate;
}
Code Review / controller.git / blobdiff