Code Review / openflowplugin.git / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | review | tree
raw | inline | side by side
connection adapter is set to autoread = false when notofications are not handled
[openflowplugin.git] / openflowplugin-impl / src / main / java / org / opendaylight / openflowplugin / impl / device / DeviceManagerImpl.java
diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceManagerImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceManagerImpl.java
index c075f3259959439e7ddb342224d0bcf77db287b8..3589e71080f93b58ae1a0d9a68891ef33941e138 100644 (file)
--- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceManagerImpl.java
+++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceManagerImpl.java
@@ -26,11 +26,14 @@ import java.util.concurrent.TimeUnit;
 import javax.annotation.CheckForNull;
 import javax.annotation.Nonnull;
 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.NotificationService;
 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
-import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
+import org.opendaylight.openflowplugin.api.ConnectionException;
 import org.opendaylight.openflowplugin.api.OFConstants;
 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
+import org.opendaylight.openflowplugin.api.openflow.connection.ThrottledConnectionsHolder;
 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
@@ -45,6 +48,7 @@ import org.opendaylight.openflowplugin.api.openflow.md.core.TranslatorKey;
 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageIntelligenceAgency;
 import org.opendaylight.openflowplugin.impl.common.MultipartRequestInputFactory;
 import org.opendaylight.openflowplugin.impl.common.NodeStaticReplyTranslatorUtil;
+import org.opendaylight.openflowplugin.impl.connection.ThrottledConnectionsHolderImpl;
 import org.opendaylight.openflowplugin.impl.device.listener.OpenflowProtocolListenerFullImpl;
 import org.opendaylight.openflowplugin.impl.rpc.RequestContextImpl;
 import org.opendaylight.openflowplugin.impl.services.OFJResult2RequestCtxFuture;
