package org.opendaylight.openflowplugin.applications.frsync;
import com.google.common.util.concurrent.ListenableFuture;
-import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
public interface SyncReactor {
/**
* @param flowcapableNodePath path to openflow augmentation of node
- * @param configTree configured node
- * @param operationalTree device reflection
- * @param dsType type of DS change
+ * @param syncupEntry configured node + device reflection
* @return synchronization outcome
* @throws InterruptedException
*/
ListenableFuture<Boolean> syncup(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
- final FlowCapableNode configTree, final FlowCapableNode operationalTree,
- final LogicalDatastoreType dsType) throws InterruptedException;
+ final SyncupEntry syncupEntry) throws InterruptedException;
}
NODES_IID.child(Node.class, new NodeKey(nodeId)).augmentation(FlowCapableNode.class);
return roTx.read(logicalDatastoreType, path).checkedGet(5000, TimeUnit.MILLISECONDS);
} catch (ReadFailedException | TimeoutException e) {
- LOG.error("error reading {}", nodeId, e);
+ LOG.error("error reading {} -> {}", nodeId.getValue(), e);
}
return Optional.absent();
if (optFuture.isPresent()) {
final ListenableFuture<Boolean> future = optFuture.get();
final Boolean ret = future.get(15000, TimeUnit.MILLISECONDS);
- LOG.debug("syncup ret {} {} {} thread:{}", dsType(), ret, nodeId.getValue(), threadName());
+ LOG.debug("syncup return in {} listener for: {} [{}] thread:{}", dsType(), nodeId.getValue(), ret, threadName());
}
} catch (InterruptedException e) {
LOG.warn("permit for forwarding rules sync not acquired: {}", nodeId.getValue());
import org.opendaylight.openflowplugin.applications.frsync.dao.FlowCapableNodeDao;
import org.opendaylight.openflowplugin.applications.frsync.dao.FlowCapableNodeSnapshotDao;
import org.opendaylight.openflowplugin.applications.frsync.util.PathUtil;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
final FlowCapableNode dataAfter,
final FlowCapableNode operationalNode) throws InterruptedException {
NodeId nodeId = PathUtil.digNodeId(nodePath);
- LOG.trace("onNodeAdded {}", nodeId);
- return reactor.syncup(nodePath, dataAfter, operationalNode, dsType());
+ LOG.trace("onNodeAdded {}", nodeId.getValue());
+ final SyncupEntry syncupEntry = new SyncupEntry(dataAfter, dsType(), operationalNode, LogicalDatastoreType.OPERATIONAL);
+ return reactor.syncup(nodePath, syncupEntry);
}
/**
final FlowCapableNode dataBefore,
final FlowCapableNode dataAfter) throws InterruptedException {
NodeId nodeId = PathUtil.digNodeId(nodePath);
- LOG.trace("onNodeUpdated {}", nodeId);
- return reactor.syncup(nodePath, dataAfter, dataBefore, dsType());
+ LOG.trace("onNodeUpdated {}", nodeId.getValue());
+ final SyncupEntry syncupEntry = new SyncupEntry(dataAfter, dsType(), dataBefore, dsType());
+ return reactor.syncup(nodePath, syncupEntry);
}
/**
private ListenableFuture<Boolean> onNodeDeleted(final InstanceIdentifier<FlowCapableNode> nodePath,
final FlowCapableNode dataBefore) throws InterruptedException {
NodeId nodeId = PathUtil.digNodeId(nodePath);
- LOG.trace("onNodeDeleted {}", nodeId);
- return reactor.syncup(nodePath, null, dataBefore, dsType());
+ LOG.trace("onNodeDeleted {}", nodeId.getValue());
+ final SyncupEntry syncupEntry = new SyncupEntry(null, dsType(), dataBefore, dsType());
+ return reactor.syncup(nodePath, syncupEntry);
}
@Override
import org.opendaylight.openflowplugin.applications.frsync.util.ModificationUtil;
import org.opendaylight.openflowplugin.applications.frsync.util.PathUtil;
import org.opendaylight.openflowplugin.applications.frsync.util.ReconciliationRegistry;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableStatisticsGatheringStatus;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.snapshot.gathering.status.grouping.SnapshotGatheringStatusEnd;
final InstanceIdentifier<FlowCapableNode> nodePath = InstanceIdentifier.create(Nodes.class)
.child(Node.class, new NodeKey(ModificationUtil.nodeId(modification)))
.augmentation(FlowCapableNode.class);
- final FlowCapableNode fcNode = ModificationUtil.flowCapableNodeAfter(modification);
- return Optional.of(reactor.syncup(nodePath, nodeConfiguration.get(), fcNode, dsType()));
+ final FlowCapableNode fcOperationalNode = ModificationUtil.flowCapableNodeAfter(modification);
+ final SyncupEntry syncupEntry = new SyncupEntry(nodeConfiguration.get(), LogicalDatastoreType.CONFIGURATION,
+ fcOperationalNode, dsType());
+ return Optional.of(reactor.syncup(nodePath, syncupEntry));
} else {
LOG.debug("Config not present for reconciliation: {}", nodeId.getValue());
return skipModification(modification);
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.openflowplugin.applications.frsync.SyncReactor;
import org.opendaylight.openflowplugin.applications.frsync.impl.clustering.DeviceMastershipManager;
import org.opendaylight.openflowplugin.applications.frsync.util.PathUtil;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
@Override
public ListenableFuture<Boolean> syncup(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
- final FlowCapableNode configTree,
- final FlowCapableNode operationalTree,
- final LogicalDatastoreType dsType) throws InterruptedException {
+ final SyncupEntry syncupEntry) throws InterruptedException {
final NodeId nodeId = PathUtil.digNodeId(flowcapableNodePath);
- LOG.trace("syncup cluster {}", nodeId.getValue());
+ LOG.trace("Syncup cluster decorator: {}", nodeId.getValue());
if (!deviceMastershipManager.isDeviceMastered(nodeId)) {
LOG.debug("Skip syncup since not master for: {}", nodeId.getValue());
return Futures.immediateFuture(Boolean.TRUE);
} else {
- return delegate.syncup(flowcapableNodePath, configTree,operationalTree, dsType);
+ return delegate.syncup(flowcapableNodePath, syncupEntry);
}
}
}
import com.google.common.util.concurrent.ListeningExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.openflowplugin.applications.frsync.SyncReactor;
import org.opendaylight.openflowplugin.applications.frsync.util.PathUtil;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
}
public ListenableFuture<Boolean> syncup(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
- final FlowCapableNode configTree, final FlowCapableNode operationalTree,
- final LogicalDatastoreType dsType) throws InterruptedException {
+ final SyncupEntry syncupEntry) throws InterruptedException {
final NodeId nodeId = PathUtil.digNodeId(flowcapableNodePath);
- LOG.trace("syncup future {}", nodeId.getValue());
+ LOG.trace("syncup future decorator: {}", nodeId.getValue());
final ListenableFuture<Boolean> syncup = executorService.submit(() -> {
final String oldThreadName = updateThreadName(nodeId);
try {
- final Boolean ret = doSyncupInFuture(flowcapableNodePath, configTree, operationalTree, dsType)
+ final Boolean ret = doSyncupInFuture(flowcapableNodePath, syncupEntry)
.get(10000, TimeUnit.MILLISECONDS);
- LOG.trace("ret {} {}", nodeId.getValue(), ret);
+ LOG.trace("syncup return in future decorator: {} [{}]", nodeId.getValue(), ret);
return true;
} catch (TimeoutException e) {
LOG.error("doSyncupInFuture timeout occured {}", nodeId.getValue(), e);
}
protected ListenableFuture<Boolean> doSyncupInFuture(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
- final FlowCapableNode configTree, final FlowCapableNode operationalTree,
- final LogicalDatastoreType dsType) throws InterruptedException {
+ final SyncupEntry syncupEntry) throws InterruptedException {
final NodeId nodeId = PathUtil.digNodeId(flowcapableNodePath);
- LOG.trace("doSyncupInFuture future {}", nodeId.getValue());
+ LOG.trace("doSyncupInFuture future decorator: {}", nodeId.getValue());
- return delegate.syncup(flowcapableNodePath, configTree, operationalTree, dsType);
+ return delegate.syncup(flowcapableNodePath, syncupEntry);
}
private String updateThreadName(NodeId nodeId) {
import java.util.Map;
import java.util.concurrent.Semaphore;
import javax.annotation.concurrent.GuardedBy;
-import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.openflowplugin.applications.frsync.SyncReactor;
import org.opendaylight.openflowplugin.applications.frsync.util.PathUtil;
-import org.opendaylight.openflowplugin.applications.frsync.util.ZipQueueEntry;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
private static final Logger LOG = LoggerFactory.getLogger(SyncReactorFutureZipDecorator.class);
@GuardedBy("compressionGuard")
- private final Map<InstanceIdentifier<FlowCapableNode>, ZipQueueEntry> compressionQueue = new HashMap<>();
+ private final Map<InstanceIdentifier<FlowCapableNode>, SyncupEntry> compressionQueue = new HashMap<>();
private final Semaphore compressionGuard = new Semaphore(1, false);
public SyncReactorFutureZipDecorator(SyncReactor delegate, ListeningExecutorService executorService) {
}
public ListenableFuture<Boolean> syncup(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
- final FlowCapableNode configTree, final FlowCapableNode operationalTree,
- final LogicalDatastoreType dsType) throws InterruptedException {
+ final SyncupEntry syncupEntry) throws InterruptedException {
final NodeId nodeId = PathUtil.digNodeId(flowcapableNodePath);
- LOG.trace("syncup zip {}", nodeId.getValue());
+ LOG.trace("syncup zip decorator: {}", nodeId.getValue());
try {
compressionGuard.acquire();
- final boolean newFutureNecessary = updateCompressionState(flowcapableNodePath, configTree, operationalTree, dsType);
+ final boolean newFutureNecessary = updateCompressionState(flowcapableNodePath, syncupEntry);
if (newFutureNecessary) {
- super.syncup(flowcapableNodePath, configTree, operationalTree, dsType);
+ super.syncup(flowcapableNodePath, syncupEntry);
}
return Futures.immediateFuture(true);
} finally {
}
protected ListenableFuture<Boolean> doSyncupInFuture(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
- final FlowCapableNode configTree, final FlowCapableNode operationalTree,
- final LogicalDatastoreType dsType) throws InterruptedException {
+ final SyncupEntry syncupEntry) throws InterruptedException {
final NodeId nodeId = PathUtil.digNodeId(flowcapableNodePath);
- LOG.trace("doSyncupInFuture zip {}", nodeId.getValue());
+ LOG.trace("doSyncupInFuture zip decorator: {}", nodeId.getValue());
- final ZipQueueEntry lastCompressionState = removeLastCompressionState(flowcapableNodePath);
+ final SyncupEntry lastCompressionState = removeLastCompressionState(flowcapableNodePath);
if (lastCompressionState == null) {
return Futures.immediateFuture(true);
} else {
- return super.doSyncupInFuture(flowcapableNodePath,
- lastCompressionState.getLeft(), lastCompressionState.getRight(), dsType);
+ return super.doSyncupInFuture(flowcapableNodePath, lastCompressionState);
}
}
* update its zip queue entry. Create/replace zip queue entry for the device with operational delta otherwise.
*/
private boolean updateCompressionState(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
- final FlowCapableNode configTree, final FlowCapableNode operationalTree,
- final LogicalDatastoreType dsType) {
- final ZipQueueEntry previousEntry = compressionQueue.get(flowcapableNodePath);
+ final SyncupEntry syncupEntry) {
+ final SyncupEntry previousEntry = compressionQueue.get(flowcapableNodePath);
- if (previousEntry != null && dsType == LogicalDatastoreType.CONFIGURATION
- && previousEntry.getDsType() == LogicalDatastoreType.CONFIGURATION) {
- putOptimizedConfigDelta(flowcapableNodePath, configTree, previousEntry);
+ if (previousEntry != null && syncupEntry.isOptimizedConfigDelta() && previousEntry.isOptimizedConfigDelta()) {
+ updateOptimizedConfigDelta(flowcapableNodePath, syncupEntry, previousEntry);
} else {
- putLatestOperationalDelta(flowcapableNodePath, configTree, operationalTree, dsType);
+ compressionQueue.put(flowcapableNodePath, syncupEntry);
}
return previousEntry == null;
}
- private void putOptimizedConfigDelta(InstanceIdentifier<FlowCapableNode> flowcapableNodePath, FlowCapableNode configTree,
- ZipQueueEntry previous) {
- compressionQueue.put(flowcapableNodePath, new ZipQueueEntry(configTree, previous.getRight(), previous.getDsType()));
+ private void updateOptimizedConfigDelta(InstanceIdentifier<FlowCapableNode> flowcapableNodePath, SyncupEntry actual,
+ SyncupEntry previous) {
+ compressionQueue.put(flowcapableNodePath, new SyncupEntry(actual.getAfter(), actual.getDsTypeAfter(),
+ previous.getBefore(), previous.getDsTypeBefore()));
}
- private void putLatestOperationalDelta(InstanceIdentifier<FlowCapableNode> flowcapableNodePath, FlowCapableNode configTree,
- FlowCapableNode operationalTree, LogicalDatastoreType dsType) {
- compressionQueue.put(flowcapableNodePath, new ZipQueueEntry(configTree, operationalTree, dsType));
- }
-
- private ZipQueueEntry removeLastCompressionState(
+ private SyncupEntry removeLastCompressionState(
final InstanceIdentifier<FlowCapableNode> flowcapableNodePath) {
try {
try {
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
-import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.openflowplugin.applications.frsync.SemaphoreKeeper;
import org.opendaylight.openflowplugin.applications.frsync.SyncReactor;
import org.opendaylight.openflowplugin.applications.frsync.util.PathUtil;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
}
public ListenableFuture<Boolean> syncup(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
- final FlowCapableNode configTree, final FlowCapableNode operationalTree,
- final LogicalDatastoreType dsType) throws InterruptedException {
+ final SyncupEntry syncupEntry) throws InterruptedException {
final NodeId nodeId = PathUtil.digNodeId(flowcapableNodePath);
- LOG.trace("syncup guard {}", nodeId.getValue());
+ LOG.trace("syncup guard decorator: {}", nodeId.getValue());
final long stampBeforeGuard = System.nanoTime();
final Semaphore guard = summonGuardAndAcquire(flowcapableNodePath);
}
final ListenableFuture<Boolean> endResult =
- delegate.syncup(flowcapableNodePath, configTree, operationalTree, dsType);
+ delegate.syncup(flowcapableNodePath, syncupEntry);
Futures.addCallback(endResult, new FutureCallback<Boolean>() {
@Override
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
-import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.openflowplugin.applications.frsync.SyncPlanPushStrategy;
import org.opendaylight.openflowplugin.applications.frsync.SyncReactor;
import org.opendaylight.openflowplugin.applications.frsync.impl.strategy.SynchronizationDiffInput;
import org.opendaylight.openflowplugin.applications.frsync.util.PathUtil;
import org.opendaylight.openflowplugin.applications.frsync.util.ReconcileUtil;
import org.opendaylight.openflowplugin.applications.frsync.util.SyncCrudCounters;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
@Override
public ListenableFuture<Boolean> syncup(final InstanceIdentifier<FlowCapableNode> nodeIdent,
- final FlowCapableNode configTree, final FlowCapableNode operationalTree,
- final LogicalDatastoreType dsType) {
-
+ final SyncupEntry syncupEntry) {
final NodeId nodeId = PathUtil.digNodeId(nodeIdent);
- LOG.trace("syncup impl {} cfg:{} oper:{}", nodeId.getValue(), configTree == null ? "is null" : "non null", operationalTree == null ? "is null" : "non null");
+ FlowCapableNode configTree = syncupEntry.getAfter();
+ FlowCapableNode operationalTree = syncupEntry.getBefore();
+
+ LOG.trace("syncup reactor {} cfg:{} oper:{}",
+ nodeId.getValue(),
+ configTree == null ? "is null" : "non null",
+ operationalTree == null ? "is null" : "non null");
final SyncCrudCounters counters = new SyncCrudCounters();
/**
final CrudCounts flowCrudCounts = counters.getFlowCrudCounts();
final CrudCounts meterCrudCounts = counters.getMeterCrudCounts();
final CrudCounts groupCrudCounts = counters.getGroupCrudCounts();
- LOG.debug("sync-outcome[{}] (added/updated/removed): flow={}/{}/{}, meter={}/{}/{}, group={}/{}/{}, took={} ms",
+ LOG.debug("syncup outcome[{}] (added/updated/removed): flow={}/{}/{}, meter={}/{}/{}, group={}/{}/{}, took={} ms",
nodeId.getValue(),
flowCrudCounts.getAdded(),
flowCrudCounts.getUpdated(),
import com.google.common.base.Function;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.openflowplugin.applications.frsync.SyncReactor;
import org.opendaylight.openflowplugin.applications.frsync.util.PathUtil;
import org.opendaylight.openflowplugin.applications.frsync.util.ReconciliationRegistry;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
}
public ListenableFuture<Boolean> syncup(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
- final FlowCapableNode configTree, final FlowCapableNode operationalTree,
- final LogicalDatastoreType dsType) throws InterruptedException {
+ final SyncupEntry syncupEntry) throws InterruptedException {
final NodeId nodeId = PathUtil.digNodeId(flowcapableNodePath);
- LOG.trace("syncup retry {}", nodeId.getValue());
+ LOG.trace("syncup retry decorator: {}", nodeId.getValue());
- if (dsType == LogicalDatastoreType.CONFIGURATION && reconciliationRegistry.isRegistered(nodeId)) {
+ if (syncupEntry.isOptimizedConfigDelta() && reconciliationRegistry.isRegistered(nodeId)) {
LOG.debug("Config change ignored because {} is in reconcile.", nodeId.getValue());
return Futures.immediateFuture(Boolean.FALSE);
}
- ListenableFuture<Boolean> syncupResult = delegate.syncup(flowcapableNodePath, configTree, operationalTree, dsType);
+ ListenableFuture<Boolean> syncupResult = delegate.syncup(flowcapableNodePath,syncupEntry);
return Futures.transform(syncupResult, new Function<Boolean, Boolean>() {
@Override
public Boolean apply(Boolean result) {
- LOG.trace("syncup ret in retry {}", result);
+ LOG.trace("syncup return in retry decorator: {} [{}]", nodeId.getValue(), result);
if (result) {
reconciliationRegistry.unregisterIfRegistered(nodeId);
return true;
@Override
public void instantiateServiceInstance() {
+ LOG.debug("FRS started for: {}", nodeId.getValue());
deviceMastered = true;
reconciliationRegistry.register(nodeId);
- LOG.trace("FRS started for: {}", nodeId.getValue());
}
@Override
public ListenableFuture<Void> closeServiceInstance() {
+ LOG.debug("FRS stopped for: {}", nodeId.getValue());
deviceMastered = false;
reconciliationRegistry.unregisterIfRegistered(nodeId);
- LOG.debug("FRS stopped for: {}", nodeId.getValue());
return Futures.immediateFuture(null);
}
LOG.debug(prefix + " failed: {} -> {}", nodeId.getValue(), Arrays.toString(errors.toArray()));
}
} else {
- LOG.debug(prefix + "reconciliation failed: {} -> null result", nodeId.getValue());
+ LOG.debug(prefix + " reconciliation failed: {} -> null result", nodeId.getValue());
}
}
@Override
public void onFailure(final Throwable t) {
- LOG.debug(prefix + "reconciliation failed seriously: {}", nodeId.getValue(), t);
+ LOG.debug(prefix + " reconciliation failed seriously: {}", nodeId.getValue(), t);
}
};
}
final Map<MeterId, Meter> meterOperationalMap,
final List<Meter> metersConfigured,
final boolean gatherUpdates) {
- LOG.trace("resolving meters for {}", nodeId);
+ LOG.trace("resolving meters for {}", nodeId.getValue());
final ItemSyncBox<Meter> syncBox = new ItemSyncBox<>();
for (Meter meter : metersConfigured) {
final Meter existingMeter = meterOperationalMap.get(meter.getMeterId());
final Map<Short, Table> tableOperationalMap,
final List<Table> tablesConfigured,
final boolean gatherUpdates) {
- LOG.trace("resolving flows in tables for {}", nodeId);
+ LOG.trace("resolving flows in tables for {}", nodeId.getValue());
final Map<TableKey, ItemSyncBox<Flow>> tableFlowSyncBoxes = new HashMap<>();
for (final Table tableConfigured : tablesConfigured) {
final List<Flow> flowsConfigured = tableConfigured.getFlow();
--- /dev/null
+/**
+ * Copyright (c) 2016 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.openflowplugin.applications.frsync.util;
+
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
+
+/**
+ * Data entry of before and after data for syncup in {@link org.opendaylight.openflowplugin.applications.frsync.SyncReactor}.
+ */
+public class SyncupEntry {
+ private final FlowCapableNode after;
+ private final LogicalDatastoreType dsTypeAfter;
+ private final FlowCapableNode before;
+ private final LogicalDatastoreType dsTypeBefore;
+
+ public SyncupEntry(final FlowCapableNode after, final LogicalDatastoreType dsTypeAfter,
+ final FlowCapableNode before, final LogicalDatastoreType dsTypeBefore) {
+ this.after = after;
+ this.dsTypeAfter = dsTypeAfter;
+ this.before = before;
+ this.dsTypeBefore = dsTypeBefore;
+ }
+
+ public FlowCapableNode getAfter() {
+ return after;
+ }
+
+ public FlowCapableNode getBefore() {
+ return before;
+ }
+
+ public LogicalDatastoreType getDsTypeAfter() {
+ return dsTypeAfter;
+ }
+
+ public LogicalDatastoreType getDsTypeBefore() {
+ return dsTypeBefore;
+ }
+
+ public boolean isOptimizedConfigDelta() {
+ return dsTypeAfter == LogicalDatastoreType.CONFIGURATION && dsTypeBefore == LogicalDatastoreType.CONFIGURATION;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ SyncupEntry that = (SyncupEntry) o;
+
+ if (after != null ? !after.equals(that.after) : that.after != null) return false;
+ if (dsTypeAfter != that.dsTypeAfter) return false;
+ if (before != null ? !before.equals(that.before) : that.before != null) return false;
+ return dsTypeBefore == that.dsTypeBefore;
+ }
+
+}
+++ /dev/null
-/**
- * Copyright (c) 2016 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.openflowplugin.applications.frsync.util;
-
-import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
-import org.opendaylight.openflowplugin.applications.frsync.impl.SyncReactorFutureZipDecorator;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
-
-/**
- * Simple compression queue entry for {@link SyncReactorFutureZipDecorator}.
- */
-public class ZipQueueEntry {
- private final FlowCapableNode after;
- private final FlowCapableNode before;
- private final LogicalDatastoreType dsTypeBefore;
-
- public ZipQueueEntry(final FlowCapableNode after, final FlowCapableNode before,
- final LogicalDatastoreType dsTypeBefore) {
- this.after = after;
- this.before = before;
- this.dsTypeBefore = dsTypeBefore;
- }
-
- public FlowCapableNode getLeft() {
- return after;
- }
-
- public FlowCapableNode getRight() {
- return before;
- }
-
- public LogicalDatastoreType getDsType() {
- return dsTypeBefore;
- }
-
-}
import org.opendaylight.openflowplugin.applications.frsync.dao.FlowCapableNodeDao;
import org.opendaylight.openflowplugin.applications.frsync.dao.FlowCapableNodeOdlDao;
import org.opendaylight.openflowplugin.applications.frsync.dao.FlowCapableNodeSnapshotDao;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
private static final NodeId NODE_ID = new NodeId("testNode");
private InstanceIdentifier<FlowCapableNode> fcNodePath;
private SimplifiedConfigListener nodeListenerConfig;
- private final LogicalDatastoreType dsType = LogicalDatastoreType.CONFIGURATION;
+ private final LogicalDatastoreType confgDS = LogicalDatastoreType.CONFIGURATION;
+ private final LogicalDatastoreType operationalDS = LogicalDatastoreType.OPERATIONAL;
@Mock
private SyncReactor reactor;
Mockito.when(db.newReadOnlyTransaction()).thenReturn(roTx);
Mockito.when(dataTreeModification.getRootPath()).thenReturn(dataTreeIdentifier);
Mockito.when(dataTreeModification.getRootNode()).thenReturn(configModification);
- Mockito.when(reactor.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Matchers.<FlowCapableNode>any(),
- Matchers.<FlowCapableNode>any(), Matchers.<LogicalDatastoreType>any())).thenReturn(Futures.immediateFuture(Boolean.TRUE));
}
@Test
}
@Test
- public void testOnDataTreeChangedSyncupAdd() throws InterruptedException {
- Mockito.when(roTx.read(LogicalDatastoreType.OPERATIONAL, fcNodePath))
- .thenReturn(Futures.immediateCheckedFuture(Optional.of(dataBefore)));
+ public void testOnDataTreeChangedAdd() throws InterruptedException {
Mockito.when(configModification.getDataBefore()).thenReturn(null);
Mockito.when(configModification.getDataAfter()).thenReturn(dataAfter);
+ final SyncupEntry syncupEntry = loadOperationalDSAndPrepareSyncupEntry(dataAfter, confgDS, dataBefore, operationalDS);
nodeListenerConfig.onDataTreeChanged(Collections.singleton(dataTreeModification));
- Mockito.verify(reactor).syncup(fcNodePath, dataAfter, dataBefore, dsType);
+ Mockito.verify(reactor).syncup(fcNodePath, syncupEntry);
Mockito.verifyNoMoreInteractions(reactor);
Mockito.verify(roTx).close();
}
@Test
- public void testOnDataTreeChangedSyncupUpdate() throws InterruptedException {
- Mockito.when(roTx.read(LogicalDatastoreType.OPERATIONAL, fcNodePath))
- .thenReturn(Futures.immediateCheckedFuture(Optional.of(dataBefore)));
+ public void testOnDataTreeChangedUpdate() throws InterruptedException {
Mockito.when(configModification.getDataBefore()).thenReturn(dataBefore);
Mockito.when(configModification.getDataAfter()).thenReturn(dataAfter);
+ final SyncupEntry syncupEntry = loadOperationalDSAndPrepareSyncupEntry(dataAfter, confgDS, dataBefore, confgDS);
nodeListenerConfig.onDataTreeChanged(Collections.singleton(dataTreeModification));
- Mockito.verify(reactor).syncup(fcNodePath, dataAfter, dataBefore, dsType);
+ Mockito.verify(reactor).syncup(fcNodePath, syncupEntry);
Mockito.verifyNoMoreInteractions(reactor);
Mockito.verify(roTx).close();
}
@Test
- public void testOnDataTreeChangedSyncupDelete() throws InterruptedException {
- Mockito.when(roTx.read(LogicalDatastoreType.OPERATIONAL, fcNodePath))
- .thenReturn(Futures.immediateCheckedFuture(Optional.of(dataBefore)));
+ public void testOnDataTreeChangedDelete() throws InterruptedException {
Mockito.when(configModification.getDataBefore()).thenReturn(dataBefore);
Mockito.when(configModification.getDataAfter()).thenReturn(null);
+ final SyncupEntry syncupEntry = loadOperationalDSAndPrepareSyncupEntry(null, confgDS, dataBefore, confgDS);
nodeListenerConfig.onDataTreeChanged(Collections.singleton(dataTreeModification));
- Mockito.verify(reactor).syncup(fcNodePath, null, dataBefore, dsType);
+ Mockito.verify(reactor).syncup(fcNodePath, syncupEntry);
Mockito.verifyNoMoreInteractions(reactor);
Mockito.verify(roTx).close();
}
Mockito.verify(roTx).close();
}
+ private SyncupEntry loadOperationalDSAndPrepareSyncupEntry(final FlowCapableNode after, final LogicalDatastoreType dsTypeAfter,
+ final FlowCapableNode before, final LogicalDatastoreType dsTypeBefore) throws InterruptedException {
+ Mockito.when(roTx.read(LogicalDatastoreType.OPERATIONAL, fcNodePath))
+ .thenReturn(Futures.immediateCheckedFuture(Optional.of(dataBefore)));
+ final SyncupEntry syncupEntry = new SyncupEntry(after, dsTypeAfter, before, dsTypeBefore);
+ Mockito.when(reactor.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Mockito.eq(syncupEntry)))
+ .thenReturn(Futures.immediateFuture(Boolean.TRUE));
+ return syncupEntry;
+ }
+
}
\ No newline at end of file
import org.opendaylight.openflowplugin.applications.frsync.dao.FlowCapableNodeSnapshotDao;
import org.opendaylight.openflowplugin.applications.frsync.impl.clustering.DeviceMastershipManager;
import org.opendaylight.openflowplugin.applications.frsync.util.ReconciliationRegistry;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev130715.DateAndTime;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableStatisticsGatheringStatus;
private static final NodeId NODE_ID = new NodeId("testNode");
private InstanceIdentifier<FlowCapableNode> fcNodePath;
private SimplifiedOperationalListener nodeListenerOperational;
- private final LogicalDatastoreType dsType = LogicalDatastoreType.OPERATIONAL;
+ private final LogicalDatastoreType configDS = LogicalDatastoreType.CONFIGURATION;
+ private final LogicalDatastoreType operationalDS = LogicalDatastoreType.OPERATIONAL;
private final SimpleDateFormat simpleDateFormat = new SimpleDateFormat(SimplifiedOperationalListener.DATE_AND_TIME_FORMAT);
@Mock
Mockito.when(reconciliationRegistry.isRegistered(NODE_ID)).thenReturn(true);
operationalAdd();
prepareFreshOperational(true);
-
- Mockito.when(roTx.read(LogicalDatastoreType.CONFIGURATION, fcNodePath))
- .thenReturn(Futures.immediateCheckedFuture(Optional.of(configNode)));
- Mockito.when(reactor.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Matchers.<FlowCapableNode>any(),
- Matchers.<FlowCapableNode>any(), Matchers.<LogicalDatastoreType>any())).thenReturn(Futures.immediateFuture(Boolean.TRUE));
+ final SyncupEntry syncupEntry = loadConfigDSAndPrepareSyncupEntry(configNode, configDS, fcOperationalNode, operationalDS);
nodeListenerOperational.onDataTreeChanged(Collections.singleton(dataTreeModification));
Mockito.verify(deviceMastershipManager).onDeviceConnected(NODE_ID);
- Mockito.verify(reactor).syncup(fcNodePath, configNode, fcOperationalNode, dsType);
+ Mockito.verify(reactor).syncup(fcNodePath, syncupEntry);
Mockito.verify(roTx).close();
}
}
@Test
- public void testOnDataTreeChangedSyncupDeletePhysical() throws Exception {
+ public void testOnDataTreeChangedDeletePhysical() throws Exception {
Mockito.when(operationalModification.getDataBefore()).thenReturn(operationalNode);
Mockito.when(operationalModification.getDataAfter()).thenReturn(null);
Mockito.when(dataTreeModification.getRootNode().getModificationType()).thenReturn(ModificationType.DELETE);
}
@Test
- public void testOnDataTreeChangedSyncupDeleteLogical() {
+ public void testOnDataTreeChangedDeleteLogical() {
Mockito.when(operationalModification.getDataBefore()).thenReturn(operationalNode);
List<NodeConnector> nodeConnectorList = Mockito.mock(List.class);
Mockito.when(operationalNode.getNodeConnector()).thenReturn(nodeConnectorList);
Mockito.when(reconciliationRegistry.isRegistered(NODE_ID)).thenReturn(true);
operationalUpdate();
prepareFreshOperational(true);
-
- Mockito.when(roTx.read(LogicalDatastoreType.CONFIGURATION, fcNodePath))
- .thenReturn(Futures.immediateCheckedFuture(Optional.of(configNode)));
- Mockito.when(reactor.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Matchers.<FlowCapableNode>any(),
- Matchers.<FlowCapableNode>any(), Matchers.<LogicalDatastoreType>any())).thenReturn(Futures.immediateFuture(Boolean.TRUE));
+ final SyncupEntry syncupEntry = loadConfigDSAndPrepareSyncupEntry(configNode, configDS, fcOperationalNode, operationalDS);
nodeListenerOperational.onDataTreeChanged(Collections.singleton(dataTreeModification));
- Mockito.verify(reactor).syncup(fcNodePath, configNode, fcOperationalNode, dsType);
+ Mockito.verify(reactor).syncup(fcNodePath, syncupEntry);
Mockito.verify(roTx).close();
}
Mockito.when(operationalModification.getDataBefore()).thenReturn(operationalNode);
Mockito.when(operationalModification.getDataAfter()).thenReturn(operationalNode);
}
+
+ private SyncupEntry loadConfigDSAndPrepareSyncupEntry(final FlowCapableNode after, final LogicalDatastoreType dsTypeAfter,
+ final FlowCapableNode before, final LogicalDatastoreType dsTypeBefore) throws InterruptedException {
+ Mockito.when(roTx.read(LogicalDatastoreType.CONFIGURATION, fcNodePath))
+ .thenReturn(Futures.immediateCheckedFuture(Optional.of(configNode)));
+ final SyncupEntry syncupEntry = new SyncupEntry(after, dsTypeAfter, before, dsTypeBefore);
+ Mockito.when(reactor.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Mockito.eq(syncupEntry)))
+ .thenReturn(Futures.immediateFuture(Boolean.TRUE));
+ return syncupEntry;
+ }
}
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.runners.MockitoJUnitRunner;
-import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.openflowplugin.applications.frsync.impl.clustering.DeviceMastershipManager;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
@Mock
private DeviceMastershipManager deviceMastershipManager;
@Mock
- private FlowCapableNode fcConfigNode;
- @Mock
- private FlowCapableNode fcOperationalNode;
+ private SyncupEntry syncupEntry;
@Before
public void setUp() {
public void testSyncupMaster() throws InterruptedException {
Mockito.when(deviceMastershipManager.isDeviceMastered(NODE_ID)).thenReturn(true);
- reactor.syncup(fcNodePath, fcConfigNode, fcOperationalNode, LogicalDatastoreType.CONFIGURATION);
+ reactor.syncup(fcNodePath, syncupEntry);
- Mockito.verify(delegate).syncup(fcNodePath, fcConfigNode, fcOperationalNode, LogicalDatastoreType.CONFIGURATION);
+ Mockito.verify(delegate).syncup(fcNodePath, syncupEntry);
Mockito.verifyNoMoreInteractions(delegate);
}
public void testSyncupSlave() throws InterruptedException {
Mockito.when(deviceMastershipManager.isDeviceMastered(NODE_ID)).thenReturn(false);
- reactor.syncup(fcNodePath, fcConfigNode, fcOperationalNode, LogicalDatastoreType.CONFIGURATION);
+ reactor.syncup(fcNodePath, syncupEntry);
Mockito.verifyZeroInteractions(delegate);
}
import org.mockito.runners.MockitoJUnitRunner;
import org.mockito.stubbing.Answer;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
private SyncReactorFutureZipDecorator reactor;
private InstanceIdentifier<FlowCapableNode> fcNodePath;
private ListeningExecutorService syncThreadPool;
+ private final LogicalDatastoreType configDS = LogicalDatastoreType.CONFIGURATION;
+ private final LogicalDatastoreType operationalDS = LogicalDatastoreType.OPERATIONAL;
@Mock
private SyncReactorGuardDecorator delegate;
+ @Mock
+ private SyncupEntry syncupEntry;
@Before
public void setUp() {
final FlowCapableNode dataAfter2 = Mockito.mock(FlowCapableNode.class);
final CountDownLatch latchForFirst = new CountDownLatch(1);
final CountDownLatch latchForNext = new CountDownLatch(1);
- final LogicalDatastoreType dsType = LogicalDatastoreType.CONFIGURATION;
- Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Matchers.<FlowCapableNode>any(),
- Matchers.<FlowCapableNode>any(), Matchers.<LogicalDatastoreType>any())).thenAnswer(new Answer<ListenableFuture<Boolean>>() {
+ final SyncupEntry first = new SyncupEntry(dataBefore, configDS, null, configDS);
+ final SyncupEntry second = new SyncupEntry(dataAfter, configDS, dataBefore, configDS);
+ final SyncupEntry third = new SyncupEntry(null, configDS, dataAfter, configDS);
+ final SyncupEntry fourth = new SyncupEntry(dataAfter2, configDS, null, configDS);
+ final SyncupEntry zipped = new SyncupEntry(dataAfter2, configDS, dataBefore, configDS);
+ final List<ListenableFuture<Boolean>> allResults = new ArrayList<>();
+
+ Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Mockito.eq(first)))
+ .thenAnswer(new Answer<ListenableFuture<Boolean>>() {
@Override
public ListenableFuture<Boolean> answer(final InvocationOnMock invocationOnMock) throws Throwable {
LOG.info("unlocking next configs");
LOG.info("unlocking first delegate");
return Futures.immediateFuture(Boolean.TRUE);
}
- }).thenReturn(Futures.immediateFuture(Boolean.TRUE));
+ });
- final List<ListenableFuture<Boolean>> allResults = new ArrayList<>();
- allResults.add(reactor.syncup(fcNodePath, dataBefore, null, dsType));
+ allResults.add(reactor.syncup(fcNodePath, first));
latchForNext.await();
- allResults.add(reactor.syncup(fcNodePath, dataAfter, dataBefore, dsType));
- allResults.add(reactor.syncup(fcNodePath, null, dataAfter, dsType));
- allResults.add(reactor.syncup(fcNodePath, dataAfter2, null, dsType));
+ mockSyncupWithEntry(second);
+ allResults.add(reactor.syncup(fcNodePath, second));
+ mockSyncupWithEntry(third);
+ allResults.add(reactor.syncup(fcNodePath, third));
+ mockSyncupWithEntry(fourth);
+ allResults.add(reactor.syncup(fcNodePath, fourth));
latchForFirst.countDown();
Futures.allAsList(allResults).get(1, TimeUnit.SECONDS);
syncThreadPool.shutdownNow();
}
final InOrder inOrder = Mockito.inOrder(delegate);
- inOrder.verify(delegate).syncup(fcNodePath, dataBefore, null, dsType);
- inOrder.verify(delegate).syncup(fcNodePath, dataAfter2, dataBefore, dsType);
+ inOrder.verify(delegate).syncup(fcNodePath, first);
+ inOrder.verify(delegate).syncup(fcNodePath, zipped);
inOrder.verifyNoMoreInteractions();
}
final FlowCapableNode dataBefore = Mockito.mock(FlowCapableNode.class);
final FlowCapableNode dataAfter = Mockito.mock(FlowCapableNode.class);
final CountDownLatch latchForNext = new CountDownLatch(1);
- final LogicalDatastoreType dsType = LogicalDatastoreType.CONFIGURATION;
- Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Matchers.<FlowCapableNode>any(),
- Matchers.<FlowCapableNode>any(), Matchers.<LogicalDatastoreType>any())).thenAnswer(new Answer<ListenableFuture<Boolean>>() {
+ final SyncupEntry first = new SyncupEntry(dataBefore, configDS, null, configDS);
+ final SyncupEntry second = new SyncupEntry(dataAfter, configDS, dataBefore, configDS);
+
+ Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Mockito.eq(first)))
+ .thenAnswer(new Answer<ListenableFuture<Boolean>>() {
@Override
public ListenableFuture<Boolean> answer(final InvocationOnMock invocationOnMock) throws Throwable {
LOG.info("unlocking next config");
latchForNext.countDown();
return Futures.immediateFuture(Boolean.TRUE);
}
- }).thenReturn(Futures.immediateFuture(Boolean.TRUE));
+ });
- reactor.syncup(fcNodePath, dataBefore, null, dsType);
+ reactor.syncup(fcNodePath, first);
latchForNext.await();
- reactor.syncup(fcNodePath, dataAfter, dataBefore, dsType);
+ mockSyncupWithEntry(second);
+ reactor.syncup(fcNodePath, second);
boolean terminated = syncThreadPool.awaitTermination(1, TimeUnit.SECONDS);
if (!terminated) {
syncThreadPool.shutdownNow();
}
final InOrder inOrder = Mockito.inOrder(delegate);
- inOrder.verify(delegate).syncup(fcNodePath, dataBefore, null, dsType);
- inOrder.verify(delegate).syncup(fcNodePath, dataAfter, dataBefore, dsType);
+ inOrder.verify(delegate).syncup(fcNodePath, first);
+ inOrder.verify(delegate).syncup(fcNodePath, second);
inOrder.verifyNoMoreInteractions();
-
}
@Test
final CountDownLatch latchForFirst = new CountDownLatch(1);
final CountDownLatch latchForNext = new CountDownLatch(1);
- Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Matchers.<FlowCapableNode>any(),
- Matchers.<FlowCapableNode>any(), Matchers.<LogicalDatastoreType>any())).thenAnswer(new Answer<ListenableFuture<Boolean>>() {
+ final SyncupEntry first = new SyncupEntry(configAfter, configDS, configBefore, configDS);
+ final SyncupEntry second = new SyncupEntry(configActual, configDS, freshOperational, operationalDS);
+
+ Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Mockito.eq(first)))
+ .thenAnswer(new Answer<ListenableFuture<Boolean>>() {
@Override
public ListenableFuture<Boolean> answer(final InvocationOnMock invocationOnMock) throws Throwable {
LOG.info("unlocking for fresh operational");
LOG.info("unlocking first delegate");
return Futures.immediateFuture(Boolean.TRUE);
}
- }).thenReturn(Futures.immediateFuture(Boolean.TRUE));
+ });
- reactor.syncup(fcNodePath, configAfter, configBefore, LogicalDatastoreType.CONFIGURATION);
+ reactor.syncup(fcNodePath, first);
latchForNext.await();
- reactor.syncup(fcNodePath, configActual, freshOperational, LogicalDatastoreType.OPERATIONAL);
+ mockSyncupWithEntry(second);
+ reactor.syncup(fcNodePath, second);
latchForFirst.countDown();
syncThreadPool.shutdown();
LOG.info("thread pool not terminated.");
syncThreadPool.shutdownNow();
}
- Mockito.verify(delegate, Mockito.times(1)).syncup(fcNodePath, configActual, freshOperational, LogicalDatastoreType.OPERATIONAL);
+ Mockito.verify(delegate, Mockito.times(1)).syncup(fcNodePath, second);
+ }
+
+ private void mockSyncupWithEntry(final SyncupEntry entry) throws InterruptedException {
+ Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Mockito.eq(entry)))
+ .thenReturn(Futures.immediateFuture(Boolean.TRUE));
}
@After
import org.mockito.runners.MockitoJUnitRunner;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.openflowplugin.applications.frsync.util.SemaphoreKeeperGuavaImpl;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
private FlowCapableNode fcConfigNode;
@Mock
private FlowCapableNode fcOperationalNode;
+ @Mock
+ private SyncupEntry syncupEntry;
@Before
public void setUp() throws Exception {
@Test
public void testSyncupSuccess() throws Exception {
- Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Matchers.<FlowCapableNode>any(),
- Matchers.<FlowCapableNode>any(), Matchers.<LogicalDatastoreType>any())).thenReturn(Futures.immediateFuture(Boolean.TRUE));
+ Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Matchers.<SyncupEntry>any()))
+ .thenReturn(Futures.immediateFuture(Boolean.TRUE));
- reactor.syncup(fcNodePath, fcConfigNode, fcOperationalNode, dsType);
+// reactor.syncup(fcNodePath, fcConfigNode, fcOperationalNode, dsType);
+ reactor.syncup(fcNodePath, syncupEntry);
- Mockito.verify(delegate).syncup(fcNodePath, fcConfigNode, fcOperationalNode, dsType);
+// Mockito.verify(delegate).syncup(fcNodePath, fcConfigNode, fcOperationalNode, dsType);
+ Mockito.verify(delegate).syncup(fcNodePath, syncupEntry);
Mockito.verifyNoMoreInteractions(delegate);
}
@Test
public void testSyncupFail() throws Exception {
- Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Matchers.<FlowCapableNode>any(),
- Matchers.<FlowCapableNode>any(), Matchers.<LogicalDatastoreType>any())).thenReturn(Futures.immediateFailedFuture(new Exception()));
+ Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Matchers.<SyncupEntry>any()))
+ .thenReturn(Futures.immediateFailedFuture(new Exception()));
- reactor.syncup(fcNodePath, fcConfigNode, fcOperationalNode, dsType);
+// reactor.syncup(fcNodePath, fcConfigNode, fcOperationalNode, dsType);
+ reactor.syncup(fcNodePath, syncupEntry);
- Mockito.verify(delegate).syncup(fcNodePath, fcConfigNode, fcOperationalNode, dsType);
+// Mockito.verify(delegate).syncup(fcNodePath, fcConfigNode, fcOperationalNode, dsType);
+ Mockito.verify(delegate).syncup(fcNodePath, syncupEntry);
Mockito.verifyNoMoreInteractions(delegate);
}
import org.opendaylight.openflowplugin.applications.frsync.impl.strategy.SynchronizationDiffInput;
import org.opendaylight.openflowplugin.applications.frsync.util.ReconcileUtil;
import org.opendaylight.openflowplugin.applications.frsync.util.SyncCrudCounters;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
.setMeter(Collections.singletonList(DSInputFactory.createMeter(2L)))
.build();
+ final SyncupEntry syncupEntry = new SyncupEntry(configFcn, LogicalDatastoreType.CONFIGURATION,
+ operationalFcn, LogicalDatastoreType.OPERATIONAL);
+
Mockito.when(syncPlanPushStrategy.executeSyncStrategy(
Matchers.<ListenableFuture<RpcResult<Void>>>any(),
Matchers.<SynchronizationDiffInput>any(),
Matchers.<SyncCrudCounters>any()))
.thenReturn(RpcResultBuilder.<Void>success().buildFuture());
- final ListenableFuture<Boolean> syncupResult = reactor.syncup(NODE_IDENT, configFcn, operationalFcn, LogicalDatastoreType.CONFIGURATION);
+ final ListenableFuture<Boolean> syncupResult = reactor.syncup(NODE_IDENT, syncupEntry);
try {
Assert.assertTrue(syncupResult.isDone());
final Boolean voidRpcResult = syncupResult.get(2, TimeUnit.SECONDS);
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.runners.MockitoJUnitRunner;
-import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.openflowplugin.applications.frsync.util.ReconciliationRegistry;
+import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
private static final NodeId NODE_ID = new NodeId("test-node");
private SyncReactorRetryDecorator reactor;
private InstanceIdentifier<FlowCapableNode> fcNodePath;
- private final LogicalDatastoreType dsType = LogicalDatastoreType.CONFIGURATION;
@Mock
private SyncReactorImpl delegate;
@Mock
private ReconciliationRegistry reconciliationRegistry;
@Mock
- private FlowCapableNode fcConfigNode;
- @Mock
- private FlowCapableNode fcOperationalNode;
+ private SyncupEntry syncupEntry;
@Before
public void setUp() {
reactor = new SyncReactorRetryDecorator(delegate, reconciliationRegistry);
InstanceIdentifier<Node> nodePath = InstanceIdentifier.create(Nodes.class).child(Node.class, new NodeKey(NODE_ID));
fcNodePath = nodePath.augmentation(FlowCapableNode.class);
-
- final Node operationalNode = Mockito.mock(Node.class);
- Mockito.when(operationalNode.getId()).thenReturn(NODE_ID);
- Mockito.when(operationalNode.getAugmentation(FlowCapableNode.class)).thenReturn(fcOperationalNode);
}
@Test
public void testSyncupSuccess() throws InterruptedException {
- Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Matchers.<FlowCapableNode>any(),
- Matchers.<FlowCapableNode>any(), Matchers.<LogicalDatastoreType>any())).thenReturn(Futures.immediateFuture(Boolean.TRUE));
+ Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Matchers.<SyncupEntry>any()))
+ .thenReturn(Futures.immediateFuture(Boolean.TRUE));
- reactor.syncup(fcNodePath, fcConfigNode, fcOperationalNode, dsType);
+ reactor.syncup(fcNodePath, syncupEntry);
- Mockito.verify(delegate).syncup(fcNodePath, fcConfigNode, fcOperationalNode, dsType);
+ Mockito.verify(delegate).syncup(fcNodePath, syncupEntry);
Mockito.verifyNoMoreInteractions(delegate);
Mockito.verify(reconciliationRegistry).unregisterIfRegistered(NODE_ID);
}
@Test
public void testSyncupFail() throws InterruptedException {
- Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Matchers.<FlowCapableNode>any(),
- Matchers.<FlowCapableNode>any(), Matchers.<LogicalDatastoreType>any())).thenReturn(Futures.immediateFuture(Boolean.FALSE));
+ Mockito.when(delegate.syncup(Matchers.<InstanceIdentifier<FlowCapableNode>>any(), Matchers.<SyncupEntry>any()))
+ .thenReturn(Futures.immediateFuture(Boolean.FALSE));
- reactor.syncup(fcNodePath, fcConfigNode, fcOperationalNode, dsType);
+ reactor.syncup(fcNodePath, syncupEntry);
- Mockito.verify(delegate).syncup(fcNodePath, fcConfigNode, fcOperationalNode, dsType);
+ Mockito.verify(delegate).syncup(fcNodePath, syncupEntry);
Mockito.verifyNoMoreInteractions(delegate);
Mockito.verify(reconciliationRegistry).register(NODE_ID);
}
@Test
public void testSyncupConfigIgnoreInRetry() throws InterruptedException {
Mockito.when(reconciliationRegistry.isRegistered(NODE_ID)).thenReturn(true);
+ Mockito.when(syncupEntry.isOptimizedConfigDelta()).thenReturn(true);
- reactor.syncup(fcNodePath, fcConfigNode, fcOperationalNode, dsType);
+ reactor.syncup(fcNodePath, syncupEntry);
Mockito.verifyZeroInteractions(delegate);
}
Code Review / openflowplugin.git / commitdiff