Merge "OPNFLWPLUG-898 : remove deprecated checkedfuture"

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

diff --git a/applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/FlowNodeReconciliationImpl.java b/applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/FlowNodeReconciliationImpl.java
index d408bc4ee35f799ed549e36cbedc2aed203a25c8..417e56e18487bd4cf72b437c4ff3d4f6519a7e4d 100644 (file)
--- a/applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/FlowNodeReconciliationImpl.java
+++ b/applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/FlowNodeReconciliationImpl.java
@@ -157,8 +157,6 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
                     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));
@@ -181,7 +179,7 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
             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();
@@ -193,6 +191,10 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
                 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();
@@ -205,22 +207,30 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
                         .setNode(nodeRef).setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS)
                         .setMessages(createMessages(nodeRef, flowNode)).build();
 
-                ListenableFuture<RpcResult<ControlBundleOutput>> openBundle
-                        = salBundleService.controlBundle(openBundleInput);
+                /* Close previously opened bundle on the openflow switch if any */
+                ListenableFuture<RpcResult<ControlBundleOutput>> closeBundle
+                        = salBundleService.controlBundle(closeBundleInput);
 
-                ListenableFuture<RpcResult<AddBundleMessagesOutput>> addBundleMessagesFuture = Futures
-                        .transformAsync(JdkFutureAdapters.listenInPoolThread(openBundle), rpcResult -> {
+                /* Open a new bundle on the switch */
+                ListenableFuture<RpcResult<ControlBundleOutput>> openBundle =
+                        Futures.transformAsync(closeBundle,
+                            rpcResult -> salBundleService.controlBundle(openBundleInput),
+                            MoreExecutors.directExecutor());
+
+                /* 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);
                         }, MoreExecutors.directExecutor());
+
+                /* Commit the bundle on the openflow switch */
                 ListenableFuture<RpcResult<ControlBundleOutput>> commitBundleFuture
                         = Futures.transformAsync(addBundleMessagesFuture, rpcResult -> {
                             if (rpcResult.isSuccessful()) {
-                                return JdkFutureAdapters
-                                        .listenInPoolThread(salBundleService.controlBundle(commitBundleInput));
+                                return salBundleService.controlBundle(commitBundleInput);
                             }
                             return Futures.immediateFuture(null);
                         }, MoreExecutors.directExecutor());