Tests for CDS DataTreeChangedListener 52/17652/12
authorJan Hajnar <jhajnar@cisco.com>
Thu, 2 Apr 2015 15:26:46 +0000 (17:26 +0200)
committerTom Pantelis <tpanteli@brocade.com>
Thu, 23 Apr 2015 08:00:29 +0000 (04:00 -0400)
* implemented tests

Change-Id: I6cc8f4011a056f19991165a5b324b3999d25a951
Signed-off-by: Jan Hajnar <jhajnar@cisco.com>
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerProxy.java
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/messages/RegisterDataTreeChangeListenerReply.java
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerActorTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerProxyTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerRegistrationActorTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ForwardingDataTreeChangeListenerTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ShardTest.java
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/utils/MockDataTreeChangeListener.java [new file with mode: 0644]

index 124724b..1a27f2e 100644 (file)
@@ -11,6 +11,7 @@ import akka.actor.ActorRef;
 import akka.actor.ActorSelection;
 import akka.actor.PoisonPill;
 import akka.dispatch.OnComplete;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import javax.annotation.concurrent.GuardedBy;
 import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
@@ -106,9 +107,19 @@ final class DataTreeChangeListenerProxy<T extends DOMDataTreeChangeListener> ext
                 } else {
                     RegisterDataTreeChangeListenerReply reply = (RegisterDataTreeChangeListenerReply) result;
                     setListenerRegistrationActor(actorContext.actorSelection(
-                            reply.getListenerRegistrationPath().path()));
+                            reply.getListenerRegistrationPath()));
                 }
             }
         }, actorContext.getClientDispatcher());
     }
+
+    @VisibleForTesting
+    ActorSelection getListenerRegistrationActor() {
+        return listenerRegistrationActor;
+    }
+
+    @VisibleForTesting
+    ActorRef getDataChangeListenerActor() {
+        return dataChangeListenerActor;
+    }
 }
index 88682ae..4c950ce 100644 (file)
@@ -7,8 +7,10 @@
  */
 package org.opendaylight.controller.cluster.datastore.messages;
 
+import akka.actor.ActorPath;
 import akka.actor.ActorRef;
 import com.google.common.base.Preconditions;
