Merge changes from topic "mdsal-4.0.3"

[openflowplugin.git] / applications / forwardingrules-manager / src / main / java / org / opendaylight / openflowplugin / applications / frm / impl / FlowForwarder.java

diff --git a/applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/FlowForwarder.java b/applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/FlowForwarder.java
index 00eda3f7c1f65b576feb3d2f95ab07ab7284dbf5..07744ffc45e17bb2cf2032b6310dfe17d4eeb25d 100644 (file)
--- a/applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/FlowForwarder.java
+++ b/applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/FlowForwarder.java
@@ -1,4 +1,4 @@
-/**
+/*
  * Copyright (c) 2014, 2017 Cisco Systems, Inc. and others.  All rights reserved.
  *
  * This program and the accompanying materials are made available under the
@@ -14,21 +14,22 @@ import static org.opendaylight.openflowplugin.applications.frm.util.FrmUtil.getN
 import static org.opendaylight.openflowplugin.applications.frm.util.FrmUtil.isFlowDependentOnGroup;
 import static org.opendaylight.openflowplugin.applications.frm.util.FrmUtil.isGroupExistsOnDevice;
 
-import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
+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 com.google.common.util.concurrent.SettableFuture;
+import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
-import org.opendaylight.controller.md.sal.binding.api.DataBroker;
-import org.opendaylight.controller.md.sal.binding.api.DataTreeIdentifier;
-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.infrautils.utils.concurrent.JdkFutures;
+import org.opendaylight.infrautils.utils.concurrent.LoggingFutures;
+import org.opendaylight.mdsal.binding.api.DataBroker;
+import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
+import org.opendaylight.mdsal.binding.api.ReadTransaction;
+import org.opendaylight.mdsal.binding.api.WriteTransaction;
+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;
@@ -71,10 +72,10 @@ import org.slf4j.LoggerFactory;
 
 /**
  * FlowForwarder It implements
- * {@link org.opendaylight.controller.md.sal.binding.api.DataTreeChangeListener}
+ * {@link org.opendaylight.mdsal.binding.api.DataTreeChangeListener}
  * for WildCardedPath to {@link Flow} and ForwardingRulesCommiter interface for
  * methods: add, update and remove {@link Flow} processing for
- * {@link org.opendaylight.controller.md.sal.binding.api.DataTreeModification}.
+ * {@link org.opendaylight.mdsal.binding.api.DataTreeModification}.
  */
 public class FlowForwarder extends AbstractListeningCommiter<Flow> {
 
@@ -83,17 +84,15 @@ public class FlowForwarder extends AbstractListeningCommiter<Flow> {
     private static final String GROUP_EXISTS_IN_DEVICE_ERROR = "GROUPEXISTS";
 
     private ListenerRegistration<FlowForwarder> listenerRegistration;
-    private final BundleFlowForwarder bundleFlowForwarder;
 
     public FlowForwarder(final ForwardingRulesManager manager, final DataBroker db) {
         super(manager, db);
-        bundleFlowForwarder = new BundleFlowForwarder(manager);
     }
 
     @Override
     @SuppressWarnings("IllegalCatch")
     public void registerListener() {
-        final DataTreeIdentifier<Flow> treeId = new DataTreeIdentifier<>(LogicalDatastoreType.CONFIGURATION,
+        final DataTreeIdentifier<Flow> treeId = DataTreeIdentifier.create(LogicalDatastoreType.CONFIGURATION,
                 getWildCardPath());
         try {
             listenerRegistration = dataBroker.registerDataTreeChangeListener(treeId, FlowForwarder.this);
@@ -126,7 +125,7 @@ public class FlowForwarder extends AbstractListeningCommiter<Flow> {
         if (tableIdValidationPrecondition(tableKey, removeDataObj)) {
             BundleId bundleId = getActiveBundle(nodeIdent, provider);
             if (bundleId != null) {
-                bundleFlowForwarder.remove(identifier, removeDataObj, nodeIdent, bundleId);
+                provider.getBundleFlowListener().remove(identifier, removeDataObj, nodeIdent, bundleId);
             } else {
                 final RemoveFlowInputBuilder builder = new RemoveFlowInputBuilder(removeDataObj);
                 builder.setFlowRef(new FlowRef(identifier));
@@ -138,9 +137,8 @@ public class FlowForwarder extends AbstractListeningCommiter<Flow> {
                 // into remove-flow input so that only a flow entry associated with
                 // a given flow object is removed.
                 builder.setTransactionUri(new Uri(provider.getNewTransactionId())).setStrict(Boolean.TRUE);
-                final Future<RpcResult<RemoveFlowOutput>> resultFuture =
-                        provider.getSalFlowService().removeFlow(builder.build());
-                JdkFutures.addErrorLogging(resultFuture, LOG, "removeFlow");
+                LoggingFutures.addErrorLogging(provider.getSalFlowService().removeFlow(builder.build()), LOG,
+                    "removeFlow");
             }
         }
     }
@@ -148,10 +146,10 @@ public class FlowForwarder extends AbstractListeningCommiter<Flow> {
     // TODO: Pull this into ForwardingRulesCommiter and override it here
 
     @Override
-    public Future<RpcResult<RemoveFlowOutput>> removeWithResult(final InstanceIdentifier<Flow> identifier,
+    public ListenableFuture<RpcResult<RemoveFlowOutput>> removeWithResult(final InstanceIdentifier<Flow> identifier,
             final Flow removeDataObj, final InstanceIdentifier<FlowCapableNode> nodeIdent) {
 
-        Future<RpcResult<RemoveFlowOutput>> resultFuture = SettableFuture.create();
+        ListenableFuture<RpcResult<RemoveFlowOutput>> resultFuture = SettableFuture.create();
         final TableKey tableKey = identifier.firstKeyOf(Table.class);
         if (tableIdValidationPrecondition(tableKey, removeDataObj)) {
             final RemoveFlowInputBuilder builder = new RemoveFlowInputBuilder(removeDataObj);
@@ -178,7 +176,7 @@ public class FlowForwarder extends AbstractListeningCommiter<Flow> {
         if (tableIdValidationPrecondition(tableKey, update)) {
             BundleId bundleId = getActiveBundle(nodeIdent, provider);
             if (bundleId != null) {
-                bundleFlowForwarder.update(identifier, original, update, nodeIdent, bundleId);
+                provider.getBundleFlowListener().update(identifier, original, update, nodeIdent, bundleId);
             } else {
                 final NodeId nodeId = getNodeIdFromNodeIdentifier(nodeIdent);
                 nodeConfigurator.enqueueJob(nodeId.getValue(), () -> {
@@ -230,7 +228,7 @@ public class FlowForwarder extends AbstractListeningCommiter<Flow> {
         if (tableIdValidationPrecondition(tableKey, addDataObj)) {
             BundleId bundleId = getActiveBundle(nodeIdent, provider);
             if (bundleId != null) {
-                return bundleFlowForwarder.add(identifier, addDataObj, nodeIdent, bundleId);
+                return provider.getBundleFlowListener().add(identifier, addDataObj, nodeIdent, bundleId);
             } else {
                 final NodeId nodeId = getNodeIdFromNodeIdentifier(nodeIdent);
                 nodeConfigurator.enqueueJob(nodeId.getValue(), () -> {
@@ -304,20 +302,20 @@ public class FlowForwarder extends AbstractListeningCommiter<Flow> {
         writeTransaction.put(LogicalDatastoreType.CONFIGURATION, getStaleFlowInstanceIdentifier(staleFlow, nodeIdent),
                 staleFlow, false);
 
-        ListenableFuture<Void> submitFuture = writeTransaction.submit();
+        FluentFuture<?> submitFuture = writeTransaction.commit();
         handleStaleFlowResultFuture(submitFuture);
     }
 
-    private void handleStaleFlowResultFuture(ListenableFuture<Void> submitFuture) {
-        Futures.addCallback(submitFuture, new FutureCallback<Void>() {
+    private void handleStaleFlowResultFuture(FluentFuture<?> submitFuture) {
+        submitFuture.addCallback(new FutureCallback<Object>() {
             @Override
-            public void onSuccess(Void result) {
+            public void onSuccess(Object result) {
                 LOG.debug("Stale Flow creation success");
             }
 
             @Override
             public void onFailure(Throwable throwable) {
-                LOG.error("Stale Flow creation failed {}", throwable);
+                LOG.error("Stale Flow creation failed", throwable);
             }
         }, MoreExecutors.directExecutor());
 
@@ -341,7 +339,7 @@ public class FlowForwarder extends AbstractListeningCommiter<Flow> {
         InstanceIdentifier<Group> groupIdent = buildGroupInstanceIdentifier(nodeIdent, groupId);
         ListenableFuture<RpcResult<AddGroupOutput>> resultFuture;
         LOG.info("Reading the group from config inventory: {}", groupId);
-        try (ReadOnlyTransaction readTransaction = provider.getReadTransaction()) {
+        try (ReadTransaction readTransaction = provider.getReadTransaction()) {
             Optional<Group> group = readTransaction.read(LogicalDatastoreType.CONFIGURATION, groupIdent).get();
             if (group.isPresent()) {
                 final AddGroupInputBuilder builder = new AddGroupInputBuilder(group.get());
@@ -410,6 +408,7 @@ public class FlowForwarder extends AbstractListeningCommiter<Flow> {
         public void onFailure(Throwable throwable) {
             LOG.error("Service call for adding flow with id {} failed for node {}",
                     getFlowId(addFlowInput.getFlowRef()), nodeId, throwable);
+            resultFuture.setException(throwable);
         }
     }
 
@@ -459,6 +458,7 @@ public class FlowForwarder extends AbstractListeningCommiter<Flow> {
         public void onFailure(Throwable throwable) {
             LOG.error("Service call for updating flow with id {} failed for node {}",
                     getFlowId(updateFlowInput.getFlowRef()), nodeId, throwable);
+            resultFuture.setException(throwable);
         }
     }
 }