import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.CheckedFuture;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.JdkFutureAdapters;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
-import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
import org.opendaylight.openflowplugin.api.OFConstants;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
import org.opendaylight.openflowplugin.applications.frm.FlowNodeReconciliation;
import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.MeterId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.AddBundleMessagesInput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.AddBundleMessagesInputBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.AddBundleMessagesOutput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.ControlBundleInput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.ControlBundleInputBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.ControlBundleOutput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.SalBundleService;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.add.bundle.messages.input.Messages;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.add.bundle.messages.input.MessagesBuilder;
}
}
- private ListenableFuture<Boolean> reconcileConfiguration(InstanceIdentifier<FlowCapableNode> connectedNode) {
+ @Override
+ public ListenableFuture<Boolean> reconcileConfiguration(InstanceIdentifier<FlowCapableNode> connectedNode) {
LOG.info("Triggering reconciliation for device {}", connectedNode.firstKeyOf(Node.class));
if (provider.isStaleMarkingEnabled()) {
LOG.info("Stale-Marking is ENABLED and proceeding with deletion of " + "stale-marked entities on switch {}",
connectedNode.toString());
reconciliationPreProcess(connectedNode);
}
- LOG.debug("Bundle based reconciliation status : {}",
- provider.isBundleBasedReconciliationEnabled() ? "Enable" : "Disable");
if (provider.isBundleBasedReconciliationEnabled()) {
BundleBasedReconciliationTask bundleBasedReconTask = new BundleBasedReconciliationTask(connectedNode);
return JdkFutureAdapters.listenInPoolThread(executor.submit(bundleBasedReconTask));
Optional<FlowCapableNode> flowNode = Optional.absent();
BundleId bundleIdValue = new BundleId(BUNDLE_ID.getAndIncrement());
BigInteger dpnId = getDpnIdFromNodeName(node);
- LOG.debug("Triggering bundle based reconciliation for device :{}", dpnId);
+ LOG.info("Triggering bundle based reconciliation for device : {}", dpnId);
ReadOnlyTransaction trans = provider.getReadTranaction();
try {
flowNode = trans.read(LogicalDatastoreType.CONFIGURATION, nodeIdentity).get();
LOG.debug("FlowNode present for Datapath ID {}", dpnId);
final NodeRef nodeRef = new NodeRef(nodeIdentity.firstIdentifierOf(Node.class));
+ final ControlBundleInput closeBundleInput = new ControlBundleInputBuilder().setNode(nodeRef)
+ .setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
+ .setType(BundleControlType.ONFBCTCLOSEREQUEST).build();
+
final ControlBundleInput openBundleInput = new ControlBundleInputBuilder().setNode(nodeRef)
.setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS).setType(BundleControlType.ONFBCTOPENREQUEST)
.build();
.setNode(nodeRef).setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
.setMessages(createMessages(nodeRef, flowNode)).build();
- Future<RpcResult<Void>> openBundle = salBundleService.controlBundle(openBundleInput);
+ /* Close previously opened bundle on the openflow switch if any */
+ ListenableFuture<RpcResult<ControlBundleOutput>> closeBundle
+ = salBundleService.controlBundle(closeBundleInput);
+
+ /* Open a new bundle on the switch */
+ ListenableFuture<RpcResult<ControlBundleOutput>> openBundle =
+ Futures.transformAsync(closeBundle,
+ rpcResult -> salBundleService.controlBundle(openBundleInput),
+ MoreExecutors.directExecutor());
- ListenableFuture<RpcResult<Void>> addBundleMessagesFuture = Futures
- .transformAsync(JdkFutureAdapters.listenInPoolThread(openBundle), rpcResult -> {
+ /* Push groups and flows via bundle add messages */
+ ListenableFuture<RpcResult<AddBundleMessagesOutput>> addBundleMessagesFuture
+ = Futures.transformAsync(openBundle, rpcResult -> {
if (rpcResult.isSuccessful()) {
- return JdkFutureAdapters
- .listenInPoolThread(salBundleService.addBundleMessages(addBundleMessagesInput));
+ return salBundleService.addBundleMessages(addBundleMessagesInput);
}
return Futures.immediateFuture(null);
- });
- ListenableFuture<RpcResult<Void>> commitBundleFuture = Futures.transformAsync(addBundleMessagesFuture,
- rpcResult -> {
- if (rpcResult.isSuccessful()) {
- return JdkFutureAdapters
- .listenInPoolThread(salBundleService.controlBundle(commitBundleInput));
- }
- return Futures.immediateFuture(null);
- });
+ }, MoreExecutors.directExecutor());
+
+ /* Commit the bundle on the openflow switch */
+ ListenableFuture<RpcResult<ControlBundleOutput>> commitBundleFuture
+ = Futures.transformAsync(addBundleMessagesFuture, rpcResult -> {
+ if (rpcResult.isSuccessful()) {
+ return salBundleService.controlBundle(commitBundleInput);
+ }
+ return Futures.immediateFuture(null);
+ }, MoreExecutors.directExecutor());
/* Bundles not supported for meters */
List<Meter> meters = flowNode.get().getMeter() != null ? flowNode.get().getMeter()
if (rpcResult.isSuccessful()) {
for (Meter meter : meters) {
final KeyedInstanceIdentifier<Meter, MeterKey> meterIdent = nodeIdentity
- .child(Meter.class, meter.getKey());
+ .child(Meter.class, meter.key());
provider.getMeterCommiter().add(meterIdent, meter, nodeIdentity);
}
}
return Futures.immediateFuture(null);
- });
+ }, MoreExecutors.directExecutor());
trans.close();
try {
BigInteger dpnId = getDpnIdFromNodeName(node);
ReadOnlyTransaction trans = provider.getReadTranaction();
- Optional<FlowCapableNode> flowNode = Optional.absent();
+ Optional<FlowCapableNode> flowNode;
// initialize the counter
int counter = 0;
try {
? flowNode.get().getTableFeatures()
: Collections.<TableFeatures>emptyList();
for (TableFeatures tableFeaturesItem : tableList) {
- TableFeaturesKey tableKey = tableFeaturesItem.getKey();
+ TableFeaturesKey tableKey = tableFeaturesItem.key();
KeyedInstanceIdentifier<TableFeatures, TableFeaturesKey> tableFeaturesII = nodeIdentity
.child(TableFeatures.class, new TableFeaturesKey(tableKey.getTableId()));
provider.getTableFeaturesCommiter().update(tableFeaturesII, tableFeaturesItem, null, nodeIdentity);
: Collections.<Meter>emptyList();
for (Meter meter : meters) {
final KeyedInstanceIdentifier<Meter, MeterKey> meterIdent = nodeIdentity.child(Meter.class,
- meter.getKey());
+ meter.key());
provider.getMeterCommiter().add(meterIdent, meter, nodeIdentity);
}
: Collections.<Table>emptyList();
for (Table table : tables) {
final KeyedInstanceIdentifier<Table, TableKey> tableIdent = nodeIdentity.child(Table.class,
- table.getKey());
+ table.key());
List<Flow> flows = table.getFlow() != null ? table.getFlow() : Collections.<Flow>emptyList();
for (Flow flow : flows) {
final KeyedInstanceIdentifier<Flow, FlowKey> flowIdent = tableIdent.child(Flow.class,
- flow.getKey());
+ flow.key());
provider.getFlowCommiter().add(flowIdent, flow, nodeIdentity);
}
}
* The group to add.
*/
private void addGroup(Map<Long, ListenableFuture<?>> map, Group group) {
- KeyedInstanceIdentifier<Group, GroupKey> groupIdent = nodeIdentity.child(Group.class, group.getKey());
+ KeyedInstanceIdentifier<Group, GroupKey> groupIdent = nodeIdentity.child(Group.class, group.key());
final Long groupId = group.getGroupId().getValue();
ListenableFuture<?> future = JdkFutureAdapters
.listenInPoolThread(provider.getGroupCommiter().add(groupIdent, group, nodeIdentity));
+ ", id=" + groupId;
LOG.debug(msg, cause);
}
- });
+ }, MoreExecutors.directExecutor());
map.put(groupId, future);
}
: Collections.<Table>emptyList();
for (Table table : tables) {
final KeyedInstanceIdentifier<Table, TableKey> tableIdent = nodeIdent.child(Table.class,
- table.getKey());
+ table.key());
List<StaleFlow> staleFlows = table.getStaleFlow() != null ? table.getStaleFlow()
: Collections.<StaleFlow>emptyList();
for (StaleFlow staleFlow : staleFlows) {
Flow toBeDeletedFlow = flowBuilder.setId(staleFlow.getId()).build();
final KeyedInstanceIdentifier<Flow, FlowKey> flowIdent = tableIdent.child(Flow.class,
- toBeDeletedFlow.getKey());
+ toBeDeletedFlow.key());
this.provider.getFlowCommiter().remove(flowIdent, toBeDeletedFlow, nodeIdent);
Group toBeDeletedGroup = groupBuilder.setGroupId(staleGroup.getGroupId()).build();
final KeyedInstanceIdentifier<Group, GroupKey> groupIdent = nodeIdent.child(Group.class,
- toBeDeletedGroup.getKey());
+ toBeDeletedGroup.key());
this.provider.getGroupCommiter().remove(groupIdent, toBeDeletedGroup, nodeIdent);
Meter toBeDeletedMeter = meterBuilder.setMeterId(staleMeter.getMeterId()).build();
final KeyedInstanceIdentifier<Meter, MeterKey> meterIdent = nodeIdent.child(Meter.class,
- toBeDeletedMeter.getKey());
+ toBeDeletedMeter.key());
this.provider.getMeterCommiter().remove(meterIdent, toBeDeletedMeter, nodeIdent);
writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleFlowIId);
}
- CheckedFuture<Void, TransactionCommitFailedException> submitFuture = writeTransaction.submit();
+ ListenableFuture<Void> submitFuture = writeTransaction.submit();
handleStaleEntityDeletionResultFuture(submitFuture);
}
writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleGroupIId);
}
- CheckedFuture<Void, TransactionCommitFailedException> submitFuture = writeTransaction.submit();
+ ListenableFuture<Void> submitFuture = writeTransaction.submit();
handleStaleEntityDeletionResultFuture(submitFuture);
}
writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleMeterIId);
}
- CheckedFuture<Void, TransactionCommitFailedException> submitFuture = writeTransaction.submit();
+ ListenableFuture<Void> submitFuture = writeTransaction.submit();
handleStaleEntityDeletionResultFuture(submitFuture);
}
return nodeIdent.child(StaleMeter.class, new StaleMeterKey(new MeterId(staleMeter.getMeterId())));
}
- private void handleStaleEntityDeletionResultFuture(
- CheckedFuture<Void, TransactionCommitFailedException> submitFuture) {
+ private void handleStaleEntityDeletionResultFuture(ListenableFuture<Void> submitFuture) {
Futures.addCallback(submitFuture, new FutureCallback<Void>() {
@Override
public void onSuccess(Void result) {
public void onFailure(Throwable throwable) {
LOG.debug("Stale entity removal failed {}", throwable);
}
- });
+ }, MoreExecutors.directExecutor());
}
private Flow getDeleteAllFlow() {
Code Review / openflowplugin.git / blobdiff