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

   1 /**
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.openflowplugin.impl.device;
   9
  10 import com.google.common.annotations.VisibleForTesting;
  11 import com.google.common.base.Function;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.base.Verify;
  14 import com.google.common.util.concurrent.AsyncFunction;
  15 import com.google.common.util.concurrent.FutureCallback;
  16 import com.google.common.util.concurrent.Futures;
  17 import com.google.common.util.concurrent.ListenableFuture;
  18 import io.netty.util.Timeout;
  19 import java.math.BigInteger;
  20 import java.util.HashMap;
  21 import java.util.List;
  22 import java.util.Map;
  23 import java.util.concurrent.ConcurrentHashMap;
  24 import java.util.concurrent.atomic.AtomicLong;
  25 import javax.annotation.CheckForNull;
  26 import javax.annotation.Nonnull;
  27 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  28 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  29 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  30 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  31 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  32 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueue;
  33 import org.opendaylight.openflowjava.protocol.api.keys.MessageTypeKey;
  34 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
  35 import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
  36 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  37 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
  38 import org.opendaylight.openflowplugin.api.openflow.device.MessageTranslator;
  39 import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
  40 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
  41 import org.opendaylight.openflowplugin.api.openflow.device.Xid;
  42 import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
  43 import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleConductor;
  44 import org.opendaylight.openflowplugin.api.openflow.md.core.SwitchConnectionDistinguisher;
  45 import org.opendaylight.openflowplugin.api.openflow.md.core.TranslatorKey;
  46 import org.opendaylight.openflowplugin.api.openflow.registry.ItemLifeCycleRegistry;
  47 import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
  48 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor;
  49 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowRegistryKey;
  50 import org.opendaylight.openflowplugin.api.openflow.registry.group.DeviceGroupRegistry;
  51 import org.opendaylight.openflowplugin.api.openflow.registry.meter.DeviceMeterRegistry;
  52 import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleKeeper;
  53 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
  54 import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener;
  55 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
  56 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
  57 import org.opendaylight.openflowplugin.extension.api.ConvertorMessageFromOFJava;
  58 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
  59 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
  60 import org.opendaylight.openflowplugin.extension.api.exception.ConversionException;
  61 import org.opendaylight.openflowplugin.extension.api.path.MessagePath;
  62 import org.opendaylight.openflowplugin.impl.common.ItemLifeCycleSourceImpl;
  63 import org.opendaylight.openflowplugin.impl.common.NodeStaticReplyTranslatorUtil;
  64 import org.opendaylight.openflowplugin.impl.device.listener.MultiMsgCollectorImpl;
  65 import org.opendaylight.openflowplugin.impl.registry.flow.DeviceFlowRegistryImpl;
  66 import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory;
  67 import org.opendaylight.openflowplugin.impl.registry.group.DeviceGroupRegistryImpl;
  68 import org.opendaylight.openflowplugin.impl.registry.meter.DeviceMeterRegistryImpl;
  69 import org.opendaylight.openflowplugin.impl.rpc.RpcContextImpl;
  70 import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtils;
  71 import org.opendaylight.openflowplugin.impl.util.MdSalRegistrationUtils;
  72 import org.opendaylight.openflowplugin.openflow.md.core.session.SwitchConnectionCookieOFImpl;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.experimenter.message.service.rev151020.ExperimenterMessageFromDevBuilder;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorRef;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorBuilder;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.PortReason;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.Error;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterMessage;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemoved;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketIn;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortGrouping;
  96 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage;
  97 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.experimenter.core.ExperimenterDataOfChoice;
  98 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.experimenter.types.rev151020.experimenter.core.message.ExperimenterMessageOfChoice;
  99 import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
 100 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
 101 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
 102 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole;
 103 import org.opendaylight.yangtools.yang.binding.DataObject;
 104 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 105 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
 106 import org.slf4j.Logger;
 107 import org.slf4j.LoggerFactory;
 108
 109 /**
 110  *
 111  */
 112 public class DeviceContextImpl implements DeviceContext, ExtensionConverterProviderKeeper {
 113
 114     private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class);
 115
 116     // TODO: drain factor should be parametrized
 117     private static final float REJECTED_DRAIN_FACTOR = 0.25f;
 118     // TODO: low water mark factor should be parametrized
 119     private static final float LOW_WATERMARK_FACTOR = 0.75f;
 120     // TODO: high water mark factor should be parametrized
 121     private static final float HIGH_WATERMARK_FACTOR = 0.95f;
 122
 123     private final ConnectionContext primaryConnectionContext;
 124     private final DeviceState deviceState;
 125     private final DataBroker dataBroker;
 126     private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
 127     private final TransactionChainManager transactionChainManager;
 128     private final DeviceFlowRegistry deviceFlowRegistry;
 129     private final DeviceGroupRegistry deviceGroupRegistry;
 130     private final DeviceMeterRegistry deviceMeterRegistry;
 131     private final PacketInRateLimiter packetInLimiter;
 132     private final MessageSpy messageSpy;
 133     private final ItemLifeCycleKeeper flowLifeCycleKeeper;
 134     private NotificationPublishService notificationPublishService;
 135     private final OutboundQueue outboundQueueProvider;
 136     private Timeout barrierTaskTimeout;
 137     private final MessageTranslator<PortGrouping, FlowCapableNodeConnector> portStatusTranslator;
 138     private final MessageTranslator<PacketInMessage, PacketReceived> packetInTranslator;
 139     private final MessageTranslator<FlowRemoved, org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved> flowRemovedTranslator;
 140     private final TranslatorLibrary translatorLibrary;
 141     private final Map<Long, NodeConnectorRef> nodeConnectorCache;
 142     private final ItemLifeCycleRegistry itemLifeCycleSourceRegistry;
 143     private RpcContext rpcContext;
 144     private ExtensionConverterProvider extensionConverterProvider;
 145
 146     private final boolean switchFeaturesMandatory;
 147     private StatisticsContext statisticsContext;
 148
 149     private final NodeId nodeId;
 150
 151     private volatile DEVICE_CONTEXT_STATE deviceCtxState;
 152     private boolean isStatisticsRpcEnabled;
 153
 154
 155     @VisibleForTesting
 156     DeviceContextImpl(@Nonnull final ConnectionContext primaryConnectionContext,
 157                       @Nonnull final DeviceState deviceState,
 158                       @Nonnull final DataBroker dataBroker,
 159                       @Nonnull final LifecycleConductor conductor,
 160                       @Nonnull final OutboundQueueProvider outboundQueueProvider,
 161                       @Nonnull final TranslatorLibrary translatorLibrary,
 162                       final boolean switchFeaturesMandatory) {
 163         this.switchFeaturesMandatory = switchFeaturesMandatory;
 164         this.primaryConnectionContext = Preconditions.checkNotNull(primaryConnectionContext);
 165         this.deviceState = Preconditions.checkNotNull(deviceState);
 166         this.dataBroker = Preconditions.checkNotNull(dataBroker);
 167         Preconditions.checkNotNull(conductor);
 168         this.outboundQueueProvider = Preconditions.checkNotNull(outboundQueueProvider);
 169         this.transactionChainManager = new TransactionChainManager(dataBroker, deviceState, conductor);
 170         auxiliaryConnectionContexts = new HashMap<>();
 171         deviceFlowRegistry = new DeviceFlowRegistryImpl();
 172         deviceGroupRegistry = new DeviceGroupRegistryImpl();
 173         deviceMeterRegistry = new DeviceMeterRegistryImpl();
 174         messageSpy = conductor.getMessageIntelligenceAgency();
 175
 176         packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(),
 177                 /*initial*/ 1000, /*initial*/2000, messageSpy, REJECTED_DRAIN_FACTOR);
 178
 179         this.translatorLibrary = translatorLibrary;
 180         portStatusTranslator = translatorLibrary.lookupTranslator(
 181                 new TranslatorKey(deviceState.getVersion(), PortGrouping.class.getName()));
 182         packetInTranslator = translatorLibrary.lookupTranslator(
 183                 new TranslatorKey(deviceState.getVersion(), PacketIn.class.getName()));
 184         flowRemovedTranslator = translatorLibrary.lookupTranslator(
 185                 new TranslatorKey(deviceState.getVersion(), FlowRemoved.class.getName()));
 186
 187
 188         nodeConnectorCache = new ConcurrentHashMap<>();
 189
 190         itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl();
 191         flowLifeCycleKeeper = new ItemLifeCycleSourceImpl();
 192         itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper);
 193         deviceCtxState = DEVICE_CONTEXT_STATE.INITIALIZATION;
 194
 195         nodeId = primaryConnectionContext.getNodeId();
 196     }
 197
 198     /**
 199      * This method is called from {@link DeviceManagerImpl} only. So we could say "posthandshake process finish"
 200      * and we are able to set a scheduler for an automatic transaction submitting by time (0,5sec).
 201      */
 202     void initialSubmitTransaction() {
 203         transactionChainManager.initialSubmitWriteTransaction();
 204     }
 205
 206     @Override
 207     public Long reserveXidForDeviceMessage() {
 208         return outboundQueueProvider.reserveEntry();
 209     }
 210
 211     @Override
 212     public void addAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 213         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 214         auxiliaryConnectionContexts.put(connectionDistinguisher, connectionContext);
 215     }
 216
 217     private static SwitchConnectionDistinguisher createConnectionDistinguisher(final ConnectionContext connectionContext) {
 218         return new SwitchConnectionCookieOFImpl(connectionContext.getFeatures().getAuxiliaryId());
 219     }
 220
 221     @Override
 222     public void removeAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 223         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 224         LOG.debug("auxiliary connection dropped: {}, nodeId:{}", connectionContext.getConnectionAdapter()
 225                 .getRemoteAddress(), nodeId);
 226         auxiliaryConnectionContexts.remove(connectionDistinguisher);
 227     }
 228
 229     @Override
 230     public DeviceState getDeviceState() {
 231         return deviceState;
 232     }
 233
 234     @Override
 235     public ReadOnlyTransaction getReadTransaction() {
 236         return dataBroker.newReadOnlyTransaction();
 237     }
 238
 239     @Override
 240     public ListenableFuture<Void> onClusterRoleChange(final OfpRole oldRole, @CheckForNull final OfpRole role) {
 241         LOG.trace("onClusterRoleChange {} for node:", role, nodeId);
 242         Preconditions.checkArgument(role != null);
 243         if (role.equals(oldRole)) {
 244             LOG.debug("Demanded role change for device {} is not changed. OldRole: {}, NewRole {}", nodeId, oldRole, role);
 245             return Futures.immediateFuture(null);
 246         }
 247         if (OfpRole.BECOMEMASTER.equals(role)) {
 248             return onDeviceTakeClusterLeadership();
 249         } else if (OfpRole.BECOMESLAVE.equals(role)) {
 250             return onDeviceLostClusterLeadership();
 251         } else {
 252             LOG.warn("Unknown OFCluster Role {} for Node {}", role, nodeId);
 253             if (null != rpcContext) {
 254                 MdSalRegistrationUtils.unregisterServices(rpcContext);
 255             }
 256             return transactionChainManager.deactivateTransactionManager();
 257         }
 258     }
 259
 260     @Override
 261     public ListenableFuture<Void> onDeviceLostClusterLeadership() {
 262         LOG.trace("onDeviceLostClusterLeadership for node: {}", nodeId);
 263         if (null != rpcContext) {
 264             MdSalRegistrationUtils.registerSlaveServices(rpcContext, OfpRole.BECOMESLAVE);
 265         }
 266         return transactionChainManager.deactivateTransactionManager();
 267     }
 268
 269     @Override
 270     public ListenableFuture<Void> onDeviceTakeClusterLeadership() {
 271         LOG.trace("onDeviceTakeClusterLeadership for node: {}", nodeId);
 272         /* validation */
 273         if (statisticsContext == null) {
 274             final String errMsg = String.format("DeviceCtx %s is up but we are missing StatisticsContext", nodeId);
 275             LOG.warn(errMsg);
 276             return Futures.immediateFailedFuture(new IllegalStateException(errMsg));
 277         }
 278         if (rpcContext == null) {
 279             final String errMsg = String.format("DeviceCtx %s is up but we are missing RpcContext", nodeId);
 280             LOG.warn(errMsg);
 281             return Futures.immediateFailedFuture(new IllegalStateException(errMsg));
 282         }
 283         /* Routed RPC registration */
 284         MdSalRegistrationUtils.registerMasterServices(getRpcContext(), DeviceContextImpl.this, OfpRole.BECOMEMASTER);
 285
 286         if (isStatisticsRpcEnabled) {
 287             MdSalRegistrationUtils.registerStatCompatibilityServices(getRpcContext(), this,
 288                     notificationPublishService, new AtomicLong());
 289         }
 290
 291         /* Prepare init info collecting */
 292         getDeviceState().setDeviceSynchronized(false);
 293         transactionChainManager.activateTransactionManager();
 294         /* Init Collecting NodeInfo */
 295         final ListenableFuture<Void> initCollectingDeviceInfo = DeviceInitializationUtils.initializeNodeInformation(
 296                 DeviceContextImpl.this, switchFeaturesMandatory);
 297         /* Init Collecting StatInfo */
 298         final ListenableFuture<Boolean> statPollFuture = Futures.transform(initCollectingDeviceInfo,
 299                 new AsyncFunction<Void, Boolean>() {
 300
 301                     @Override
 302                     public ListenableFuture<Boolean> apply(@Nonnull final Void input) throws Exception {
 303                         getStatisticsContext().statListForCollectingInitialization();
 304                         return getStatisticsContext().gatherDynamicData();
 305                     }
 306                 });
 307
 308         return Futures.transform(statPollFuture, new Function<Boolean, Void>() {
 309
 310             @Override
 311             public Void apply(final Boolean input) {
 312                 if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
 313                     final String errMsg = String.format("We lost connection for Device %s, context has to be closed.",
 314                             getDeviceState().getNodeId());
 315                     LOG.warn(errMsg);
 316                     throw new IllegalStateException(errMsg);
 317                 }
 318                 if (!input) {
 319                     final String errMsg = String.format("Get Initial Device %s information fails",
 320                             getDeviceState().getNodeId());
 321                     LOG.warn(errMsg);
 322                     throw new IllegalStateException(errMsg);
 323                 }
 324                 LOG.debug("Get Initial Device {} information is successful", nodeId);
 325                 getDeviceState().setDeviceSynchronized(true);
 326                 initialSubmitTransaction();
 327                 getDeviceState().setStatisticsPollingEnabledProp(true);
 328                 return null;
 329             }
 330         });
 331     }
 332
 333     @Override
 334     public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
 335                                                           final InstanceIdentifier<T> path, final T data) throws Exception {
 336         transactionChainManager.writeToTransaction(store, path, data);
 337     }
 338
 339     @Override
 340     public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) throws Exception {
 341         transactionChainManager.addDeleteOperationTotTxChain(store, path);
 342     }
 343
 344     @Override
 345     public boolean submitTransaction() {
 346         return transactionChainManager.submitWriteTransaction();
 347     }
 348
 349     @Override
 350     public ConnectionContext getPrimaryConnectionContext() {
 351         return primaryConnectionContext;
 352     }
 353
 354     @Override
 355     public ConnectionContext getAuxiliaryConnectiobContexts(final BigInteger cookie) {
 356         return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue()));
 357     }
 358
 359     @Override
 360     public DeviceFlowRegistry getDeviceFlowRegistry() {
 361         return deviceFlowRegistry;
 362     }
 363
 364     @Override
 365     public DeviceGroupRegistry getDeviceGroupRegistry() {
 366         return deviceGroupRegistry;
 367     }
 368
 369     @Override
 370     public DeviceMeterRegistry getDeviceMeterRegistry() {
 371         return deviceMeterRegistry;
 372     }
 373
 374     @Override
 375     public void processReply(final OfHeader ofHeader) {
 376         if (ofHeader instanceof Error) {
 377             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 378         } else {
 379             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 380         }
 381     }
 382
 383     @Override
 384     public void processReply(final Xid xid, final List<MultipartReply> ofHeaderList) {
 385         for (final MultipartReply multipartReply : ofHeaderList) {
 386             messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 387         }
 388     }
 389
 390     @Override
 391     public void processFlowRemovedMessage(final FlowRemoved flowRemoved) {
 392         final ItemLifecycleListener itemLifecycleListener = flowLifeCycleKeeper.getItemLifecycleListener();
 393         if (itemLifecycleListener != null) {
 394             //1. translate to general flow (table, priority, match, cookie)
 395             final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification =
 396                     flowRemovedTranslator.translate(flowRemoved, this.getDeviceState(), null);
 397             //2. create registry key
 398             final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(flowRemovedNotification);
 399             //3. lookup flowId
 400             final FlowDescriptor flowDescriptor = deviceFlowRegistry.retrieveIdForFlow(flowRegKey);
 401             //4. if flowId present:
 402             if (flowDescriptor != null) {
 403                 // a) construct flow path
 404                 final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = getDeviceState().getNodeInstanceIdentifier()
 405                         .augmentation(FlowCapableNode.class)
 406                         .child(Table.class, flowDescriptor.getTableKey())
 407                         .child(Flow.class, new FlowKey(flowDescriptor.getFlowId()));
 408                 // b) notify listener
 409                 itemLifecycleListener.onRemoved(flowPath);
 410             } else {
 411                 LOG.debug("flow id not found: nodeId={} tableId={}, priority={}",
 412                         getDeviceState().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority());
 413             }
 414         }
 415     }
 416
 417     @Override
 418     public void processPortStatusMessage(final PortStatusMessage portStatus) {
 419         messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 420         final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatus, this.getDeviceState(), null);
 421
 422         final KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> iiToNodeConnector = provideIIToNodeConnector(portStatus.getPortNo(), portStatus.getVersion());
 423         try {
 424             if (portStatus.getReason().equals(PortReason.OFPPRADD) || portStatus.getReason().equals(PortReason.OFPPRMODIFY)) {
 425                 // because of ADD status node connector has to be created
 426                 final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey());
 427                 nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build());
 428                 nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector);
 429                 writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build());
 430             } else if (portStatus.getReason().equals(PortReason.OFPPRDELETE)) {
 431                 addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
 432             }
 433             submitTransaction();
 434         } catch (final Exception e) {
 435             LOG.warn("Error processing port status message: {}", e.getMessage());
 436         }
 437     }
 438
 439     private KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> provideIIToNodeConnector(final long portNo, final short version) {
 440         final InstanceIdentifier<Node> iiToNodes = deviceState.getNodeInstanceIdentifier();
 441         final BigInteger dataPathId = deviceState.getFeatures().getDatapathId();
 442         final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), portNo, version);
 443         return iiToNodes.child(NodeConnector.class, new NodeConnectorKey(nodeConnectorId));
 444     }
 445
 446     @Override
 447     public void processPacketInMessage(final PacketInMessage packetInMessage) {
 448         messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH);
 449         final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter();
 450         final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, this.getDeviceState(), null);
 451
 452         if (packetReceived == null) {
 453             LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress());
 454             messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE);
 455             return;
 456         } else {
 457             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_OUT_SUCCESS);
 458         }
 459
 460         if (!packetInLimiter.acquirePermit()) {
 461             LOG.debug("Packet limited");
 462             // TODO: save packet into emergency slot if possible
 463             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PACKET_IN_LIMIT_REACHED_AND_DROPPED);
 464             return;
 465         }
 466
 467         final ListenableFuture<?> offerNotification = notificationPublishService.offerNotification(packetReceived);
 468         if (NotificationPublishService.REJECTED.equals(offerNotification)) {
 469             LOG.debug("notification offer rejected");
 470             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 471             packetInLimiter.drainLowWaterMark();
 472             packetInLimiter.releasePermit();
 473             return;
 474         }
 475
 476         Futures.addCallback(offerNotification, new FutureCallback<Object>() {
 477             @Override
 478             public void onSuccess(final Object result) {
 479                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 480                 packetInLimiter.releasePermit();
 481             }
 482
 483             @Override
 484             public void onFailure(final Throwable t) {
 485                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 486                 LOG.debug("notification offer failed: {}", t.getMessage());
 487                 LOG.trace("notification offer failed..", t);
 488                 packetInLimiter.releasePermit();
 489             }
 490         });
 491     }
 492
 493     @Override
 494     public void processExperimenterMessage(final ExperimenterMessage notification) {
 495         // lookup converter
 496         final ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice();
 497         final MessageTypeKey<? extends ExperimenterDataOfChoice> key = new MessageTypeKey<>(
 498                 deviceState.getVersion(),
 499                 (Class<? extends ExperimenterDataOfChoice>) vendorData.getImplementedInterface());
 500         final ConvertorMessageFromOFJava<ExperimenterDataOfChoice, MessagePath> messageConverter = extensionConverterProvider.getMessageConverter(key);
 501         if (messageConverter == null) {
 502             LOG.warn("custom converter for {}[OF:{}] not found",
 503                     notification.getExperimenterDataOfChoice().getImplementedInterface(),
 504                     deviceState.getVersion());
 505             return;
 506         }
 507         // build notification
 508         final ExperimenterMessageOfChoice messageOfChoice;
 509         try {
 510             messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION);
 511             final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new ExperimenterMessageFromDevBuilder()
 512                 .setNode(new NodeRef(deviceState.getNodeInstanceIdentifier()))
 513                     .setExperimenterMessageOfChoice(messageOfChoice);
 514             // publish
 515             notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
 516         } catch (final ConversionException e) {
 517             LOG.error("Conversion of experimenter notification failed", e);
 518         }
 519     }
 520
 521     @Override
 522     public TranslatorLibrary oook() {
 523         return translatorLibrary;
 524     }
 525
 526     @Override
 527     public synchronized void close() {
 528         LOG.debug("closing deviceContext: {}, nodeId:{}",
 529                 getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress(),
 530                 getDeviceState().getNodeId());
 531         // NOOP
 532         throw new UnsupportedOperationException("Autocloseble.close will be removed soon");
 533     }
 534
 535     @Override
 536     public void setCurrentBarrierTimeout(final Timeout timeout) {
 537         barrierTaskTimeout = timeout;
 538     }
 539
 540     @Override
 541     public Timeout getBarrierTaskTimeout() {
 542         return barrierTaskTimeout;
 543     }
 544
 545     @Override
 546     public void setNotificationPublishService(final NotificationPublishService notificationPublishService) {
 547         this.notificationPublishService = notificationPublishService;
 548     }
 549
 550     @Override
 551     public MessageSpy getMessageSpy() {
 552         return messageSpy;
 553     }
 554
 555     @Override
 556     public void onPublished() {
 557         Verify.verify(DEVICE_CONTEXT_STATE.INITIALIZATION.equals(deviceCtxState));
 558         deviceCtxState = DEVICE_CONTEXT_STATE.WORKING;
 559         primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 560         for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) {
 561             switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 562         }
 563     }
 564
 565     @Override
 566     public MultiMsgCollector getMultiMsgCollector(final RequestContext<List<MultipartReply>> requestContext) {
 567         return new MultiMsgCollectorImpl(this, requestContext);
 568     }
 569
 570     @Override
 571     public NodeConnectorRef lookupNodeConnectorRef(final Long portNumber) {
 572         return nodeConnectorCache.get(portNumber);
 573     }
 574
 575     @Override
 576     public void storeNodeConnectorRef(@Nonnull final Long portNumber, @Nonnull final NodeConnectorRef nodeConnectorRef) {
 577         nodeConnectorCache.put(
 578                 Preconditions.checkNotNull(portNumber),
 579                 Preconditions.checkNotNull(nodeConnectorRef));
 580     }
 581
 582     @Override
 583     public void updatePacketInRateLimit(final long upperBound) {
 584         packetInLimiter.changeWaterMarks((int) (LOW_WATERMARK_FACTOR * upperBound), (int) (HIGH_WATERMARK_FACTOR * upperBound));
 585     }
 586
 587     @Override
 588     public ItemLifeCycleRegistry getItemLifeCycleSourceRegistry() {
 589         return itemLifeCycleSourceRegistry;
 590     }
 591
 592     @Override
 593     public void setRpcContext(final RpcContext rpcContext) {
 594         this.rpcContext = rpcContext;
 595     }
 596
 597     @Override
 598     public RpcContext getRpcContext() {
 599         return rpcContext;
 600     }
 601
 602     @Override
 603     public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
 604         this.extensionConverterProvider = extensionConverterProvider;
 605     }
 606
 607     @Override
 608     public ExtensionConverterProvider getExtensionConverterProvider() {
 609         return extensionConverterProvider;
 610     }
 611
 612     @Override
 613     public void setStatisticsContext(final StatisticsContext statisticsContext) {
 614         this.statisticsContext = statisticsContext;
 615     }
 616
 617     @Override
 618     public StatisticsContext getStatisticsContext() {
 619         return statisticsContext;
 620     }
 621
 622     @Override
 623     public synchronized void shutdownConnection() {
 624         LOG.debug("Shutdown method for node {}", nodeId);
 625         deviceState.setValid(false);
 626         if (DEVICE_CONTEXT_STATE.TERMINATION.equals(deviceCtxState)) {
 627             LOG.debug("DeviceCtx for Node {} is in termination process.", nodeId);
 628             return;
 629         }
 630         deviceCtxState = DEVICE_CONTEXT_STATE.TERMINATION;
 631
 632         if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
 633             LOG.debug("ConnectionCtx for Node {} is in RIP state.", deviceState.getNodeId());
 634             return;
 635         }
 636         /* Terminate Auxiliary Connection */
 637         for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
 638             LOG.debug("Closing auxiliary connection {}", connectionContext.getNodeId());
 639             connectionContext.closeConnection(false);
 640         }
 641         /* Terminate Primary Connection */
 642         getPrimaryConnectionContext().closeConnection(true);
 643         /* Close all Group Registry */
 644         deviceGroupRegistry.close();
 645         deviceFlowRegistry.close();
 646         deviceMeterRegistry.close();
 647     }
 648
 649     @Override
 650     public void setStatisticsRpcEnabled(boolean isStatisticsRpcEnabled) {
 651         this.isStatisticsRpcEnabled = isStatisticsRpcEnabled;
 652     }
 653
 654     @Override
 655     public DEVICE_CONTEXT_STATE getDeviceContextState() {
 656         return deviceCtxState;
 657     }
 658
 659     @Override
 660     public ListenableFuture<Void> shuttingDownDataStoreTransactions() {
 661         deviceState.setValid(false);
 662         return transactionChainManager.shuttingDown();
 663     }
 664
 665     @VisibleForTesting
 666     TransactionChainManager getTransactionChainManager() {
 667         return this.transactionChainManager;
 668     }
 669 }