Switch to MD-SAL APIs

[openflowplugin.git] / openflowplugin-impl / src / main / java / org / opendaylight / openflowplugin / impl / device / DeviceContextImpl.java

diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.java
index 4295abc57c68d05619de6399ee3b7998484e647a..c44a2605bb375cb0e898e34c09b81a21a685f7eb 100644 (file)
--- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.java
+++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.java
@@ -1,4 +1,4 @@
-/**
+/*
  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
  *
  * This program and the accompanying materials are made available under the
@@ -16,10 +16,10 @@ import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import io.netty.util.HashedWheelTimer;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -27,10 +27,10 @@ import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
-import org.opendaylight.controller.md.sal.binding.api.DataBroker;
-import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
-import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
-import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.mdsal.binding.api.DataBroker;
+import org.opendaylight.mdsal.binding.api.NotificationPublishService;
+import org.opendaylight.mdsal.binding.api.ReadTransaction;
+import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
 import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
 import org.opendaylight.openflowjava.protocol.api.keys.MessageTypeKey;
@@ -130,7 +130,7 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi
     private final HashedWheelTimer hashedWheelTimer;
     private final DeviceState deviceState;
     private final DataBroker dataBroker;
-    private final Collection<RequestContext<?>> requestContexts = new HashSet<>();
+    private final Collection<RequestContext<?>> requestContexts = ConcurrentHashMap.newKeySet();
     private final MessageSpy messageSpy;
     private final MessageTranslator<PortGrouping, FlowCapableNodeConnector> portStatusTranslator;
     private final MessageTranslator<PacketInMessage, PacketReceived> packetInTranslator;
@@ -218,7 +218,7 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi
     }
 
     @Override
-    public ReadOnlyTransaction getReadTransaction() {
+    public ReadTransaction getReadTransaction() {
         return dataBroker.newReadOnlyTransaction();
     }
 
@@ -258,6 +258,11 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi
         return initialized.get() && transactionChainManager.submitTransaction();
     }
 
+    @Override
+    public boolean syncSubmitTransaction() {
+        return initialized.get() && transactionChainManager.submitTransaction(true);
+    }
+
     @Override
     public ConnectionContext getPrimaryConnectionContext() {
         return primaryConnectionContext;
@@ -323,7 +328,6 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi
         if (initialized.get()) {
             try {
                 writePortStatusMessage(portStatus);
-                submitTransaction();
             } catch (final Exception e) {
                 LOG.warn("Error processing port status message for port {} on device {}",
                         portStatus.getPortNo(), getDeviceInfo(), e);
@@ -346,15 +350,15 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi
                                 OpenflowVersion.get(deviceInfo.getVersion()))));
 
         writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, new NodeConnectorBuilder()
-                .setKey(iiToNodeConnector.getKey())
+                .withKey(iiToNodeConnector.getKey())
                 .addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new
                         FlowCapableNodeConnectorStatisticsDataBuilder().build())
                 .addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector)
                 .build());
-        submitTransaction();
+        syncSubmitTransaction();
         if (PortReason.OFPPRDELETE.equals(portStatusMessage.getReason())) {
             addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
-            submitTransaction();
+            syncSubmitTransaction();
         }
     }
 
@@ -495,8 +499,8 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi
         if (Objects.nonNull(implementedInterface) && org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service
                 .rev130709.PacketInMessage.class.equals(implementedInterface)) {
             final org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709
-                    .PacketInMessage packetInMessage = org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service
-                    .rev130709.PacketInMessage.class.cast(message);
+                    .PacketInMessage packetInMessage = (org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service
+                .rev130709.PacketInMessage) message;
 
             handlePacketInMessage(packetInMessage, implementedInterface, packetInMessage.getMatch());
             return true;
@@ -553,8 +557,8 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi
     }
 
     @Override
-    public ListenableFuture<Void> closeServiceInstance() {
-        final ListenableFuture<Void> listenableFuture = initialized.get()
+    public ListenableFuture<?> closeServiceInstance() {
+        final ListenableFuture<?> listenableFuture = initialized.get()
                 ? transactionChainManager.deactivateTransactionManager()
                 : Futures.immediateFuture(null);
 
@@ -591,11 +595,11 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi
             deviceFlowRegistry.close();
             deviceMeterRegistry.close();
 
-            final ListenableFuture<Void> txChainShuttingDown = transactionChainManager.shuttingDown();
+            final ListenableFuture<?> txChainShuttingDown = transactionChainManager.shuttingDown();
 
-            Futures.addCallback(txChainShuttingDown, new FutureCallback<Void>() {
+            Futures.addCallback(txChainShuttingDown, new FutureCallback<Object>() {
                 @Override
-                public void onSuccess(@Nullable final Void result) {
+                public void onSuccess(final Object result) {
                     transactionChainManager.close();
                     transactionChainManager = null;
                 }
@@ -659,7 +663,7 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi
                     deviceInfo.toString()));
         }
 
-        final ListenableFuture<List<com.google.common.base.Optional<FlowCapableNode>>> deviceFlowRegistryFill =
+        final ListenableFuture<List<Optional<FlowCapableNode>>> deviceFlowRegistryFill =
                 getDeviceFlowRegistry().fill();
         Futures.addCallback(deviceFlowRegistryFill,
                 new DeviceFlowRegistryCallback(deviceFlowRegistryFill, contextChainMastershipWatcher),
@@ -704,20 +708,19 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi
         hasState.set(true);
     }
 
-    private class DeviceFlowRegistryCallback implements FutureCallback<List<com.google.common.base
-            .Optional<FlowCapableNode>>> {
-        private final ListenableFuture<List<com.google.common.base.Optional<FlowCapableNode>>> deviceFlowRegistryFill;
+    private class DeviceFlowRegistryCallback implements FutureCallback<List<Optional<FlowCapableNode>>> {
+        private final ListenableFuture<List<Optional<FlowCapableNode>>> deviceFlowRegistryFill;
         private final ContextChainMastershipWatcher contextChainMastershipWatcher;
 
         DeviceFlowRegistryCallback(
-                ListenableFuture<List<com.google.common.base.Optional<FlowCapableNode>>> deviceFlowRegistryFill,
+                ListenableFuture<List<Optional<FlowCapableNode>>> deviceFlowRegistryFill,
                 ContextChainMastershipWatcher contextChainMastershipWatcher) {
             this.deviceFlowRegistryFill = deviceFlowRegistryFill;
             this.contextChainMastershipWatcher = contextChainMastershipWatcher;
         }
 
         @Override
-        public void onSuccess(@Nullable List<com.google.common.base.Optional<FlowCapableNode>> result) {
+        public void onSuccess(@Nullable List<Optional<FlowCapableNode>> result) {
             if (LOG.isDebugEnabled()) {
                 // Count all flows we read from datastore for debugging purposes.
                 // This number do not always represent how many flows were actually added
@@ -728,7 +731,8 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi
                         .stream()
                         .flatMap(Collection::stream)
                         .filter(Objects::nonNull)
-                        .flatMap(flowCapableNodeOptional -> flowCapableNodeOptional.asSet().stream())
+                        .flatMap(flowCapableNodeOptional
+                            -> com.google.common.base.Optional.fromJavaUtil(flowCapableNodeOptional).asSet().stream())
                         .filter(Objects::nonNull)
                         .filter(flowCapableNode -> Objects.nonNull(flowCapableNode.getTable()))
                         .flatMap(flowCapableNode -> flowCapableNode.getTable().stream())
@@ -751,8 +755,7 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi
                     LOG.debug("Cancelled filling flow registry with flows for node: {}", deviceInfo);
                 }
             } else {
-                LOG.warn("Failed filling flow registry with flows for node: {} with exception: {}", deviceInfo,
-                         throwable);
+                LOG.warn("Failed filling flow registry with flows for node: {}", deviceInfo, throwable);
             }
             contextChainMastershipWatcher.onNotAbleToStartMastership(
                     deviceInfo,