*/
package org.opendaylight.openflowplugin.applications.frm;
-import java.util.concurrent.Future;
+import com.google.common.util.concurrent.ListenableFuture;
import org.opendaylight.mdsal.binding.api.ClusteredDataTreeChangeListener;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yangtools.yang.binding.DataObject;
* @return A future associated with RPC task. {@code null} is set to the future
* if this method does not invoke RPC.
*/
- Future<? extends RpcResult<?>> add(InstanceIdentifier<D> identifier, D add,
+ ListenableFuture<? extends RpcResult<?>> add(InstanceIdentifier<D> identifier, D add,
InstanceIdentifier<FlowCapableNode> nodeIdent);
/**
*/
void createStaleMarkEntity(InstanceIdentifier<D> identifier, D del, InstanceIdentifier<FlowCapableNode> nodeIdent);
- Future<? extends RpcResult<?>> removeWithResult(InstanceIdentifier<D> identifier, D del,
+ ListenableFuture<? extends RpcResult<?>> removeWithResult(InstanceIdentifier<D> identifier, D del,
InstanceIdentifier<FlowCapableNode> nodeIdent);
}
import static org.opendaylight.openflowplugin.applications.frm.util.FrmUtil.isFlowDependentOnGroup;
import static org.opendaylight.openflowplugin.applications.frm.util.FrmUtil.isGroupExistsOnDevice;
-import com.google.common.util.concurrent.FluentFuture;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import java.util.Optional;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
import org.opendaylight.infrautils.utils.concurrent.LoggingFutures;
import org.opendaylight.mdsal.binding.api.DataBroker;
import org.opendaylight.mdsal.binding.api.ReadTransaction;
import org.opendaylight.mdsal.binding.api.WriteTransaction;
+import org.opendaylight.mdsal.common.api.CommitInfo;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesManager;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Uri;
}
@Override
- public Future<? extends RpcResult<?>> add(final InstanceIdentifier<Flow> identifier, final Flow addDataObj,
- final InstanceIdentifier<FlowCapableNode> nodeIdent) {
-
- final TableKey tableKey = identifier.firstKeyOf(Table.class);
- if (tableIdValidationPrecondition(tableKey, addDataObj)) {
- BundleId bundleId = getActiveBundle(nodeIdent, provider);
- if (bundleId != null) {
- return provider.getBundleFlowListener().add(identifier, addDataObj, nodeIdent, bundleId);
- } else {
- final String nodeId = getNodeIdValueFromNodeIdentifier(nodeIdent);
- nodeConfigurator.enqueueJob(nodeId, () -> {
- final AddFlowInputBuilder builder = new AddFlowInputBuilder(addDataObj);
-
- builder.setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)));
- builder.setFlowRef(new FlowRef(identifier));
- builder.setFlowTable(new FlowTableRef(nodeIdent.child(Table.class, tableKey)));
- builder.setTransactionUri(new Uri(provider.getNewTransactionId()));
- Uint32 groupId = isFlowDependentOnGroup(addDataObj);
- if (groupId != null) {
- LOG.trace("The flow {} is dependent on group {}. Checking if the group is already present",
- getFlowId(new FlowRef(identifier)), groupId);
- if (isGroupExistsOnDevice(nodeIdent, groupId, provider)) {
- LOG.trace("The dependent group {} is already programmed. Adding the flow {}", groupId,
- getFlowId(new FlowRef(identifier)));
- return provider.getSalFlowService().addFlow(builder.build());
- } else {
- LOG.trace("The dependent group {} isn't programmed yet. Pushing the group", groupId);
- ListenableFuture<RpcResult<AddGroupOutput>> groupFuture = pushDependentGroup(nodeIdent,
- groupId);
- SettableFuture<RpcResult<AddFlowOutput>> resultFuture = SettableFuture.create();
- Futures.addCallback(groupFuture, new AddFlowCallBack(builder.build(), nodeId, groupId,
- resultFuture), MoreExecutors.directExecutor());
- return resultFuture;
- }
- }
+ public ListenableFuture<? extends RpcResult<?>> add(final InstanceIdentifier<Flow> identifier,
+ final Flow addDataObj, final InstanceIdentifier<FlowCapableNode> nodeIdent) {
+ final var tableKey = identifier.firstKeyOf(Table.class);
+ if (!tableIdValidationPrecondition(tableKey, addDataObj)) {
+ return Futures.immediateFuture(null);
+ }
+ final var bundleId = getActiveBundle(nodeIdent, provider);
+ if (bundleId != null) {
+ return provider.getBundleFlowListener().add(identifier, addDataObj, nodeIdent, bundleId);
+ }
- LOG.trace("The flow {} is not dependent on any group. Adding the flow",
- getFlowId(new FlowRef(identifier)));
+ final String nodeId = getNodeIdValueFromNodeIdentifier(nodeIdent);
+ return nodeConfigurator.enqueueJob(nodeId, () -> {
+ final var builder = new AddFlowInputBuilder(addDataObj)
+ .setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)))
+ .setFlowRef(new FlowRef(identifier))
+ .setFlowTable(new FlowTableRef(nodeIdent.child(Table.class, tableKey)))
+ .setTransactionUri(new Uri(provider.getNewTransactionId()));
+ final var groupId = isFlowDependentOnGroup(addDataObj);
+ if (groupId != null) {
+ LOG.trace("The flow {} is dependent on group {}. Checking if the group is already present",
+ getFlowId(new FlowRef(identifier)), groupId);
+ if (isGroupExistsOnDevice(nodeIdent, groupId, provider)) {
+ LOG.trace("The dependent group {} is already programmed. Adding the flow {}", groupId,
+ getFlowId(new FlowRef(identifier)));
return provider.getSalFlowService().addFlow(builder.build());
- });
+ }
+
+ LOG.trace("The dependent group {} isn't programmed yet. Pushing the group", groupId);
+ final var groupFuture = pushDependentGroup(nodeIdent, groupId);
+ final var resultFuture = SettableFuture.<RpcResult<AddFlowOutput>>create();
+ Futures.addCallback(groupFuture, new AddFlowCallBack(builder.build(), nodeId, groupId,
+ resultFuture), MoreExecutors.directExecutor());
+ return resultFuture;
}
- }
- return Futures.immediateFuture(null);
+
+ LOG.trace("The flow {} is not dependent on any group. Adding the flow",
+ getFlowId(new FlowRef(identifier)));
+ return provider.getSalFlowService().addFlow(builder.build());
+ });
}
@Override
writeTransaction.put(LogicalDatastoreType.CONFIGURATION, getStaleFlowInstanceIdentifier(staleFlow, nodeIdent),
staleFlow);
- FluentFuture<?> submitFuture = writeTransaction.commit();
- handleStaleFlowResultFuture(submitFuture);
- }
-
- private static void handleStaleFlowResultFuture(final FluentFuture<?> submitFuture) {
- submitFuture.addCallback(new FutureCallback<Object>() {
+ writeTransaction.commit().addCallback(new FutureCallback<CommitInfo>() {
@Override
- public void onSuccess(final Object result) {
+ public void onSuccess(final CommitInfo result) {
LOG.debug("Stale Flow creation success");
}
LOG.error("Stale Flow creation failed", throwable);
}
}, MoreExecutors.directExecutor());
-
}
private static InstanceIdentifier<org.opendaylight.yang.gen.v1.urn.opendaylight
import com.google.common.util.concurrent.FluentFuture;
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 com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.concurrent.atomic.AtomicLong;
import org.opendaylight.mdsal.binding.api.DataBroker;
import org.opendaylight.mdsal.binding.api.ReadTransaction;
-import org.opendaylight.mdsal.binding.api.WriteTransaction;
+import org.opendaylight.mdsal.common.api.CommitInfo;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
import org.opendaylight.openflowplugin.api.OFConstants;
import org.opendaylight.openflowplugin.api.openflow.FlowGroupCacheManager;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
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;
connectedNode);
reconciliationPreProcess(connectedNode);
}
- if (provider.isBundleBasedReconciliationEnabled()) {
- BundleBasedReconciliationTask bundleBasedReconTask = new BundleBasedReconciliationTask(connectedNode);
- return JdkFutureAdapters.listenInPoolThread(executor.submit(bundleBasedReconTask));
- } else {
- ReconciliationTask reconciliationTask = new ReconciliationTask(connectedNode);
- return JdkFutureAdapters.listenInPoolThread(executor.submit(reconciliationTask));
- }
+
+ return Futures.submit(provider.isBundleBasedReconciliationEnabled()
+ ? new BundleBasedReconciliationTask(connectedNode) : new ReconciliationTask(connectedNode), executor);
}
@Override
public void flowNodeDisconnected(final InstanceIdentifier<FlowCapableNode> disconnectedNode) {
- String node = disconnectedNode.firstKeyOf(Node.class).getId().getValue();
- BigInteger dpnId = getDpnIdFromNodeName(node);
+ final var node = disconnectedNode.firstKeyOf(Node.class).getId().getValue();
+ // FIXME: BigInteger.toString() called here
+ final var dpnId = getDpnIdFromNodeName(node);
reconciliationStates.remove(dpnId.toString());
}
}
if (flowNode.isPresent()) {
- ReconciliationState reconciliationState = new ReconciliationState(
- STARTED, LocalDateTime.now());
+ final var reconciliationState = new ReconciliationState(STARTED, LocalDateTime.now());
//put the dpn info into the map
reconciliationStates.put(dpnId.toString(), reconciliationState);
LOG.debug("FlowNode present for Datapath ID {}", dpnId);
OF_EVENT_LOG.debug("Bundle Reconciliation Start, Node: {}", dpnId);
- final NodeRef nodeRef = new NodeRef(nodeIdentity.firstIdentifierOf(Node.class));
+ final var nodeRef = new NodeRef(nodeIdentity.firstIdentifierOf(Node.class));
- final ControlBundleInput closeBundleInput = new ControlBundleInputBuilder().setNode(nodeRef)
+ final var closeBundleInput = new ControlBundleInputBuilder().setNode(nodeRef)
.setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
.setType(BundleControlType.ONFBCTCLOSEREQUEST).build();
- final ControlBundleInput openBundleInput = new ControlBundleInputBuilder().setNode(nodeRef)
+ final var openBundleInput = new ControlBundleInputBuilder().setNode(nodeRef)
.setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
.setType(BundleControlType.ONFBCTOPENREQUEST).build();
- final ControlBundleInput commitBundleInput = new ControlBundleInputBuilder().setNode(nodeRef)
+ final var commitBundleInput = new ControlBundleInputBuilder().setNode(nodeRef)
.setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
.setType(BundleControlType.ONFBCTCOMMITREQUEST).build();
- final AddBundleMessagesInput deleteAllFlowGroupsInput = new AddBundleMessagesInputBuilder()
+ final var deleteAllFlowGroupsInput = new AddBundleMessagesInputBuilder()
.setNode(nodeRef).setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
.setMessages(createMessages(nodeRef)).build();
LOG.debug("Closing openflow bundle for device {}", dpnId);
/* Close previously opened bundle on the openflow switch if any */
- ListenableFuture<RpcResult<ControlBundleOutput>> closeBundle
- = salBundleService.controlBundle(closeBundleInput);
+ final var closeBundle = salBundleService.controlBundle(closeBundleInput);
/* Open a new bundle on the switch */
- ListenableFuture<RpcResult<ControlBundleOutput>> openBundle
- = Futures.transformAsync(closeBundle, rpcResult -> {
- if (rpcResult.isSuccessful()) {
- LOG.debug("Existing bundle is successfully closed for device {}", dpnId);
- }
- return salBundleService.controlBundle(openBundleInput);
- }, service);
-
- /* Push groups and flows via bundle add messages */
- ListenableFuture<RpcResult<AddBundleMessagesOutput>> deleteAllFlowGroupsFuture
- = Futures.transformAsync(openBundle, rpcResult -> {
- if (rpcResult.isSuccessful()) {
- LOG.debug("Open bundle is successful for device {}", dpnId);
- return salBundleService.addBundleMessages(deleteAllFlowGroupsInput);
- }
- return Futures.immediateFuture(null);
- }, service);
+ final var openBundle = Futures.transformAsync(closeBundle, rpcResult -> {
+ if (rpcResult.isSuccessful()) {
+ LOG.debug("Existing bundle is successfully closed for device {}", dpnId);
+ }
+ return salBundleService.controlBundle(openBundleInput);
+ }, service);
+
+ /* Push groups and flows via bundle add messages */
+ final var deleteAllFlowGroupsFuture = Futures.transformAsync(openBundle, rpcResult -> {
+ if (rpcResult.isSuccessful()) {
+ LOG.debug("Open bundle is successful for device {}", dpnId);
+ return salBundleService.addBundleMessages(deleteAllFlowGroupsInput);
+ }
+ return Futures.immediateFuture(null);
+ }, service);
/* Push flows and groups via bundle add messages */
- Optional<FlowCapableNode> finalFlowNode = flowNode;
- ListenableFuture<List<RpcResult<AddBundleMessagesOutput>>> addbundlesFuture =
- Futures.transformAsync(deleteAllFlowGroupsFuture, rpcResult -> {
- if (rpcResult.isSuccessful()) {
- LOG.debug("Adding delete all flow/group message is successful for device {}", dpnId);
- return Futures.allAsList(addBundleMessages(finalFlowNode.orElseThrow(), bundleIdValue,
- nodeIdentity));
- }
- return Futures.immediateFuture(null);
- }, service);
+ final var finalFlowNode = flowNode;
+ final var addbundlesFuture = Futures.transformAsync(deleteAllFlowGroupsFuture, rpcResult -> {
+ if (rpcResult.isSuccessful()) {
+ LOG.debug("Adding delete all flow/group message is successful for device {}", dpnId);
+ return Futures.allAsList(addBundleMessages(finalFlowNode.orElseThrow(), bundleIdValue,
+ nodeIdentity));
+ }
+ return Futures.immediateFuture(null);
+ }, service);
- /* Commit the bundle on the openflow switch */
- ListenableFuture<RpcResult<ControlBundleOutput>> commitBundleFuture =
- Futures.transformAsync(addbundlesFuture, rpcResult -> {
- LOG.debug("Adding bundle messages completed for device {}", dpnId);
- return salBundleService.controlBundle(commitBundleInput);
- }, service);
+ /* Commit the bundle on the openflow switch */
+ final var commitBundleFuture = Futures.transformAsync(addbundlesFuture, rpcResult -> {
+ LOG.debug("Adding bundle messages completed for device {}", dpnId);
+ return salBundleService.controlBundle(commitBundleInput);
+ }, service);
/* Bundles not supported for meters */
- Collection<Meter> meters = finalFlowNode.orElseThrow().nonnullMeter().values();
+ final var meters = finalFlowNode.orElseThrow().nonnullMeter().values();
Futures.transformAsync(commitBundleFuture,
rpcResult -> {
if (rpcResult.isSuccessful()) {
- for (Meter meter : meters) {
- final KeyedInstanceIdentifier<Meter, MeterKey> meterIdent = nodeIdentity
- .child(Meter.class, meter.key());
- provider.getMeterCommiter().add(meterIdent, meter, nodeIdentity);
+ for (var meter : meters) {
+ provider.getMeterCommiter().add(nodeIdentity.child(Meter.class, meter.key()), meter,
+ nodeIdentity);
}
}
return Futures.immediateFuture(null);
}, service);
+
try {
- RpcResult<ControlBundleOutput> bundleFuture = commitBundleFuture.get();
+ final var bundleFuture = commitBundleFuture.get();
if (bundleFuture != null && bundleFuture.isSuccessful()) {
reconciliationState.setState(COMPLETED, LocalDateTime.now());
LOG.debug("Completing bundle based reconciliation for device ID:{}", dpnId);
private void addGroup(final Map<Uint32, ListenableFuture<?>> map, final Group group) {
KeyedInstanceIdentifier<Group, GroupKey> groupIdent = nodeIdentity.child(Group.class, group.key());
final Uint32 groupId = group.getGroupId().getValue();
- ListenableFuture<?> future = JdkFutureAdapters
- .listenInPoolThread(provider.getGroupCommiter().add(groupIdent, group, nodeIdentity));
+ final var future = provider.getGroupCommiter().add(groupIdent, group, nodeIdentity);
- Futures.addCallback(future, new FutureCallback<Object>() {
+ Futures.addCallback(future, new FutureCallback<RpcResult<?>>() {
@Override
- public void onSuccess(final Object result) {
+ public void onSuccess(final RpcResult<?> result) {
if (LOG.isTraceEnabled()) {
LOG.trace("add-group RPC completed: node={}, id={}",
nodeIdentity.firstKeyOf(Node.class).getId().getValue(), groupId);
}
private void deleteDSStaleFlows(final List<InstanceIdentifier<StaleFlow>> flowsForBulkDelete) {
- WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
-
- for (InstanceIdentifier<StaleFlow> staleFlowIId : flowsForBulkDelete) {
+ final var writeTransaction = dataBroker.newWriteOnlyTransaction();
+ for (var staleFlowIId : flowsForBulkDelete) {
writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleFlowIId);
}
-
- FluentFuture<?> submitFuture = writeTransaction.commit();
- handleStaleEntityDeletionResultFuture(submitFuture);
+ handleStaleEntityDeletionResultFuture(writeTransaction.commit());
}
private void deleteDSStaleGroups(final List<InstanceIdentifier<StaleGroup>> groupsForBulkDelete) {
- WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
-
- for (InstanceIdentifier<StaleGroup> staleGroupIId : groupsForBulkDelete) {
+ final var writeTransaction = dataBroker.newWriteOnlyTransaction();
+ for (var staleGroupIId : groupsForBulkDelete) {
writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleGroupIId);
}
-
- FluentFuture<?> submitFuture = writeTransaction.commit();
- handleStaleEntityDeletionResultFuture(submitFuture);
+ handleStaleEntityDeletionResultFuture(writeTransaction.commit());
}
private void deleteDSStaleMeters(final List<InstanceIdentifier<StaleMeter>> metersForBulkDelete) {
- WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
-
- for (InstanceIdentifier<StaleMeter> staleMeterIId : metersForBulkDelete) {
+ final var writeTransaction = dataBroker.newWriteOnlyTransaction();
+ for (var staleMeterIId : metersForBulkDelete) {
writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleMeterIId);
}
-
- FluentFuture<?> submitFuture = writeTransaction.commit();
- handleStaleEntityDeletionResultFuture(submitFuture);
+ handleStaleEntityDeletionResultFuture(writeTransaction.commit());
}
private static InstanceIdentifier<org.opendaylight.yang.gen.v1.urn.opendaylight
return futureList;
}
- private static void handleStaleEntityDeletionResultFuture(final FluentFuture<?> submitFuture) {
- submitFuture.addCallback(new FutureCallback<Object>() {
+ private static void handleStaleEntityDeletionResultFuture(final FluentFuture<? extends CommitInfo> submitFuture) {
+ submitFuture.addCallback(new FutureCallback<CommitInfo>() {
@Override
- public void onSuccess(final Object result) {
+ public void onSuccess(final CommitInfo result) {
LOG.debug("Stale entity removal success");
}
import static org.opendaylight.openflowplugin.applications.frm.util.FrmUtil.getActiveBundle;
import static org.opendaylight.openflowplugin.applications.frm.util.FrmUtil.getNodeIdValueFromNodeIdentifier;
-import com.google.common.util.concurrent.FluentFuture;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
-import java.util.concurrent.Future;
import org.opendaylight.infrautils.utils.concurrent.LoggingFutures;
import org.opendaylight.mdsal.binding.api.DataBroker;
-import org.opendaylight.mdsal.binding.api.WriteTransaction;
+import org.opendaylight.mdsal.common.api.CommitInfo;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesManager;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Uri;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.AddGroupInput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.AddGroupInputBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.AddGroupOutput;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.RemoveGroupInput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.RemoveGroupInputBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.RemoveGroupOutput;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.UpdateGroupInput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.UpdateGroupInputBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.UpdateGroupOutput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.group.update.OriginalGroupBuilder;
} else {
final String nodeId = getNodeIdValueFromNodeIdentifier(nodeIdent);
nodeConfigurator.enqueueJob(nodeId, () -> {
- final Group group = removeDataObj;
- final RemoveGroupInput removeGroup = new RemoveGroupInputBuilder(group)
+ final var removeGroup = new RemoveGroupInputBuilder(removeDataObj)
.setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)))
.setGroupRef(new GroupRef(identifier))
.setTransactionUri(new Uri(provider.getNewTransactionId()))
.build();
- final ListenableFuture<RpcResult<RemoveGroupOutput>> resultFuture =
- provider.getSalGroupService()
- .removeGroup(removeGroup);
+ final var resultFuture = provider.getSalGroupService() .removeGroup(removeGroup);
Futures.addCallback(resultFuture,
new RemoveGroupCallBack(removeDataObj.getGroupId().getValue(), nodeId),
MoreExecutors.directExecutor());
- LoggingFutures.addErrorLogging(resultFuture, LOG, "removeGroup");
- return resultFuture;
+ return LoggingFutures.addErrorLogging(resultFuture, LOG, "removeGroup");
});
}
}
// TODO: Pull this into ForwardingRulesCommiter and override it here
@Override
- public Future<RpcResult<RemoveGroupOutput>> removeWithResult(final InstanceIdentifier<Group> identifier,
+ public ListenableFuture<RpcResult<RemoveGroupOutput>> removeWithResult(final InstanceIdentifier<Group> identifier,
final Group removeDataObj, final InstanceIdentifier<FlowCapableNode> nodeIdent) {
-
- final Group group = removeDataObj;
- final RemoveGroupInputBuilder builder = new RemoveGroupInputBuilder(group);
-
- builder.setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)));
- builder.setGroupRef(new GroupRef(identifier));
- builder.setTransactionUri(new Uri(provider.getNewTransactionId()));
- return provider.getSalGroupService().removeGroup(builder.build());
+ return provider.getSalGroupService().removeGroup(new RemoveGroupInputBuilder(removeDataObj)
+ .setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)))
+ .setGroupRef(new GroupRef(identifier))
+ .setTransactionUri(new Uri(provider.getNewTransactionId()))
+ .build());
}
@Override
} else {
final String nodeId = getNodeIdValueFromNodeIdentifier(nodeIdent);
nodeConfigurator.enqueueJob(nodeId, () -> {
- final Group originalGroup = original;
- final Group updatedGroup = update;
- final UpdateGroupInputBuilder builder = new UpdateGroupInputBuilder();
- builder.setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)));
- builder.setGroupRef(new GroupRef(identifier));
- builder.setTransactionUri(new Uri(provider.getNewTransactionId()));
- builder.setUpdatedGroup(new UpdatedGroupBuilder(updatedGroup).build());
- builder.setOriginalGroup(new OriginalGroupBuilder(originalGroup).build());
- UpdateGroupInput updateGroupInput = builder.build();
- final ListenableFuture<RpcResult<UpdateGroupOutput>> resultFuture = provider.getSalGroupService()
- .updateGroup(updateGroupInput);
- LoggingFutures.addErrorLogging(resultFuture, LOG, "updateGroup");
+ final var updateGroupInput = new UpdateGroupInputBuilder()
+ .setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)))
+ .setGroupRef(new GroupRef(identifier))
+ .setTransactionUri(new Uri(provider.getNewTransactionId()))
+ .setUpdatedGroup(new UpdatedGroupBuilder(update).build())
+ .setOriginalGroup(new OriginalGroupBuilder(original).build())
+ .build();
+ final var resultFuture = LoggingFutures.addErrorLogging(
+ provider.getSalGroupService().updateGroup(updateGroupInput), LOG, "updateGroup");
Futures.addCallback(resultFuture,
new UpdateGroupCallBack(updateGroupInput.getOriginalGroup().getGroupId().getValue(), nodeId),
MoreExecutors.directExecutor());
}
@Override
- public Future<? extends RpcResult<?>> add(final InstanceIdentifier<Group> identifier, final Group addDataObj,
- final InstanceIdentifier<FlowCapableNode> nodeIdent) {
- BundleId bundleId = getActiveBundle(nodeIdent, provider);
+ public ListenableFuture<? extends RpcResult<?>> add(final InstanceIdentifier<Group> identifier,
+ final Group addDataObj, final InstanceIdentifier<FlowCapableNode> nodeIdent) {
+ final var bundleId = getActiveBundle(nodeIdent, provider);
if (bundleId != null) {
return provider.getBundleGroupListener().add(identifier, addDataObj, nodeIdent, bundleId);
- } else {
- final String nodeId = getNodeIdValueFromNodeIdentifier(nodeIdent);
- return nodeConfigurator
- .enqueueJob(nodeId, () -> {
- final Group group = addDataObj;
- final AddGroupInputBuilder builder = new AddGroupInputBuilder(group);
- builder.setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)));
- builder.setGroupRef(new GroupRef(identifier));
- builder.setTransactionUri(new Uri(provider.getNewTransactionId()));
- AddGroupInput addGroupInput = builder.build();
- final ListenableFuture<RpcResult<AddGroupOutput>> resultFuture;
- resultFuture = provider.getSalGroupService().addGroup(addGroupInput);
- Futures.addCallback(resultFuture,
- new AddGroupCallBack(addGroupInput.getGroupId().getValue(), nodeId),
- MoreExecutors.directExecutor());
- return resultFuture;
- });
}
+
+ final String nodeId = getNodeIdValueFromNodeIdentifier(nodeIdent);
+ return nodeConfigurator.enqueueJob(nodeId, () -> {
+ final var addGroupInput = new AddGroupInputBuilder(addDataObj)
+ .setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)))
+ .setGroupRef(new GroupRef(identifier))
+ .setTransactionUri(new Uri(provider.getNewTransactionId()))
+ .build();
+ final var resultFuture = provider.getSalGroupService().addGroup(addGroupInput);
+ Futures.addCallback(resultFuture,
+ new AddGroupCallBack(addGroupInput.getGroupId().getValue(), nodeId),
+ MoreExecutors.directExecutor());
+ return resultFuture;
+ });
}
@Override
public void createStaleMarkEntity(final InstanceIdentifier<Group> identifier, final Group del,
final InstanceIdentifier<FlowCapableNode> nodeIdent) {
LOG.debug("Creating Stale-Mark entry for the switch {} for Group {} ", nodeIdent, del);
- StaleGroup staleGroup = makeStaleGroup(identifier, del, nodeIdent);
- persistStaleGroup(staleGroup, nodeIdent);
-
- }
-
- private static StaleGroup makeStaleGroup(final InstanceIdentifier<Group> identifier, final Group del,
- final InstanceIdentifier<FlowCapableNode> nodeIdent) {
- StaleGroupBuilder staleGroupBuilder = new StaleGroupBuilder(del);
- return staleGroupBuilder.setGroupId(del.getGroupId()).build();
- }
-
- private void persistStaleGroup(final StaleGroup staleGroup, final InstanceIdentifier<FlowCapableNode> nodeIdent) {
- WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
+ final var staleGroup = new StaleGroupBuilder(del).setGroupId(del.getGroupId()).build();
+ final var writeTransaction = dataBroker.newWriteOnlyTransaction();
writeTransaction.put(LogicalDatastoreType.CONFIGURATION, getStaleGroupInstanceIdentifier(staleGroup, nodeIdent),
staleGroup);
-
- FluentFuture<?> submitFuture = writeTransaction.commit();
- handleStaleGroupResultFuture(submitFuture);
- }
-
- private static void handleStaleGroupResultFuture(final FluentFuture<?> submitFuture) {
- submitFuture.addCallback(new FutureCallback<Object>() {
+ writeTransaction.commit().addCallback(new FutureCallback<CommitInfo>() {
@Override
- public void onSuccess(final Object result) {
+ public void onSuccess(final CommitInfo result) {
LOG.debug("Stale Group creation success");
}
*/
package org.opendaylight.openflowplugin.applications.frm.impl;
-import com.google.common.util.concurrent.FluentFuture;
import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
-import java.util.concurrent.Future;
import org.opendaylight.infrautils.utils.concurrent.LoggingFutures;
import org.opendaylight.mdsal.binding.api.DataBroker;
-import org.opendaylight.mdsal.binding.api.WriteTransaction;
+import org.opendaylight.mdsal.common.api.CommitInfo;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesManager;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Uri;
@Override
public void remove(final InstanceIdentifier<Meter> identifier, final Meter removeDataObj,
final InstanceIdentifier<FlowCapableNode> nodeIdent) {
-
- final RemoveMeterInputBuilder builder = new RemoveMeterInputBuilder(removeDataObj);
-
- builder.setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)));
- builder.setMeterRef(new MeterRef(identifier));
- builder.setTransactionUri(new Uri(provider.getNewTransactionId()));
-
- LoggingFutures.addErrorLogging(provider.getSalMeterService().removeMeter(builder.build()), LOG,
- "removeMeter");
+ LoggingFutures.addErrorLogging(provider.getSalMeterService()
+ .removeMeter(new RemoveMeterInputBuilder(removeDataObj)
+ .setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)))
+ .setMeterRef(new MeterRef(identifier))
+ .setTransactionUri(new Uri(provider.getNewTransactionId()))
+ .build()),
+ LOG, "removeMeter");
}
@Override
- public Future<RpcResult<RemoveMeterOutput>> removeWithResult(final InstanceIdentifier<Meter> identifier,
+ public ListenableFuture<RpcResult<RemoveMeterOutput>> removeWithResult(final InstanceIdentifier<Meter> identifier,
final Meter removeDataObj, final InstanceIdentifier<FlowCapableNode> nodeIdent) {
-
- final RemoveMeterInputBuilder builder = new RemoveMeterInputBuilder(removeDataObj);
-
- builder.setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)));
- builder.setMeterRef(new MeterRef(identifier));
- builder.setTransactionUri(new Uri(provider.getNewTransactionId()));
- return provider.getSalMeterService().removeMeter(builder.build());
+ return provider.getSalMeterService().removeMeter(new RemoveMeterInputBuilder(removeDataObj)
+ .setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)))
+ .setMeterRef(new MeterRef(identifier))
+ .setTransactionUri(new Uri(provider.getNewTransactionId()))
+ .build());
}
@Override
public void update(final InstanceIdentifier<Meter> identifier, final Meter original, final Meter update,
final InstanceIdentifier<FlowCapableNode> nodeIdent) {
-
- final UpdateMeterInputBuilder builder = new UpdateMeterInputBuilder();
-
- builder.setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)));
- builder.setMeterRef(new MeterRef(identifier));
- builder.setTransactionUri(new Uri(provider.getNewTransactionId()));
- builder.setUpdatedMeter(new UpdatedMeterBuilder(update).build());
- builder.setOriginalMeter(new OriginalMeterBuilder(original).build());
-
- LoggingFutures.addErrorLogging(provider.getSalMeterService().updateMeter(builder.build()), LOG,
- "updateMeter");
+ LoggingFutures.addErrorLogging(provider.getSalMeterService().updateMeter(new UpdateMeterInputBuilder()
+ .setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)))
+ .setMeterRef(new MeterRef(identifier))
+ .setTransactionUri(new Uri(provider.getNewTransactionId()))
+ .setUpdatedMeter(new UpdatedMeterBuilder(update).build())
+ .setOriginalMeter(new OriginalMeterBuilder(original).build())
+ .build()), LOG, "updateMeter");
}
@Override
- public Future<RpcResult<AddMeterOutput>> add(final InstanceIdentifier<Meter> identifier, final Meter addDataObj,
- final InstanceIdentifier<FlowCapableNode> nodeIdent) {
-
- final AddMeterInputBuilder builder = new AddMeterInputBuilder(addDataObj);
-
- builder.setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)));
- builder.setMeterRef(new MeterRef(identifier));
- builder.setTransactionUri(new Uri(provider.getNewTransactionId()));
- return provider.getSalMeterService().addMeter(builder.build());
+ public ListenableFuture<RpcResult<AddMeterOutput>> add(final InstanceIdentifier<Meter> identifier,
+ final Meter addDataObj, final InstanceIdentifier<FlowCapableNode> nodeIdent) {
+ return provider.getSalMeterService().addMeter(new AddMeterInputBuilder(addDataObj)
+ .setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)))
+ .setMeterRef(new MeterRef(identifier))
+ .setTransactionUri(new Uri(provider.getNewTransactionId()))
+ .build());
}
@Override
public void createStaleMarkEntity(final InstanceIdentifier<Meter> identifier, final Meter del,
final InstanceIdentifier<FlowCapableNode> nodeIdent) {
LOG.debug("Creating Stale-Mark entry for the switch {} for meter {} ", nodeIdent, del);
- StaleMeter staleMeter = makeStaleMeter(del);
- persistStaleMeter(staleMeter, nodeIdent);
- }
-
- private static StaleMeter makeStaleMeter(final Meter del) {
- StaleMeterBuilder staleMeterBuilder = new StaleMeterBuilder(del);
- return staleMeterBuilder.setMeterId(del.getMeterId()).build();
- }
-
- private void persistStaleMeter(final StaleMeter staleMeter, final InstanceIdentifier<FlowCapableNode> nodeIdent) {
- WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
- writeTransaction.put(LogicalDatastoreType.CONFIGURATION, getStaleMeterInstanceIdentifier(staleMeter, nodeIdent),
- staleMeter);
- FluentFuture<?> submitFuture = writeTransaction.commit();
- handleStaleMeterResultFuture(submitFuture);
- }
-
- private static void handleStaleMeterResultFuture(final FluentFuture<?> submitFuture) {
- submitFuture.addCallback(new FutureCallback<Object>() {
+ final var staleMeter = new StaleMeterBuilder(del).setMeterId(del.getMeterId()).build();
+ final var writeTransaction = dataBroker.newWriteOnlyTransaction();
+ writeTransaction.put(LogicalDatastoreType.CONFIGURATION,
+ nodeIdent.child(StaleMeter.class, new StaleMeterKey(new MeterId(staleMeter.getMeterId()))), staleMeter);
+ writeTransaction.commit().addCallback(new FutureCallback<CommitInfo>() {
@Override
- public void onSuccess(final Object result) {
+ public void onSuccess(final CommitInfo result) {
LOG.debug("Stale Meter creation success");
}
}
}, MoreExecutors.directExecutor());
}
-
- private static InstanceIdentifier<org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819
- .meters.StaleMeter> getStaleMeterInstanceIdentifier(
- final StaleMeter staleMeter, final InstanceIdentifier<FlowCapableNode> nodeIdent) {
- return nodeIdent.child(StaleMeter.class, new StaleMeterKey(new MeterId(staleMeter.getMeterId())));
- }
}
package org.opendaylight.openflowplugin.applications.frm.impl;
import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
import java.util.Collections;
-import java.util.concurrent.Future;
import org.opendaylight.infrautils.utils.concurrent.LoggingFutures;
import org.opendaylight.mdsal.binding.api.DataBroker;
import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesManager;
}
@Override
- public
Future<? extends RpcResult<?>> add(final InstanceIdentifier<TableFeatures> identifier,
+ public
ListenableFuture<RpcResult<?>> add(final InstanceIdentifier<TableFeatures> identifier,
final TableFeatures addDataObj, final InstanceIdentifier<FlowCapableNode> nodeIdent) {
return Futures.immediateFuture(null);
}
}
@Override
- public
Future<? extends RpcResult<?>> removeWithResult(final InstanceIdentifier<TableFeatures> identifier,
+ public
ListenableFuture<RpcResult<?>> removeWithResult(final InstanceIdentifier<TableFeatures> identifier,
final TableFeatures del, final InstanceIdentifier<FlowCapableNode> nodeIdent) {
- return null;
+ return Futures.immediateFuture(null);
}
}
// // at
// // org.opendaylight.openflowjava.protocol.impl.serialization.match.AbstractOxmMatchEntrySerializer
// // .serializeHeader(AbstractOxmMatchEntrySerializer.java:31
-// // allResults.add(JdkFutureAdapters.listenInPoolThread(
-// // tableForwarder.update(tableFeaturesII, null, tableFeaturesItem, nodeIdent)));
+// // allResults.add(
+// // tableForwarder.update(tableFeaturesII, null, tableFeaturesItem, nodeIdent));
// }
// }
// }
Code Review / openflowplugin.git / commitdiff