X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FDataChangeListenerRegistrationProxyTest.java;h=b338329740cbef8e8737b1e1fbed18ce97c6e60c;hb=204f45f8b3233dbea87e2c8065914f0d2a0ded07;hp=c27993f97b9fef669c03f0c378e83f8474944e96;hpb=1bc01a15b1e7811ee59249eab7e815408518e354;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DataChangeListenerRegistrationProxyTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DataChangeListenerRegistrationProxyTest.java index c27993f97b..b338329740 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DataChangeListenerRegistrationProxyTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DataChangeListenerRegistrationProxyTest.java @@ -7,7 +7,11 @@ */ package org.opendaylight.controller.cluster.datastore; -import java.util.concurrent.TimeUnit; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; import akka.actor.ActorRef; import akka.actor.ActorSystem; import akka.actor.Props; @@ -16,12 +20,16 @@ import akka.dispatch.ExecutionContexts; import akka.dispatch.Futures; import akka.testkit.JavaTestKit; import akka.util.Timeout; +import com.google.common.util.concurrent.MoreExecutors; +import com.google.common.util.concurrent.Uninterruptibles; +import java.util.concurrent.TimeUnit; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; -import org.opendaylight.controller.cluster.datastore.messages.ActorNotInitialized; +import org.opendaylight.controller.cluster.datastore.config.Configuration; +import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException; import org.opendaylight.controller.cluster.datastore.messages.CloseDataChangeListenerRegistration; import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard; import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound; @@ -29,23 +37,18 @@ import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener; import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply; import org.opendaylight.controller.cluster.datastore.utils.ActorContext; -import org.opendaylight.controller.cluster.datastore.utils.DoNothingActor; +import org.opendaylight.controller.cluster.datastore.utils.Dispatchers; +import org.opendaylight.controller.cluster.raft.utils.DoNothingActor; import org.opendaylight.controller.md.cluster.datastore.model.TestModel; import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker; import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener; +import org.opendaylight.controller.md.sal.dom.api.ClusteredDOMDataChangeListener; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; -import com.google.common.util.concurrent.MoreExecutors; -import com.google.common.util.concurrent.Uninterruptibles; import scala.concurrent.ExecutionContextExecutor; import scala.concurrent.Future; import scala.concurrent.duration.FiniteDuration; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.any; -import static org.mockito.Mockito.doReturn; -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.eq; /** * Unit tests for DataChangeListenerRegistrationProxy. @@ -66,7 +69,6 @@ public class DataChangeListenerRegistrationProxyTest extends AbstractActorTest { Assert.assertEquals(mockListener, proxy.getInstance()); } - @SuppressWarnings("unchecked") @Test(timeout=10000) public void testSuccessfulRegistration() { new JavaTestKit(getSystem()) {{ @@ -95,8 +97,9 @@ public class DataChangeListenerRegistrationProxyTest extends AbstractActorTest { RegisterChangeListener registerMsg = expectMsgClass(timeout, RegisterChangeListener.class); Assert.assertEquals("getPath", path, registerMsg.getPath()); Assert.assertEquals("getScope", scope, registerMsg.getScope()); + Assert.assertEquals("isRegisterOnAllInstances", false, registerMsg.isRegisterOnAllInstances()); - reply(new RegisterChangeListenerReply(getRef().path())); + reply(new RegisterChangeListenerReply(getRef())); for(int i = 0; (i < 20 * 5) && proxy.getListenerRegistrationActor() == null; i++) { Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS); @@ -121,6 +124,64 @@ public class DataChangeListenerRegistrationProxyTest extends AbstractActorTest { }}; } + @Test(timeout=10000) + public void testSuccessfulRegistrationForClusteredListener() { + new JavaTestKit(getSystem()) {{ + ActorContext actorContext = new ActorContext(getSystem(), getRef(), + mock(ClusterWrapper.class), mock(Configuration.class)); + + AsyncDataChangeListener> mockClusteredListener = + Mockito.mock(ClusteredDOMDataChangeListener.class); + + final DataChangeListenerRegistrationProxy proxy = new DataChangeListenerRegistrationProxy( + "shard-1", actorContext, mockClusteredListener); + + final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME); + final DataChangeScope scope = AsyncDataBroker.DataChangeScope.ONE; + new Thread() { + @Override + public void run() { + proxy.init(path, scope); + } + + }.start(); + + FiniteDuration timeout = duration("5 seconds"); + FindLocalShard findLocalShard = expectMsgClass(timeout, FindLocalShard.class); + Assert.assertEquals("getShardName", "shard-1", findLocalShard.getShardName()); + + reply(new LocalShardFound(getRef())); + + RegisterChangeListener registerMsg = expectMsgClass(timeout, RegisterChangeListener.class); + Assert.assertEquals("getPath", path, registerMsg.getPath()); + Assert.assertEquals("getScope", scope, registerMsg.getScope()); + Assert.assertEquals("isRegisterOnAllInstances", true, registerMsg.isRegisterOnAllInstances()); + + reply(new RegisterChangeListenerReply(getRef())); + + for(int i = 0; (i < 20 * 5) && proxy.getListenerRegistrationActor() == null; i++) { + Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS); + } + + Assert.assertEquals("getListenerRegistrationActor", getSystem().actorSelection(getRef().path()), + proxy.getListenerRegistrationActor()); + + watch(proxy.getDataChangeListenerActor()); + + proxy.close(); + + // The listener registration actor should get a Close message + expectMsgClass(timeout, CloseDataChangeListenerRegistration.SERIALIZABLE_CLASS); + + // The DataChangeListener actor should be terminated + expectMsgClass(timeout, Terminated.class); + + proxy.close(); + + expectNoMsg(); + }}; + } + @Test(timeout=10000) public void testLocalShardNotFound() { new JavaTestKit(getSystem()) {{ @@ -173,7 +234,7 @@ public class DataChangeListenerRegistrationProxyTest extends AbstractActorTest { FindLocalShard findLocalShard = expectMsgClass(timeout, FindLocalShard.class); Assert.assertEquals("getShardName", "shard-1", findLocalShard.getShardName()); - reply(new ActorNotInitialized()); + reply(new NotInitializedException("not initialized")); new Within(duration("1 seconds")) { @Override @@ -193,22 +254,24 @@ public class DataChangeListenerRegistrationProxyTest extends AbstractActorTest { "testFailedRegistration"); doReturn(mockActor).when(mockActorSystem).actorOf(any(Props.class)); ExecutionContextExecutor executor = ExecutionContexts.fromExecutor( - MoreExecutors.sameThreadExecutor()); - doReturn(executor).when(mockActorSystem).dispatcher(); + MoreExecutors.directExecutor()); + ActorContext actorContext = mock(ActorContext.class); + doReturn(executor).when(actorContext).getClientDispatcher(); + String shardName = "shard-1"; final DataChangeListenerRegistrationProxy proxy = new DataChangeListenerRegistrationProxy( shardName, actorContext, mockListener); doReturn(mockActorSystem).when(actorContext).getActorSystem(); doReturn(duration("5 seconds")).when(actorContext).getOperationDuration(); - doReturn(Futures.successful(getRef())).when(actorContext).findLocalShardAsync(eq(shardName), - any(Timeout.class)); + doReturn(Futures.successful(getRef())).when(actorContext).findLocalShardAsync(eq(shardName)); doReturn(Futures.failed(new RuntimeException("mock"))). when(actorContext).executeOperationAsync(any(ActorRef.class), any(Object.class), any(Timeout.class)); + doReturn(mock(DatastoreContext.class)).when(actorContext).getDatastoreContext(); proxy.init(YangInstanceIdentifier.of(TestModel.TEST_QNAME), AsyncDataBroker.DataChangeScope.ONE); @@ -218,7 +281,6 @@ public class DataChangeListenerRegistrationProxyTest extends AbstractActorTest { }}; } - @SuppressWarnings("unchecked") @Test public void testCloseBeforeRegistration() { new JavaTestKit(getSystem()) {{ @@ -228,18 +290,20 @@ public class DataChangeListenerRegistrationProxyTest extends AbstractActorTest { final DataChangeListenerRegistrationProxy proxy = new DataChangeListenerRegistrationProxy( shardName, actorContext, mockListener); + doReturn(DatastoreContext.newBuilder().build()).when(actorContext).getDatastoreContext(); + doReturn(getSystem().dispatchers().defaultGlobalDispatcher()).when(actorContext).getClientDispatcher(); doReturn(getSystem()).when(actorContext).getActorSystem(); + doReturn(Dispatchers.DEFAULT_DISPATCHER_PATH).when(actorContext).getNotificationDispatcherPath(); doReturn(getSystem().actorSelection(getRef().path())). when(actorContext).actorSelection(getRef().path()); doReturn(duration("5 seconds")).when(actorContext).getOperationDuration(); - doReturn(Futures.successful(getRef())).when(actorContext).findLocalShardAsync(eq(shardName), - any(Timeout.class)); + doReturn(Futures.successful(getRef())).when(actorContext).findLocalShardAsync(eq(shardName)); Answer> answer = new Answer>() { @Override public Future answer(InvocationOnMock invocation) { proxy.close(); - return Futures.successful((Object)new RegisterChangeListenerReply(getRef().path())); + return Futures.successful((Object)new RegisterChangeListenerReply(getRef())); } };