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, false);
 337     }
 338
 339     @Override
 340     public <T extends DataObject> void writeToTransactionWithParentsSlow(LogicalDatastoreType store, InstanceIdentifier<T> path, T data) throws Exception {
 341         transactionChainManager.writeToTransaction(store, path, data, true);
 342     }
 343
 344     @Override
 345     public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) throws Exception {
 346         transactionChainManager.addDeleteOperationTotTxChain(store, path);
 347     }
 348
 349     @Override
 350     public boolean submitTransaction() {
 351         return transactionChainManager.submitWriteTransaction();
 352     }
 353
 354     @Override
 355     public ConnectionContext getPrimaryConnectionContext() {
 356         return primaryConnectionContext;
 357     }
 358
 359     @Override
 360     public ConnectionContext getAuxiliaryConnectiobContexts(final BigInteger cookie) {
 361         return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue()));
 362     }
 363
 364     @Override
 365     public DeviceFlowRegistry getDeviceFlowRegistry() {
 366         return deviceFlowRegistry;
 367     }
 368
 369     @Override
 370     public DeviceGroupRegistry getDeviceGroupRegistry() {
 371         return deviceGroupRegistry;
 372     }
 373
 374     @Override
 375     public DeviceMeterRegistry getDeviceMeterRegistry() {
 376         return deviceMeterRegistry;
 377     }
 378
 379     @Override
 380     public void processReply(final OfHeader ofHeader) {
 381         if (ofHeader instanceof Error) {
 382             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 383         } else {
 384             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 385         }
 386     }
 387
 388     @Override
 389     public void processReply(final Xid xid, final List<MultipartReply> ofHeaderList) {
 390         for (final MultipartReply multipartReply : ofHeaderList) {
 391             messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 392         }
 393     }
 394
 395     @Override
 396     public void processFlowRemovedMessage(final FlowRemoved flowRemoved) {
 397         final ItemLifecycleListener itemLifecycleListener = flowLifeCycleKeeper.getItemLifecycleListener();
 398         if (itemLifecycleListener != null) {
 399             //1. translate to general flow (table, priority, match, cookie)
 400             final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification =
 401                     flowRemovedTranslator.translate(flowRemoved, this.getDeviceState(), null);
 402             //2. create registry key
 403             final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(flowRemovedNotification);
 404             //3. lookup flowId
 405             final FlowDescriptor flowDescriptor = deviceFlowRegistry.retrieveIdForFlow(flowRegKey);
 406             //4. if flowId present:
 407             if (flowDescriptor != null) {
 408                 // a) construct flow path
 409                 final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = getDeviceState().getNodeInstanceIdentifier()
 410                         .augmentation(FlowCapableNode.class)
 411                         .child(Table.class, flowDescriptor.getTableKey())
 412                         .child(Flow.class, new FlowKey(flowDescriptor.getFlowId()));
 413                 // b) notify listener
 414                 itemLifecycleListener.onRemoved(flowPath);
 415                 // c) trigger off a notification
 416                 notificationPublishService.offerNotification(flowRemovedNotification);
 417             } else {
 418                 LOG.debug("flow id not found: nodeId={} tableId={}, priority={}",
 419                         getDeviceState().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority());
 420             }
 421         }
 422     }
 423
 424     @Override
 425     public void processPortStatusMessage(final PortStatusMessage portStatus) {
 426         messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 427         final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatus, this.getDeviceState(), null);
 428
 429         final KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> iiToNodeConnector = provideIIToNodeConnector(portStatus.getPortNo(), portStatus.getVersion());
 430         try {
 431             if (portStatus.getReason().equals(PortReason.OFPPRADD) || portStatus.getReason().equals(PortReason.OFPPRMODIFY)) {
 432                 // because of ADD status node connector has to be created
 433                 final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey());
 434                 nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build());
 435                 nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector);
 436                 writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build());
 437             } else if (portStatus.getReason().equals(PortReason.OFPPRDELETE)) {
 438                 addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
 439             }
 440             submitTransaction();
 441         } catch (final Exception e) {
 442             LOG.warn("Error processing port status message: {}", e.getMessage());
 443         }
 444     }
 445
 446     private KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> provideIIToNodeConnector(final long portNo, final short version) {
 447         final InstanceIdentifier<Node> iiToNodes = deviceState.getNodeInstanceIdentifier();
 448         final BigInteger dataPathId = deviceState.getFeatures().getDatapathId();
 449         final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), portNo, version);
 450         return iiToNodes.child(NodeConnector.class, new NodeConnectorKey(nodeConnectorId));
 451     }
 452
 453     @Override
 454     public void processPacketInMessage(final PacketInMessage packetInMessage) {
 455         messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH);
 456         final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter();
 457         final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, this.getDeviceState(), null);
 458
 459         if (packetReceived == null) {
 460             LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress());
 461             messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE);
 462             return;
 463         } else {
 464             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_OUT_SUCCESS);
 465         }
 466
 467         if (!packetInLimiter.acquirePermit()) {
 468             LOG.debug("Packet limited");
 469             // TODO: save packet into emergency slot if possible
 470             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PACKET_IN_LIMIT_REACHED_AND_DROPPED);
 471             return;
 472         }
 473
 474         final ListenableFuture<?> offerNotification = notificationPublishService.offerNotification(packetReceived);
 475         if (NotificationPublishService.REJECTED.equals(offerNotification)) {
 476             LOG.debug("notification offer rejected");
 477             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 478             packetInLimiter.drainLowWaterMark();
 479             packetInLimiter.releasePermit();
 480             return;
 481         }
 482
 483         Futures.addCallback(offerNotification, new FutureCallback<Object>() {
 484             @Override
 485             public void onSuccess(final Object result) {
 486                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 487                 packetInLimiter.releasePermit();
 488             }
 489
 490             @Override
 491             public void onFailure(final Throwable t) {
 492                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 493                 LOG.debug("notification offer failed: {}", t.getMessage());
 494                 LOG.trace("notification offer failed..", t);
 495                 packetInLimiter.releasePermit();
 496             }
 497         });
 498     }
 499
 500     @Override
 501     public void processExperimenterMessage(final ExperimenterMessage notification) {
 502         // lookup converter
 503         final ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice();
 504         final MessageTypeKey<? extends ExperimenterDataOfChoice> key = new MessageTypeKey<>(
 505                 deviceState.getVersion(),
 506                 (Class<? extends ExperimenterDataOfChoice>) vendorData.getImplementedInterface());
 507         final ConvertorMessageFromOFJava<ExperimenterDataOfChoice, MessagePath> messageConverter = extensionConverterProvider.getMessageConverter(key);
 508         if (messageConverter == null) {
 509             LOG.warn("custom converter for {}[OF:{}] not found",
 510                     notification.getExperimenterDataOfChoice().getImplementedInterface(),
 511                     deviceState.getVersion());
 512             return;
 513         }
 514         // build notification
 515         final ExperimenterMessageOfChoice messageOfChoice;
 516         try {
 517             messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION);
 518             final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new ExperimenterMessageFromDevBuilder()
 519                 .setNode(new NodeRef(deviceState.getNodeInstanceIdentifier()))
 520                     .setExperimenterMessageOfChoice(messageOfChoice);
 521             // publish
 522             notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
 523         } catch (final ConversionException e) {
 524             LOG.error("Conversion of experimenter notification failed", e);
 525         }
 526     }
 527
 528     @Override
 529     public TranslatorLibrary oook() {
 530         return translatorLibrary;
 531     }
 532
 533     @Override
 534     public synchronized void close() {
 535         LOG.debug("closing deviceContext: {}, nodeId:{}",
 536                 getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress(),
 537                 getDeviceState().getNodeId());
 538         // NOOP
 539         throw new UnsupportedOperationException("Autocloseble.close will be removed soon");
 540     }
 541
 542     @Override
 543     public void setCurrentBarrierTimeout(final Timeout timeout) {
 544         barrierTaskTimeout = timeout;
 545     }
 546
 547     @Override
 548     public Timeout getBarrierTaskTimeout() {
 549         return barrierTaskTimeout;
 550     }
 551
 552     @Override
 553     public void setNotificationPublishService(final NotificationPublishService notificationPublishService) {
 554         this.notificationPublishService = notificationPublishService;
 555     }
 556
 557     @Override
 558     public MessageSpy getMessageSpy() {
 559         return messageSpy;
 560     }
 561
 562     @Override
 563     public void onPublished() {
 564         Verify.verify(DEVICE_CONTEXT_STATE.INITIALIZATION.equals(deviceCtxState));
 565         deviceCtxState = DEVICE_CONTEXT_STATE.WORKING;
 566         primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 567         for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) {
 568             switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 569         }
 570     }
 571
 572     @Override
 573     public MultiMsgCollector getMultiMsgCollector(final RequestContext<List<MultipartReply>> requestContext) {
 574         return new MultiMsgCollectorImpl(this, requestContext);
 575     }
 576
 577     @Override
 578     public NodeConnectorRef lookupNodeConnectorRef(final Long portNumber) {
 579         return nodeConnectorCache.get(portNumber);
 580     }
 581
 582     @Override
 583     public void storeNodeConnectorRef(@Nonnull final Long portNumber, @Nonnull final NodeConnectorRef nodeConnectorRef) {
 584         nodeConnectorCache.put(
 585                 Preconditions.checkNotNull(portNumber),
 586                 Preconditions.checkNotNull(nodeConnectorRef));
 587     }
 588
 589     @Override
 590     public void updatePacketInRateLimit(final long upperBound) {
 591         packetInLimiter.changeWaterMarks((int) (LOW_WATERMARK_FACTOR * upperBound), (int) (HIGH_WATERMARK_FACTOR * upperBound));
 592     }
 593
 594     @Override
 595     public ItemLifeCycleRegistry getItemLifeCycleSourceRegistry() {
 596         return itemLifeCycleSourceRegistry;
 597     }
 598
 599     @Override
 600     public void setRpcContext(final RpcContext rpcContext) {
 601         this.rpcContext = rpcContext;
 602     }
 603
 604     @Override
 605     public RpcContext getRpcContext() {
 606         return rpcContext;
 607     }
 608
 609     @Override
 610     public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
 611         this.extensionConverterProvider = extensionConverterProvider;
 612     }
 613
 614     @Override
 615     public ExtensionConverterProvider getExtensionConverterProvider() {
 616         return extensionConverterProvider;
 617     }
 618
 619     @Override
 620     public void setStatisticsContext(final StatisticsContext statisticsContext) {
 621         this.statisticsContext = statisticsContext;
 622     }
 623
 624     @Override
 625     public StatisticsContext getStatisticsContext() {
 626         return statisticsContext;
 627     }
 628
 629     @Override
 630     public synchronized void shutdownConnection() {
 631         LOG.debug("Shutdown method for node {}", nodeId);
 632         deviceState.setValid(false);
 633         if (DEVICE_CONTEXT_STATE.TERMINATION.equals(deviceCtxState)) {
 634             LOG.debug("DeviceCtx for Node {} is in termination process.", nodeId);
 635             return;
 636         }
 637         deviceCtxState = DEVICE_CONTEXT_STATE.TERMINATION;
 638
 639         if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
 640             LOG.debug("ConnectionCtx for Node {} is in RIP state.", deviceState.getNodeId());
 641             return;
 642         }
 643         /* Terminate Auxiliary Connection */
 644         for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
 645             LOG.debug("Closing auxiliary connection {}", connectionContext.getNodeId());
 646             connectionContext.closeConnection(false);
 647         }
 648         /* Terminate Primary Connection */
 649         getPrimaryConnectionContext().closeConnection(true);
 650         /* Close all Group Registry */
 651         deviceGroupRegistry.close();
 652         deviceFlowRegistry.close();
 653         deviceMeterRegistry.close();
 654     }
 655
 656     @Override
 657     public void setStatisticsRpcEnabled(boolean isStatisticsRpcEnabled) {
 658         this.isStatisticsRpcEnabled = isStatisticsRpcEnabled;
 659     }
 660
 661     @Override
 662     public DEVICE_CONTEXT_STATE getDeviceContextState() {
 663         return deviceCtxState;
 664     }
 665
 666     @Override
 667     public ListenableFuture<Void> shuttingDownDataStoreTransactions() {
 668         deviceState.setValid(false);
 669         return transactionChainManager.shuttingDown();
 670     }
 671
 672     @VisibleForTesting
 673     TransactionChainManager getTransactionChainManager() {
 674         return this.transactionChainManager;
 675     }
 676 }