import com.google.common.base.Optional;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.Uninterruptibles;
import com.typesafe.config.ConfigFactory;
import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
import org.opendaylight.controller.cluster.datastore.ClusterWrapper;
import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
import org.opendaylight.controller.cluster.datastore.DatastoreContext;
+import org.opendaylight.controller.cluster.datastore.DatastoreContextFactory;
import org.opendaylight.controller.cluster.datastore.config.Configuration;
import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound;
+import org.opendaylight.controller.cluster.raft.utils.EchoActor;
+import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
private final ActorRef actorRef;
private final Map<String,Object> findPrimaryResponses = Maps.newHashMap();
- private MockShardManager(boolean found, ActorRef actorRef){
+ private MockShardManager(final boolean found, final ActorRef actorRef){
this.found = found;
this.actorRef = actorRef;
}
- @Override public void onReceive(Object message) throws Exception {
+ @Override public void onReceive(final Object message) throws Exception {
if(message instanceof FindPrimary) {
FindPrimary fp = (FindPrimary)message;
Object resp = findPrimaryResponses.get(fp.getShardName());
}
}
- void addFindPrimaryResp(String shardName, Object resp) {
+ void addFindPrimaryResp(final String shardName, final Object resp) {
findPrimaryResponses.put(shardName, resp);
}
this.actorRef = null;
}
- MockShardManagerCreator(boolean found, ActorRef actorRef) {
+ MockShardManagerCreator(final boolean found, final ActorRef actorRef) {
this.found = found;
this.actorRef = actorRef;
}
assertEquals(false, actorContext.isPathLocal("akka.tcp://system@127.0.0.1:2551/"));
}
- @Test
- public void testResolvePathForRemoteActor() {
- ActorContext actorContext =
- new ActorContext(getSystem(), mock(ActorRef.class), mock(
- ClusterWrapper.class),
- mock(Configuration.class));
-
- String actual = actorContext.resolvePath(
- "akka.tcp://system@127.0.0.1:2550/user/shardmanager/shard",
- "akka://system/user/shardmanager/shard/transaction");
-
- String expected = "akka.tcp://system@127.0.0.1:2550/user/shardmanager/shard/transaction";
-
- assertEquals(expected, actual);
- }
-
- @Test
- public void testResolvePathForLocalActor() {
- ActorContext actorContext =
- new ActorContext(getSystem(), mock(ActorRef.class), mock(ClusterWrapper.class),
- mock(Configuration.class));
-
- String actual = actorContext.resolvePath(
- "akka://system/user/shardmanager/shard",
- "akka://system/user/shardmanager/shard/transaction");
-
- String expected = "akka://system/user/shardmanager/shard/transaction";
-
- assertEquals(expected, actual);
- }
-
- @Test
- public void testResolvePathForRemoteActorWithProperRemoteAddress() {
- ActorContext actorContext =
- new ActorContext(getSystem(), mock(ActorRef.class), mock(ClusterWrapper.class),
- mock(Configuration.class));
-
- String actual = actorContext.resolvePath(
- "akka.tcp://system@7.0.0.1:2550/user/shardmanager/shard",
- "akka.tcp://system@7.0.0.1:2550/user/shardmanager/shard/transaction");
-
- String expected = "akka.tcp://system@7.0.0.1:2550/user/shardmanager/shard/transaction";
-
- assertEquals(expected, actual);
- }
-
-
@Test
public void testClientDispatcherIsGlobalDispatcher(){
ActorContext actorContext =
mock(Configuration.class), DatastoreContext.newBuilder().build(), new PrimaryShardInfoFutureCache());
assertEquals(getSystem().dispatchers().defaultGlobalDispatcher(), actorContext.getClientDispatcher());
-
}
@Test
assertNotEquals(actorSystem.dispatchers().defaultGlobalDispatcher(), actorContext.getClientDispatcher());
- actorSystem.shutdown();
-
+ actorSystem.terminate();
}
@Test
DatastoreContext newContext = DatastoreContext.newBuilder().operationTimeoutInSeconds(6).
shardTransactionCommitTimeoutInSeconds(8).build();
- actorContext.setDatastoreContext(newContext);
+ DatastoreContextFactory mockContextFactory = mock(DatastoreContextFactory.class);
+ Mockito.doReturn(newContext).when(mockContextFactory).getBaseDatastoreContext();
+
+ actorContext.setDatastoreContext(mockContextFactory);
- expectMsgClass(duration("5 seconds"), DatastoreContext.class);
+ expectMsgClass(duration("5 seconds"), DatastoreContextFactory.class);
Assert.assertSame("getDatastoreContext", newContext, actorContext.getDatastoreContext());
TestActorRef<MessageCollectorActor> shardManager =
TestActorRef.create(getSystem(), Props.create(MessageCollectorActor.class));
- DatastoreContext dataStoreContext = DatastoreContext.newBuilder().dataStoreType("config").
+ DatastoreContext dataStoreContext = DatastoreContext.newBuilder().
+ logicalStoreType(LogicalDatastoreType.CONFIGURATION).
shardLeaderElectionTimeout(100, TimeUnit.MILLISECONDS).build();
final String expPrimaryPath = "akka://test-system/find-primary-shard";
new ActorContext(getSystem(), shardManager, mock(ClusterWrapper.class),
mock(Configuration.class), dataStoreContext, new PrimaryShardInfoFutureCache()) {
@Override
- protected Future<Object> doAsk(ActorRef actorRef, Object message, Timeout timeout) {
+ protected Future<Object> doAsk(final ActorRef actorRef, final Object message, final Timeout timeout) {
return Futures.successful((Object) new RemotePrimaryShardFound(expPrimaryPath, expPrimaryVersion));
}
};
TestActorRef<MessageCollectorActor> shardManager =
TestActorRef.create(getSystem(), Props.create(MessageCollectorActor.class));
- DatastoreContext dataStoreContext = DatastoreContext.newBuilder().dataStoreType("config").
+ DatastoreContext dataStoreContext = DatastoreContext.newBuilder().
+ logicalStoreType(LogicalDatastoreType.CONFIGURATION).
shardLeaderElectionTimeout(100, TimeUnit.MILLISECONDS).build();
final DataTree mockDataTree = Mockito.mock(DataTree.class);
new ActorContext(getSystem(), shardManager, mock(ClusterWrapper.class),
mock(Configuration.class), dataStoreContext, new PrimaryShardInfoFutureCache()) {
@Override
- protected Future<Object> doAsk(ActorRef actorRef, Object message, Timeout timeout) {
+ protected Future<Object> doAsk(final ActorRef actorRef, final Object message, final Timeout timeout) {
return Futures.successful((Object) new LocalPrimaryShardFound(expPrimaryPath, mockDataTree));
}
};
testFindPrimaryExceptions(new NotInitializedException("not initialized"));
}
- private void testFindPrimaryExceptions(final Object expectedException) throws Exception {
+ private static void testFindPrimaryExceptions(final Object expectedException) throws Exception {
TestActorRef<MessageCollectorActor> shardManager =
TestActorRef.create(getSystem(), Props.create(MessageCollectorActor.class));
- DatastoreContext dataStoreContext = DatastoreContext.newBuilder().dataStoreType("config").
+ DatastoreContext dataStoreContext = DatastoreContext.newBuilder().
+ logicalStoreType(LogicalDatastoreType.CONFIGURATION).
shardLeaderElectionTimeout(100, TimeUnit.MILLISECONDS).build();
ActorContext actorContext =
new ActorContext(getSystem(), shardManager, mock(ClusterWrapper.class),
mock(Configuration.class), dataStoreContext, new PrimaryShardInfoFutureCache()) {
@Override
- protected Future<Object> doAsk(ActorRef actorRef, Object message, Timeout timeout) {
+ protected Future<Object> doAsk(final ActorRef actorRef, final Object message, final Timeout timeout) {
return Futures.successful(expectedException);
}
};
mock(ClusterWrapper.class), mockConfig,
DatastoreContext.newBuilder().shardInitializationTimeout(200, TimeUnit.MILLISECONDS).build(), new PrimaryShardInfoFutureCache());
- actorContext.broadcast(new TestMessage());
+ actorContext.broadcast(new Function<Short, Object>() {
+ @Override
+ public Object apply(final Short v) {
+ return new TestMessage();
+ }
+ }, TestMessage.class);
- expectFirstMatching(shardActorRef1, TestMessage.class);
- expectFirstMatching(shardActorRef2, TestMessage.class);
+ MessageCollectorActor.expectFirstMatching(shardActorRef1, TestMessage.class);
+ MessageCollectorActor.expectFirstMatching(shardActorRef2, TestMessage.class);
}};
}
- private <T> T expectFirstMatching(ActorRef actor, Class<T> clazz) {
- int count = 5000 / 50;
- for(int i = 0; i < count; i++) {
- try {
- T message = (T) MessageCollectorActor.getFirstMatching(actor, clazz);
- if(message != null) {
- return message;
- }
- } catch (Exception e) {}
-
- Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
- }
-
- Assert.fail("Did not receive message of type " + clazz);
- return null;
- }
}
Code Review / controller.git / blobdiff