@@ -101,17 +105,19 @@ public class DeviceManagerImpl implements DeviceManager, AutoCloseable {
 
     private static final long TICK_DURATION = 500; // 0.5 sec.
     private ScheduledThreadPoolExecutor spyPool;
-    private int spyRate = 10;
+    private final int spyRate = 10;
 
     private final DataBroker dataBroker;
     private final HashedWheelTimer hashedWheelTimer;
     private RequestContextStack emptyRequestContextStack;
     private TranslatorLibrary translatorLibrary;
     private DeviceInitializationPhaseHandler deviceInitPhaseHandler;
-    private NotificationProviderService notificationService;
-    private final List<DeviceContext> synchronizedDeviceContextsList = Collections
-            .<DeviceContext>synchronizedList(new ArrayList<DeviceContext>());
+    private NotificationService notificationService;
+    private NotificationPublishService notificationPublishService;
+
+    private final List<DeviceContext> deviceContexts = new ArrayList<DeviceContext>();
     private final MessageIntelligenceAgency messageIntelligenceAgency = new MessageIntelligenceAgencyImpl();
+    private final ThrottledConnectionsHolder throttledConnectionsHolder;
 
     public DeviceManagerImpl(@Nonnull final DataBroker dataBroker) {
         this.dataBroker = Preconditions.checkNotNull(dataBroker);
@@ -139,6 +145,8 @@ public class DeviceManagerImpl implements DeviceManager, AutoCloseable {
         };
         spyPool = new ScheduledThreadPoolExecutor(1);
         spyPool.scheduleAtFixedRate(messageIntelligenceAgency, spyRate, spyRate, TimeUnit.SECONDS);
+
+        throttledConnectionsHolder = new ThrottledConnectionsHolderImpl(hashedWheelTimer);
     }
 
     @Override
@@ -149,7 +157,7 @@ public class DeviceManagerImpl implements DeviceManager, AutoCloseable {
     @Override
     public void onDeviceContextLevelUp(final DeviceContext deviceContext) {
         // final phase - we have to add new Device to MD-SAL DataStore
-        Preconditions.checkNotNull(deviceContext != null);
+        Preconditions.checkNotNull(deviceContext);
         ((DeviceContextImpl) deviceContext).submitTransaction();
         new BarrierTaskBuilder(deviceContext).buildAndFireBarrierTask();
     }
@@ -160,9 +168,10 @@ public class DeviceManagerImpl implements DeviceManager, AutoCloseable {
 
         final DeviceState deviceState = new DeviceStateImpl(connectionContext.getFeatures(), connectionContext.getNodeId());
 
-        final DeviceContext deviceContext = new DeviceContextImpl(connectionContext, deviceState, dataBroker, hashedWheelTimer, messageIntelligenceAgency);
+        final DeviceContext deviceContext = new DeviceContextImpl(connectionContext, deviceState, dataBroker, hashedWheelTimer, messageIntelligenceAgency, throttledConnectionsHolder);
 
         deviceContext.setNotificationService(notificationService);
+        deviceContext.setNotificationPublishService(notificationPublishService);
         deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, deviceState.getNodeInstanceIdentifier(), new NodeBuilder().setId(deviceState.getNodeId()).build());
 
         connectionContext.setDeviceDisconnectedHandler(deviceContext);
@@ -174,16 +183,17 @@ public class DeviceManagerImpl implements DeviceManager, AutoCloseable {
 
         deviceContext.attachOpenflowMessageListener(messageListener);
 
-        ListenableFuture<List<RpcResult<List<MultipartReply>>>> deviceFeaturesFuture = null;
+        final ListenableFuture<List<RpcResult<List<MultipartReply>>>> deviceFeaturesFuture;
 
-        if (connectionContext.getFeatures().getVersion() == OFConstants.OFP_VERSION_1_0) {
+        final Short version = connectionContext.getFeatures().getVersion();
+        if (OFConstants.OFP_VERSION_1_0 == version) {
             final CapabilitiesV10 capabilitiesV10 = connectionContext.getFeatures().getCapabilitiesV10();
 
-            createEmptyFlowCapableNodeInDs(deviceContext);
-            makeEmptyTables(deviceContext, deviceContext.getDeviceState().getNodeInstanceIdentifier(), connectionContext.getFeatures().getTables());
             DeviceStateUtil.setDeviceStateBasedOnV10Capabilities(deviceState, capabilitiesV10);
 
             deviceFeaturesFuture = createDeviceFeaturesForOF10(messageListener, deviceContext, deviceState);
+            // create empty tables after device description is processed
+            chainTableTrunkWriteOF10(deviceContext, deviceFeaturesFuture);
 
             for (final PortGrouping port : connectionContext.getFeatures().getPhyPort()) {
                 final short ofVersion = deviceContext.getDeviceState().getVersion();
@@ -200,15 +210,18 @@ public class DeviceManagerImpl implements DeviceManager, AutoCloseable {
                 final InstanceIdentifier<NodeConnector> connectorII = deviceState.getNodeInstanceIdentifier().child(NodeConnector.class, connector.getKey());
                 deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, connectorII, connector);
             }
-        } else if (connectionContext.getFeatures().getVersion() == OFConstants.OFP_VERSION_1_3) {
+        } else if (OFConstants.OFP_VERSION_1_3 == version) {
             final Capabilities capabilities = connectionContext.getFeatures().getCapabilities();
             DeviceStateUtil.setDeviceStateBasedOnV13Capabilities(deviceState, capabilities);
             deviceFeaturesFuture = createDeviceFeaturesForOF13(messageListener, deviceContext, deviceState);
+        } else {
+            deviceFeaturesFuture = Futures.immediateFailedFuture(new ConnectionException("Unsupported version " + version));
         }
 
         Futures.addCallback(deviceFeaturesFuture, new FutureCallback<List<RpcResult<List<MultipartReply>>>>() {
             @Override
             public void onSuccess(final List<RpcResult<List<MultipartReply>>> result) {
+                deviceContext.getDeviceState().setValid(true);
                 deviceInitPhaseHandler.onDeviceContextLevelUp(deviceContext);
             }
 
@@ -221,22 +234,44 @@ public class DeviceManagerImpl implements DeviceManager, AutoCloseable {
         });
     }
 
+    private void chainTableTrunkWriteOF10(final DeviceContext deviceContext, final ListenableFuture<List<RpcResult<List<MultipartReply>>>> deviceFeaturesFuture) {
+        Futures.addCallback(deviceFeaturesFuture, new FutureCallback<List<RpcResult<List<MultipartReply>>>>() {
+            @Override
+            public void onSuccess(final List<RpcResult<List<MultipartReply>>> results) {
+                boolean allSucceeded = true;
+                for (final RpcResult<List<MultipartReply>> rpcResult : results) {
+                    allSucceeded &= rpcResult.isSuccessful();
+                }
+                if (allSucceeded) {
+                    createEmptyFlowCapableNodeInDs(deviceContext);
+                    makeEmptyTables(deviceContext, deviceContext.getDeviceState().getNodeInstanceIdentifier(),
+                            deviceContext.getDeviceState().getFeatures().getTables());
+                }
+            }
+
+            @Override
+            public void onFailure(final Throwable t) {
+                //NOOP
+            }
+        });
+    }
+
 
-    private ListenableFuture<RpcResult<List<MultipartReply>>> processReplyDesc(OpenflowProtocolListenerFullImpl messageListener,
-                                                                               DeviceContext deviceContext,
-                                                                               DeviceState deviceState) {
+    private ListenableFuture<RpcResult<List<MultipartReply>>> processReplyDesc(final OpenflowProtocolListenerFullImpl messageListener,
+                                                                               final DeviceContext deviceContext,
+                                                                               final DeviceState deviceState) {
         final ListenableFuture<RpcResult<List<MultipartReply>>> replyDesc = getNodeStaticInfo(messageListener,
                 MultipartType.OFPMPDESC, deviceContext, deviceState.getNodeInstanceIdentifier(), deviceState.getVersion());
         return replyDesc;
     }
 
-    private ListenableFuture<List<RpcResult<List<MultipartReply>>>> createDeviceFeaturesForOF10(OpenflowProtocolListenerFullImpl messageListener,
-                                                                                                DeviceContext deviceContext,
-                                                                                                DeviceState deviceState) {
+    private ListenableFuture<List<RpcResult<List<MultipartReply>>>> createDeviceFeaturesForOF10(final OpenflowProtocolListenerFullImpl messageListener,
+                                                                                                final DeviceContext deviceContext,
+                                                                                                final DeviceState deviceState) {
         return Futures.allAsList(Arrays.asList(processReplyDesc(messageListener, deviceContext, deviceState)));
     }
 
-    private ListenableFuture<List<RpcResult<List<MultipartReply>>>> createDeviceFeaturesForOF13(OpenflowProtocolListenerFullImpl messageListener,
+    private ListenableFuture<List<RpcResult<List<MultipartReply>>>> createDeviceFeaturesForOF13(final OpenflowProtocolListenerFullImpl messageListener,
                                                                                                 final DeviceContext deviceContext,
                                                                                                 final DeviceState deviceState) {
         final ListenableFuture<RpcResult<List<MultipartReply>>> replyDesc = processReplyDesc(messageListener, deviceContext, deviceState);
@@ -403,25 +438,30 @@ public class DeviceManagerImpl implements DeviceManager, AutoCloseable {
     }
 
     @Override
-    public void setNotificationService(final NotificationProviderService notificationServiceParam) {
+    public void setNotificationService(final NotificationService notificationServiceParam) {
         notificationService = notificationServiceParam;
     }
 
+    @Override
+    public void setNotificationPublishService(final NotificationPublishService notificationService) {
+        this.notificationPublishService = notificationService;
+    }
+
     @Override
     public void close() throws Exception {
-        for (DeviceContext deviceContext : synchronizedDeviceContextsList) {
+        for (final DeviceContext deviceContext : deviceContexts) {
             deviceContext.close();
         }
     }
 
     private static void createEmptyFlowCapableNodeInDs(final DeviceContext deviceContext) {
-        FlowCapableNodeBuilder flowCapableNodeBuilder = new FlowCapableNodeBuilder();
+        final FlowCapableNodeBuilder flowCapableNodeBuilder = new FlowCapableNodeBuilder();
         final InstanceIdentifier<FlowCapableNode> fNodeII = deviceContext.getDeviceState().getNodeInstanceIdentifier().augmentation(FlowCapableNode.class);
         deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, fNodeII, flowCapableNodeBuilder.build());
     }
 
     @Override
     public void onDeviceContextClosed(final DeviceContext deviceContext) {
-        synchronizedDeviceContextsList.remove(deviceContext);
+        deviceContexts.remove(deviceContext);
     }
 }
OpenFlow Plugin