+
 import java.io.Serializable;
 
 /**
@@ -22,7 +24,7 @@ public final class RegisterDataTreeChangeListenerReply implements Serializable {
         this.listenerRegistrationPath = Preconditions.checkNotNull(listenerRegistrationPath);
     }
 
-    public ActorRef getListenerRegistrationPath() {
-        return listenerRegistrationPath;
+    public ActorPath getListenerRegistrationPath() {
+        return listenerRegistrationPath.path();
     }
 }
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerActorTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerActorTest.java
new file mode 100644 (file)
index 0000000..37a6197
--- /dev/null
@@ -0,0 +1,140 @@
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.datastore;
+
+import akka.actor.ActorRef;
+import akka.actor.DeadLetter;
+import akka.actor.Props;
+import akka.testkit.JavaTestKit;
+import com.google.common.collect.ImmutableList;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Matchers;
+import org.mockito.Mockito;
+import org.opendaylight.controller.cluster.datastore.messages.DataTreeChanged;
+import org.opendaylight.controller.cluster.datastore.messages.DataTreeChangedReply;
+import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
+
+public class DataTreeChangeListenerActorTest extends AbstractActorTest {
+
+    @SuppressWarnings({ "rawtypes", "unchecked" })
+    @Test
+    public void testDataChangedWhenNotificationsAreEnabled(){
+        new JavaTestKit(getSystem()) {{
+            final DataTreeCandidate mockTreeCandidate = Mockito.mock(DataTreeCandidate.class);
+            final ImmutableList<DataTreeCandidate> mockCandidates = ImmutableList.of(mockTreeCandidate);
+            final DOMDataTreeChangeListener mockListener = Mockito.mock(DOMDataTreeChangeListener.class);
+            final Props props = DataTreeChangeListenerActor.props(mockListener);
+            final ActorRef subject = getSystem().actorOf(props, "testDataTreeChangedNotificationsEnabled");
+
+            // Let the DataChangeListener know that notifications should be enabled
+            subject.tell(new EnableNotification(true), getRef());
+
+            subject.tell(new DataTreeChanged(mockCandidates),
+                    getRef());
+
+            expectMsgClass(DataTreeChangedReply.class);
+
+            Mockito.verify(mockListener).onDataTreeChanged(mockCandidates);
+        }};
+    }
+
+    @SuppressWarnings({ "rawtypes", "unchecked" })
+    @Test
+    public void testDataChangedWhenNotificationsAreDisabled(){
+        new JavaTestKit(getSystem()) {{
+            final DataTreeCandidate mockTreeCandidate = Mockito.mock(DataTreeCandidate.class);
+            final ImmutableList<DataTreeCandidate> mockCandidates = ImmutableList.of(mockTreeCandidate);
+            final DOMDataTreeChangeListener mockListener = Mockito.mock(DOMDataTreeChangeListener.class);
+            final Props props = DataTreeChangeListenerActor.props(mockListener);
+            final ActorRef subject =
+                    getSystem().actorOf(props, "testDataTreeChangedNotificationsDisabled");
+
+            subject.tell(new DataTreeChanged(mockCandidates),
+                    getRef());
+
+            new Within(duration("1 seconds")) {
+                @Override
+                protected void run() {
+                    expectNoMsg();
+
+                    Mockito.verify(mockListener, Mockito.never()).onDataTreeChanged(
+                            Matchers.anyCollectionOf(DataTreeCandidate.class));
+                }
+            };
+        }};
+    }
+
+    @SuppressWarnings({ "rawtypes", "unchecked" })
+    @Test
+    public void testDataChangedWithNoSender(){
+        new JavaTestKit(getSystem()) {{
+            final DataTreeCandidate mockTreeCandidate = Mockito.mock(DataTreeCandidate.class);
+            final ImmutableList<DataTreeCandidate> mockCandidates = ImmutableList.of(mockTreeCandidate);
+            final DOMDataTreeChangeListener mockListener = Mockito.mock(DOMDataTreeChangeListener.class);
+            final Props props = DataTreeChangeListenerActor.props(mockListener);
+            final ActorRef subject = getSystem().actorOf(props, "testDataTreeChangedWithNoSender");
+
+            getSystem().eventStream().subscribe(getRef(), DeadLetter.class);
+
+            subject.tell(new DataTreeChanged(mockCandidates), ActorRef.noSender());
+
+            // Make sure no DataChangedReply is sent to DeadLetters.
+            while(true) {
+                DeadLetter deadLetter;
+                try {
+                    deadLetter = expectMsgClass(duration("1 seconds"), DeadLetter.class);
+                } catch (AssertionError e) {
+                    // Timed out - got no DeadLetter - this is good
+                    break;
+                }
+
+                // We may get DeadLetters for other messages we don't care about.
+                Assert.assertFalse("Unexpected DataTreeChangedReply",
+                        deadLetter.message() instanceof DataTreeChangedReply);
+            }
+        }};
+    }
+
+    @SuppressWarnings({ "rawtypes", "unchecked" })
+    @Test
+    public void testDataChangedWithListenerRuntimeEx(){
+        new JavaTestKit(getSystem()) {{
+            final DataTreeCandidate mockTreeCandidate1 = Mockito.mock(DataTreeCandidate.class);
+            final ImmutableList<DataTreeCandidate> mockCandidates1 = ImmutableList.of(mockTreeCandidate1);
+            final DataTreeCandidate mockTreeCandidate2 = Mockito.mock(DataTreeCandidate.class);
+            final ImmutableList<DataTreeCandidate> mockCandidates2 = ImmutableList.of(mockTreeCandidate2);
+            final DataTreeCandidate mockTreeCandidate3 = Mockito.mock(DataTreeCandidate.class);
+            final ImmutableList<DataTreeCandidate> mockCandidates3 = ImmutableList.of(mockTreeCandidate3);
+
+            final DOMDataTreeChangeListener mockListener = Mockito.mock(DOMDataTreeChangeListener.class);
+            Mockito.doThrow(new RuntimeException("mock")).when(mockListener).onDataTreeChanged(mockCandidates2);
+
+            Props props = DataTreeChangeListenerActor.props(mockListener);
+            ActorRef subject = getSystem().actorOf(props, "testDataTreeChangedWithListenerRuntimeEx");
+
+            // Let the DataChangeListener know that notifications should be enabled
+            subject.tell(new EnableNotification(true), getRef());
+
+            subject.tell(new DataTreeChanged(mockCandidates1),getRef());
+            expectMsgClass(DataTreeChangedReply.class);
+
+            subject.tell(new DataTreeChanged(mockCandidates2),getRef());
+            expectMsgClass(DataTreeChangedReply.class);
+
+            subject.tell(new DataTreeChanged(mockCandidates3),getRef());
+            expectMsgClass(DataTreeChangedReply.class);
+
+            Mockito.verify(mockListener).onDataTreeChanged(mockCandidates1);
+            Mockito.verify(mockListener).onDataTreeChanged(mockCandidates2);
+            Mockito.verify(mockListener).onDataTreeChanged(mockCandidates3);
+        }};
+    }
+}
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerProxyTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerProxyTest.java
new file mode 100644 (file)
index 0000000..0dc0706
--- /dev/null
@@ -0,0 +1,240 @@
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.datastore;
+
+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;
+import akka.actor.Terminated;
+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.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
+import org.opendaylight.controller.cluster.datastore.messages.CloseDataTreeChangeListenerRegistration;
+import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
+import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
+import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
+import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener;
+import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListenerReply;
+import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.Dispatchers;
+import org.opendaylight.controller.cluster.datastore.utils.DoNothingActor;
+import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import scala.concurrent.ExecutionContextExecutor;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+
+public class DataTreeChangeListenerProxyTest extends AbstractActorTest {
+    @SuppressWarnings("unchecked")
+    private final DOMDataTreeChangeListener mockListener = mock(DOMDataTreeChangeListener.class);
+
+    @Test(timeout=10000)
+    public void testSuccessfulRegistration() {
+        new JavaTestKit(getSystem()) {{
+            ActorContext actorContext = new ActorContext(getSystem(), getRef(),
+                    mock(ClusterWrapper.class), mock(Configuration.class));
+
+            final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy =
+                    new DataTreeChangeListenerProxy<>(actorContext, mockListener);
+
+            final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
+            new Thread() {
+                @Override
+                public void run() {
+                    proxy.init("shard-1", path);
+                }
+
+            }.start();
+
+            FiniteDuration timeout = duration("5 seconds");
+            FindLocalShard findLocalShard = expectMsgClass(timeout, FindLocalShard.class);
+            Assert.assertEquals("getShardName", "shard-1", findLocalShard.getShardName());
+
+            reply(new LocalShardFound(getRef()));
+
+            RegisterDataTreeChangeListener registerMsg = expectMsgClass(timeout, RegisterDataTreeChangeListener.class);
+            Assert.assertEquals("getPath", path, registerMsg.getPath());
+
+            reply(new RegisterDataTreeChangeListenerReply(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, CloseDataTreeChangeListenerRegistration.class);
+
+            // The DataChangeListener actor should be terminated
+            expectMsgClass(timeout, Terminated.class);
+
+            proxy.close();
+
+            expectNoMsg();
+        }};
+    }
+
+    @Test(timeout=10000)
+    public void testLocalShardNotFound() {
+        new JavaTestKit(getSystem()) {{
+            ActorContext actorContext = new ActorContext(getSystem(), getRef(),
+                    mock(ClusterWrapper.class), mock(Configuration.class));
+
+            final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy =
+                    new DataTreeChangeListenerProxy<>(actorContext, mockListener);
+
+            final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
+            new Thread() {
+                @Override
+                public void run() {
+                    proxy.init("shard-1", path);
+                }
+
+            }.start();
+
+            FiniteDuration timeout = duration("5 seconds");
+            FindLocalShard findLocalShard = expectMsgClass(timeout, FindLocalShard.class);
+            Assert.assertEquals("getShardName", "shard-1", findLocalShard.getShardName());
+
+            reply(new LocalShardNotFound("shard-1"));
+
+            expectNoMsg(duration("1 seconds"));
+        }};
+    }
+
+    @Test(timeout=10000)
+    public void testLocalShardNotInitialized() {
+        new JavaTestKit(getSystem()) {{
+            ActorContext actorContext = new ActorContext(getSystem(), getRef(),
+                    mock(ClusterWrapper.class), mock(Configuration.class));
+
+            final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy =
+                    new DataTreeChangeListenerProxy<>(actorContext, mockListener);
+
+            final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
+            new Thread() {
+                @Override
+                public void run() {
+                    proxy.init("shard-1", path);
+                }
+
+            }.start();
+
+            FiniteDuration timeout = duration("5 seconds");
+            FindLocalShard findLocalShard = expectMsgClass(timeout, FindLocalShard.class);
+            Assert.assertEquals("getShardName", "shard-1", findLocalShard.getShardName());
+
+            reply(new NotInitializedException("not initialized"));
+
+            new Within(duration("1 seconds")) {
+                @Override
+                protected void run() {
+                    expectNoMsg();
+                }
+            };
+        }};
+    }
+
+    @Test
+    public void testFailedRegistration() {
+        new JavaTestKit(getSystem()) {{
+            ActorSystem mockActorSystem = mock(ActorSystem.class);
+
+            ActorRef mockActor = getSystem().actorOf(Props.create(DoNothingActor.class),
+                    "testFailedRegistration");
+            doReturn(mockActor).when(mockActorSystem).actorOf(any(Props.class));
+            ExecutionContextExecutor executor = ExecutionContexts.fromExecutor(
+                    MoreExecutors.sameThreadExecutor());
+
+
+            ActorContext actorContext = mock(ActorContext.class);
+            final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
+
+            doReturn(executor).when(actorContext).getClientDispatcher();
+            doReturn(mockActorSystem).when(actorContext).getActorSystem();
+
+            String shardName = "shard-1";
+            final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy =
+                    new DataTreeChangeListenerProxy<>(actorContext, mockListener);
+
+            doReturn(duration("5 seconds")).when(actorContext).getOperationDuration();
+            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("shard-1", path);
+
+            Assert.assertEquals("getListenerRegistrationActor", null,
+                    proxy.getListenerRegistrationActor());
+        }};
+    }
+
+    @Test
+    public void testCloseBeforeRegistration() {
+        new JavaTestKit(getSystem()) {{
+            ActorContext actorContext = mock(ActorContext.class);
+
+            String shardName = "shard-1";
+
+            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));
+
+            final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy =
+                    new DataTreeChangeListenerProxy<>(actorContext, mockListener);
+
+
+            Answer<Future<Object>> answer = new Answer<Future<Object>>() {
+                @Override
+                public Future<Object> answer(InvocationOnMock invocation) {
+                    proxy.close();
+                    return Futures.successful((Object)new RegisterDataTreeChangeListenerReply(getRef()));
+                }
+            };
+
+            doAnswer(answer).when(actorContext).executeOperationAsync(any(ActorRef.class),
+                    any(Object.class), any(Timeout.class));
+
+            proxy.init(shardName, YangInstanceIdentifier.of(TestModel.TEST_QNAME));
+
+            expectMsgClass(duration("5 seconds"), CloseDataTreeChangeListenerRegistration.class);
+
+            Assert.assertEquals("getListenerRegistrationActor", null,
+                    proxy.getListenerRegistrationActor());
+        }};
+    }
+}
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerRegistrationActorTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerRegistrationActorTest.java
new file mode 100644 (file)
index 0000000..5695911
--- /dev/null
@@ -0,0 +1,44 @@
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.datastore;
+
+import akka.actor.ActorRef;
+import akka.actor.Props;
+import akka.testkit.JavaTestKit;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.opendaylight.controller.cluster.datastore.messages.CloseDataTreeChangeListenerRegistration;
+import org.opendaylight.controller.cluster.datastore.messages.CloseDataTreeChangeListenerRegistrationReply;
+import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
+import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
+import org.opendaylight.yangtools.concepts.ListenerRegistration;
+
+public class DataTreeChangeListenerRegistrationActorTest extends AbstractActorTest {
+    private static final InMemoryDOMDataStore store = new InMemoryDOMDataStore("OPER", MoreExecutors.sameThreadExecutor());
+
+    static {
+        store.onGlobalContextUpdated(TestModel.createTestContext());
+    }
+
+    @SuppressWarnings("rawtypes")
+    @Test
+    public void testOnReceiveCloseListenerRegistration() throws Exception {
+        new JavaTestKit(getSystem()) {{
+            final ListenerRegistration mockListenerReg = Mockito.mock(ListenerRegistration.class);
+            final Props props = DataTreeChangeListenerRegistrationActor.props(mockListenerReg);
+            final ActorRef subject = getSystem().actorOf(props, "testCloseListenerRegistration");
+
+            subject.tell(CloseDataTreeChangeListenerRegistration.getInstance(), getRef());
+
+            expectMsgClass(duration("1 second"), CloseDataTreeChangeListenerRegistrationReply.class);
+
+            Mockito.verify(mockListenerReg).close();
+        }};
+    }
+}
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ForwardingDataTreeChangeListenerTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ForwardingDataTreeChangeListenerTest.java
new file mode 100644 (file)
index 0000000..fb8baf1
--- /dev/null
@@ -0,0 +1,37 @@
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.datastore;
+
+import akka.actor.ActorRef;
+import akka.actor.Props;
+import java.util.Arrays;
+import java.util.Collection;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.opendaylight.controller.cluster.datastore.messages.DataTreeChanged;
+import org.opendaylight.controller.cluster.datastore.utils.MessageCollectorActor;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
+
+public class ForwardingDataTreeChangeListenerTest extends AbstractActorTest {
+
+    @Test
+    public void testOnDataChanged() throws Exception {
+        final Props props = Props.create(MessageCollectorActor.class);
+        final ActorRef actorRef = getSystem().actorOf(props);
+
+        ForwardingDataTreeChangeListener forwardingListener = new ForwardingDataTreeChangeListener(
+                getSystem().actorSelection(actorRef.path()));
+
+        Collection<DataTreeCandidate> expected = Arrays.asList(Mockito.mock(DataTreeCandidate.class));
+        forwardingListener.onDataTreeChanged(expected);
+
+        DataTreeChanged actual = MessageCollectorActor.expectFirstMatching(actorRef, DataTreeChanged.class);
+        Assert.assertSame(expected, actual.getChanges());
+    }
+}
index 3d28672..4cbc121 100644 (file)
@@ -57,6 +57,8 @@ import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
+import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener;
+import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListenerReply;
 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
 import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
@@ -66,6 +68,7 @@ import org.opendaylight.controller.cluster.datastore.modification.MutableComposi
 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
 import org.opendaylight.controller.cluster.datastore.utils.MessageCollectorActor;
 import org.opendaylight.controller.cluster.datastore.utils.MockDataChangeListener;
+import org.opendaylight.controller.cluster.datastore.utils.MockDataTreeChangeListener;
 import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
 import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener;
 import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListenerReply;
@@ -246,6 +249,110 @@ public class ShardTest extends AbstractShardTest {
         }};
     }
 
+    @Test
+    public void testRegisterDataTreeChangeListener() throws Exception {
+        new ShardTestKit(getSystem()) {{
+            TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
+                    newShardProps(), "testRegisterDataTreeChangeListener");
+
+            waitUntilLeader(shard);
+
+            shard.tell(new UpdateSchemaContext(SchemaContextHelper.full()), ActorRef.noSender());
+
+            MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
+            ActorRef dclActor = getSystem().actorOf(DataTreeChangeListenerActor.props(listener),
+                    "testRegisterDataTreeChangeListener-DataTreeChangeListener");
+
+            shard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor), getRef());
+
+            RegisterDataTreeChangeListenerReply reply = expectMsgClass(duration("3 seconds"),
+                    RegisterDataTreeChangeListenerReply.class);
+            String replyPath = reply.getListenerRegistrationPath().toString();
+            assertTrue("Incorrect reply path: " + replyPath, replyPath.matches(
+                    "akka:\\/\\/test\\/user\\/testRegisterDataTreeChangeListener\\/\\$.*"));
+
+            YangInstanceIdentifier path = TestModel.TEST_PATH;
+            writeToStore(shard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+
+            listener.waitForChangeEvents();
+
+            dclActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
+            shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
+        }};
+    }
+
+    @SuppressWarnings("serial")
+    @Test
+    public void testDataTreeChangeListenerNotifiedWhenNotTheLeaderOnRegistration() throws Exception {
+        new ShardTestKit(getSystem()) {{
+            final CountDownLatch onFirstElectionTimeout = new CountDownLatch(1);
+            final CountDownLatch onChangeListenerRegistered = new CountDownLatch(1);
+            Creator<Shard> creator = new Creator<Shard>() {
+                boolean firstElectionTimeout = true;
+
+                @Override
+                public Shard create() throws Exception {
+                    return new Shard(shardID, Collections.<String,String>emptyMap(),
+                            dataStoreContextBuilder.persistent(false).build(), SCHEMA_CONTEXT) {
+                        @Override
+                        public void onReceiveCommand(final Object message) throws Exception {
+                            if(message instanceof ElectionTimeout && firstElectionTimeout) {
+                                firstElectionTimeout = false;
+                                final ActorRef self = getSelf();
+                                new Thread() {
+                                    @Override
+                                    public void run() {
+                                        Uninterruptibles.awaitUninterruptibly(
+                                                onChangeListenerRegistered, 5, TimeUnit.SECONDS);
+                                        self.tell(message, self);
+                                    }
+                                }.start();
+
+                                onFirstElectionTimeout.countDown();
+                            } else {
+                                super.onReceiveCommand(message);
+                            }
+                        }
+                    };
+                }
+            };
+
+            MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
+            ActorRef dclActor = getSystem().actorOf(DataTreeChangeListenerActor.props(listener),
+                    "testDataTreeChangeListenerNotifiedWhenNotTheLeaderOnRegistration-DataChangeListener");
+
+            TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
+                    Props.create(new DelegatingShardCreator(creator)),
+                    "testDataTreeChangeListenerNotifiedWhenNotTheLeaderOnRegistration");
+
+            YangInstanceIdentifier path = TestModel.TEST_PATH;
+            writeToStore(shard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+
+            assertEquals("Got first ElectionTimeout", true,
+                    onFirstElectionTimeout.await(5, TimeUnit.SECONDS));
+
+            shard.tell(new RegisterDataTreeChangeListener(path, dclActor), getRef());
+            RegisterDataTreeChangeListenerReply reply = expectMsgClass(duration("5 seconds"),
+                    RegisterDataTreeChangeListenerReply.class);
+            assertNotNull("getListenerRegistratioznPath", reply.getListenerRegistrationPath());
+
+            shard.tell(new FindLeader(), getRef());
+            FindLeaderReply findLeadeReply =
+                    expectMsgClass(duration("5 seconds"), FindLeaderReply.class);
+            assertNull("Expected the shard not to be the leader", findLeadeReply.getLeaderActor());
+
+            writeToStore(shard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+
+            onChangeListenerRegistered.countDown();
+
+            // TODO: investigate why we do not receive data chage events
+            listener.waitForChangeEvents();
+
+            dclActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
+            shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
+        }};
+    }
+
     @Test
     public void testCreateTransaction(){
         new ShardTestKit(getSystem()) {{
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/utils/MockDataTreeChangeListener.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/utils/MockDataTreeChangeListener.java
new file mode 100644 (file)
index 0000000..d06fc43
--- /dev/null
@@ -0,0 +1,61 @@
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.datastore.utils;
+
+import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.Uninterruptibles;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
+
+import javax.annotation.Nonnull;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+public class MockDataTreeChangeListener implements DOMDataTreeChangeListener {
+
+    private final List<Collection<DataTreeCandidate>> changeList =
+            Collections.synchronizedList(Lists.<Collection<DataTreeCandidate>>newArrayList());
+
+    private volatile CountDownLatch changeLatch;
+    private int expChangeEventCount;
+
+    public MockDataTreeChangeListener(int expChangeEventCount) {
+        reset(expChangeEventCount);
+    }
+
+    public void reset(int expChangeEventCount) {
+        changeLatch = new CountDownLatch(expChangeEventCount);
+        this.expChangeEventCount = expChangeEventCount;
+        changeList.clear();
+    }
+
+    @Override
+    public void onDataTreeChanged(@Nonnull final Collection<DataTreeCandidate> changes) {
+        changeList.add(changes);
+        changeLatch.countDown();
+    }
+
+    public void waitForChangeEvents() {
+        boolean done = Uninterruptibles.awaitUninterruptibly(changeLatch, 5, TimeUnit.SECONDS);
+        if(!done) {
+            fail(String.format("Missing change notifications. Expected: %d. Actual: %d",
+                    expChangeEventCount, (expChangeEventCount - changeLatch.getCount())));
+        }
+    }
+
+    public void expectNoMoreChanges(String assertMsg) {
+        Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
+        assertEquals(assertMsg, expChangeEventCount, changeList.size());
+    }
+}

©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.