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