2 * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
9 package org.opendaylight.openflowplugin.applications.frsync.impl;
11 import com.google.common.util.concurrent.Futures;
12 import com.google.common.util.concurrent.ListenableFuture;
13 import com.google.common.util.concurrent.ListeningExecutorService;
14 import com.google.common.util.concurrent.MoreExecutors;
15 import com.google.common.util.concurrent.ThreadFactoryBuilder;
16 import java.util.ArrayList;
17 import java.util.List;
18 import java.util.concurrent.CountDownLatch;
19 import java.util.concurrent.ExecutorService;
20 import java.util.concurrent.Executors;
21 import java.util.concurrent.TimeUnit;
22 import org.junit.After;
23 import org.junit.Before;
24 import org.junit.Test;
25 import org.junit.runner.RunWith;
26 import org.mockito.InOrder;
27 import org.mockito.Matchers;
28 import org.mockito.Mock;
29 import org.mockito.Mockito;
30 import org.mockito.invocation.InvocationOnMock;
31 import org.mockito.runners.MockitoJUnitRunner;
32 import org.mockito.stubbing.Answer;
33 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
34 import org.opendaylight.openflowplugin.applications.frsync.SyncReactor;
35 import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
37 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
38 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
39 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
40 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
41 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
42 import org.slf4j.Logger;
43 import org.slf4j.LoggerFactory;
46 * Test for {@link SyncReactorFutureZipDecorator}.
48 @RunWith(MockitoJUnitRunner.class)
49 public class SyncReactorFutureZipDecoratorTest {
51 private static final Logger LOG = LoggerFactory.getLogger(SyncReactorFutureZipDecoratorTest.class);
52 private static final NodeId NODE_ID = new NodeId("testNode");
53 private SyncReactorFutureZipDecorator reactor;
54 private InstanceIdentifier<FlowCapableNode> fcNodePath;
55 private ListeningExecutorService syncThreadPool;
56 private final LogicalDatastoreType configDS = LogicalDatastoreType.CONFIGURATION;
57 private final LogicalDatastoreType operationalDS = LogicalDatastoreType.OPERATIONAL;
60 private SyncReactor delegate;
62 private SyncupEntry syncupEntry;
66 final ExecutorService executorService = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder()
68 .setNameFormat("frsync-test-%d")
69 .setUncaughtExceptionHandler((thread, e) -> LOG.error("Uncaught exception {}", thread, e))
71 syncThreadPool = MoreExecutors.listeningDecorator(executorService);
72 reactor = new SyncReactorFutureZipDecorator(delegate, syncThreadPool);
73 fcNodePath = InstanceIdentifier.create(Nodes.class).child(Node.class, new NodeKey(NODE_ID))
74 .augmentation(FlowCapableNode.class);
78 public void testSyncupWithOptimizedConfigDeltaCompression() throws Exception {
79 final FlowCapableNode dataBefore = Mockito.mock(FlowCapableNode.class);
80 final FlowCapableNode dataAfter = Mockito.mock(FlowCapableNode.class);
81 final FlowCapableNode dataAfter2 = Mockito.mock(FlowCapableNode.class);
82 final CountDownLatch latchForFirst = new CountDownLatch(1);
83 final CountDownLatch latchForNext = new CountDownLatch(1);
85 final SyncupEntry first = new SyncupEntry(dataBefore, configDS, null, operationalDS);
86 final SyncupEntry second = new SyncupEntry(dataAfter, configDS, dataBefore, configDS);
87 final SyncupEntry third = new SyncupEntry(null, configDS, dataAfter, configDS);
88 final SyncupEntry fourth = new SyncupEntry(dataAfter2, configDS, null, configDS);
89 final SyncupEntry zipped = new SyncupEntry(dataAfter2, configDS, dataBefore, configDS);
90 final List<ListenableFuture<Boolean>> allResults = new ArrayList<>();
92 Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Mockito.eq(first)))
93 .thenAnswer(new Answer<ListenableFuture<Boolean>>() {
95 public ListenableFuture<Boolean> answer(final InvocationOnMock invocationOnMock) throws Throwable {
96 LOG.info("unlocking next configs");
97 latchForNext.countDown();
98 latchForFirst.await();
99 LOG.info("unlocking first delegate");
100 return Futures.immediateFuture(Boolean.TRUE);
104 allResults.add(reactor.syncup(fcNodePath, first));
105 latchForNext.await();
107 mockSyncupWithEntry(second);
108 allResults.add(reactor.syncup(fcNodePath, second));
109 mockSyncupWithEntry(third);
110 allResults.add(reactor.syncup(fcNodePath, third));
111 mockSyncupWithEntry(fourth);
112 allResults.add(reactor.syncup(fcNodePath, fourth));
113 latchForFirst.countDown();
115 Futures.allAsList(allResults).get(1, TimeUnit.SECONDS);
116 LOG.info("all configs done");
118 syncThreadPool.shutdown();
119 boolean terminated = syncThreadPool.awaitTermination(1, TimeUnit.SECONDS);
121 LOG.info("thread pool not terminated.");
122 syncThreadPool.shutdownNow();
124 final InOrder inOrder = Mockito.inOrder(delegate);
125 inOrder.verify(delegate).syncup(fcNodePath, first);
126 inOrder.verify(delegate).syncup(fcNodePath, zipped);
127 inOrder.verifyNoMoreInteractions();
131 public void testSyncupConfigEmptyQueue() throws Exception {
132 final FlowCapableNode dataBefore = Mockito.mock(FlowCapableNode.class);
133 final FlowCapableNode dataAfter = Mockito.mock(FlowCapableNode.class);
134 final CountDownLatch latchForNext = new CountDownLatch(1);
136 final SyncupEntry first = new SyncupEntry(dataBefore, configDS, null, configDS);
137 final SyncupEntry second = new SyncupEntry(dataAfter, configDS, dataBefore, configDS);
139 Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Mockito.eq(first)))
140 .thenAnswer(new Answer<ListenableFuture<Boolean>>() {
142 public ListenableFuture<Boolean> answer(final InvocationOnMock invocationOnMock) throws Throwable {
143 LOG.info("unlocking next config");
144 latchForNext.countDown();
145 return Futures.immediateFuture(Boolean.TRUE);
149 reactor.syncup(fcNodePath, first);
150 latchForNext.await();
151 mockSyncupWithEntry(second);
152 reactor.syncup(fcNodePath, second);
154 boolean terminated = syncThreadPool.awaitTermination(1, TimeUnit.SECONDS);
156 LOG.info("thread pool not terminated.");
157 syncThreadPool.shutdownNow();
159 final InOrder inOrder = Mockito.inOrder(delegate);
160 inOrder.verify(delegate).syncup(fcNodePath, first);
161 inOrder.verify(delegate).syncup(fcNodePath, second);
162 inOrder.verifyNoMoreInteractions();
166 public void testSyncupRewriteZipEntryWithOperationalDelta() throws Exception {
167 final FlowCapableNode configBefore = Mockito.mock(FlowCapableNode.class);
168 final FlowCapableNode configAfter = Mockito.mock(FlowCapableNode.class);
169 final FlowCapableNode configActual = Mockito.mock(FlowCapableNode.class);
170 final FlowCapableNode freshOperational = Mockito.mock(FlowCapableNode.class);
171 final CountDownLatch latchForFirst = new CountDownLatch(1);
172 final CountDownLatch latchForNext = new CountDownLatch(1);
174 final SyncupEntry first = new SyncupEntry(configAfter, configDS, configBefore, configDS);
175 final SyncupEntry second = new SyncupEntry(configActual, configDS, freshOperational, operationalDS);
177 Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Mockito.eq(first)))
178 .thenAnswer(new Answer<ListenableFuture<Boolean>>() {
180 public ListenableFuture<Boolean> answer(final InvocationOnMock invocationOnMock) throws Throwable {
181 LOG.info("unlocking for fresh operational");
182 latchForNext.countDown();
183 latchForFirst.await();
184 LOG.info("unlocking first delegate");
185 return Futures.immediateFuture(Boolean.TRUE);
189 reactor.syncup(fcNodePath, first);
190 latchForNext.await();
192 mockSyncupWithEntry(second);
193 reactor.syncup(fcNodePath, second);
194 latchForFirst.countDown();
196 syncThreadPool.shutdown();
197 boolean terminated = syncThreadPool.awaitTermination(1, TimeUnit.SECONDS);
199 LOG.info("thread pool not terminated.");
200 syncThreadPool.shutdownNow();
202 Mockito.verify(delegate, Mockito.times(1)).syncup(fcNodePath, second);
205 private void mockSyncupWithEntry(final SyncupEntry entry) throws InterruptedException {
206 Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Mockito.eq(entry)))
207 .thenReturn(Futures.immediateFuture(Boolean.TRUE));
211 public void tearDown() {
212 syncThreadPool.shutdownNow();