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             } else {
 416                 LOG.debug("flow id not found: nodeId={} tableId={}, priority={}",
 417                         getDeviceState().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority());
 418             }
 419         }
 420     }
 421
 422     @Override
 423     public void processPortStatusMessage(final PortStatusMessage portStatus) {
 424         messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 425         final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatus, this.getDeviceState(), null);
 426
 427         final KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> iiToNodeConnector = provideIIToNodeConnector(portStatus.getPortNo(), portStatus.getVersion());
 428         try {
 429             if (portStatus.getReason().equals(PortReason.OFPPRADD) || portStatus.getReason().equals(PortReason.OFPPRMODIFY)) {
 430                 // because of ADD status node connector has to be created
 431                 final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey());
 432                 nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build());
 433                 nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector);
 434                 writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build());
 435             } else if (portStatus.getReason().equals(PortReason.OFPPRDELETE)) {
 436                 addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
 437             }
 438             submitTransaction();
 439         } catch (final Exception e) {
 440             LOG.warn("Error processing port status message: {}", e.getMessage());
 441         }
 442     }
 443
 444     private KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> provideIIToNodeConnector(final long portNo, final short version) {
 445         final InstanceIdentifier<Node> iiToNodes = deviceState.getNodeInstanceIdentifier();
 446         final BigInteger dataPathId = deviceState.getFeatures().getDatapathId();
 447         final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), portNo, version);
 448         return iiToNodes.child(NodeConnector.class, new NodeConnectorKey(nodeConnectorId));
 449     }
 450
 451     @Override
 452     public void processPacketInMessage(final PacketInMessage packetInMessage) {
 453         messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH);
 454         final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter();
 455         final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, this.getDeviceState(), null);
 456
 457         if (packetReceived == null) {
 458             LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress());
 459             messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE);
 460             return;
 461         } else {
 462             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_OUT_SUCCESS);
 463         }
 464
 465         if (!packetInLimiter.acquirePermit()) {
 466             LOG.debug("Packet limited");
 467             // TODO: save packet into emergency slot if possible
 468             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PACKET_IN_LIMIT_REACHED_AND_DROPPED);
 469             return;
 470         }
 471
 472         final ListenableFuture<?> offerNotification = notificationPublishService.offerNotification(packetReceived);
 473         if (NotificationPublishService.REJECTED.equals(offerNotification)) {
 474             LOG.debug("notification offer rejected");
 475             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 476             packetInLimiter.drainLowWaterMark();
 477             packetInLimiter.releasePermit();
 478             return;
 479         }
 480
 481         Futures.addCallback(offerNotification, new FutureCallback<Object>() {
 482             @Override
 483             public void onSuccess(final Object result) {
 484                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 485                 packetInLimiter.releasePermit();
 486             }
 487
 488             @Override
 489             public void onFailure(final Throwable t) {
 490                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 491                 LOG.debug("notification offer failed: {}", t.getMessage());
 492                 LOG.trace("notification offer failed..", t);
 493                 packetInLimiter.releasePermit();
 494             }
 495         });
 496     }
 497
 498     @Override
 499     public void processExperimenterMessage(final ExperimenterMessage notification) {
 500         // lookup converter
 501         final ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice();
 502         final MessageTypeKey<? extends ExperimenterDataOfChoice> key = new MessageTypeKey<>(
 503                 deviceState.getVersion(),
 504                 (Class<? extends ExperimenterDataOfChoice>) vendorData.getImplementedInterface());
 505         final ConvertorMessageFromOFJava<ExperimenterDataOfChoice, MessagePath> messageConverter = extensionConverterProvider.getMessageConverter(key);
 506         if (messageConverter == null) {
 507             LOG.warn("custom converter for {}[OF:{}] not found",
 508                     notification.getExperimenterDataOfChoice().getImplementedInterface(),
 509                     deviceState.getVersion());
 510             return;
 511         }
 512         // build notification
 513         final ExperimenterMessageOfChoice messageOfChoice;
 514         try {
 515             messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION);
 516             final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new ExperimenterMessageFromDevBuilder()
 517                 .setNode(new NodeRef(deviceState.getNodeInstanceIdentifier()))
 518                     .setExperimenterMessageOfChoice(messageOfChoice);
 519             // publish
 520             notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
 521         } catch (final ConversionException e) {
 522             LOG.error("Conversion of experimenter notification failed", e);
 523         }
 524     }
 525
 526     @Override
 527     public TranslatorLibrary oook() {
 528         return translatorLibrary;
 529     }
 530
 531     @Override
 532     public synchronized void close() {
 533         LOG.debug("closing deviceContext: {}, nodeId:{}",
 534                 getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress(),
 535                 getDeviceState().getNodeId());
 536         // NOOP
 537         throw new UnsupportedOperationException("Autocloseble.close will be removed soon");
 538     }
 539
 540     @Override
 541     public void setCurrentBarrierTimeout(final Timeout timeout) {
 542         barrierTaskTimeout = timeout;
 543     }
 544
 545     @Override
 546     public Timeout getBarrierTaskTimeout() {
 547         return barrierTaskTimeout;
 548     }
 549
 550     @Override
 551     public void setNotificationPublishService(final NotificationPublishService notificationPublishService) {
 552         this.notificationPublishService = notificationPublishService;
 553     }
 554
 555     @Override
 556     public MessageSpy getMessageSpy() {
 557         return messageSpy;
 558     }
 559
 560     @Override
 561     public void onPublished() {
 562         Verify.verify(DEVICE_CONTEXT_STATE.INITIALIZATION.equals(deviceCtxState));
 563         deviceCtxState = DEVICE_CONTEXT_STATE.WORKING;
 564         primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 565         for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) {
 566             switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 567         }
 568     }
 569
 570     @Override
 571     public MultiMsgCollector getMultiMsgCollector(final RequestContext<List<MultipartReply>> requestContext) {
 572         return new MultiMsgCollectorImpl(this, requestContext);
 573     }
 574
 575     @Override
 576     public NodeConnectorRef lookupNodeConnectorRef(final Long portNumber) {
 577         return nodeConnectorCache.get(portNumber);
 578     }
 579
 580     @Override
 581     public void storeNodeConnectorRef(@Nonnull final Long portNumber, @Nonnull final NodeConnectorRef nodeConnectorRef) {
 582         nodeConnectorCache.put(
 583                 Preconditions.checkNotNull(portNumber),
 584                 Preconditions.checkNotNull(nodeConnectorRef));
 585     }
 586
 587     @Override
 588     public void updatePacketInRateLimit(final long upperBound) {
 589         packetInLimiter.changeWaterMarks((int) (LOW_WATERMARK_FACTOR * upperBound), (int) (HIGH_WATERMARK_FACTOR * upperBound));
 590     }
 591
 592     @Override
 593     public ItemLifeCycleRegistry getItemLifeCycleSourceRegistry() {
 594         return itemLifeCycleSourceRegistry;
 595     }
 596
 597     @Override
 598     public void setRpcContext(final RpcContext rpcContext) {
 599         this.rpcContext = rpcContext;
 600     }
 601
 602     @Override
 603     public RpcContext getRpcContext() {
 604         return rpcContext;
 605     }
 606
 607     @Override
 608     public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
 609         this.extensionConverterProvider = extensionConverterProvider;
 610     }
 611
 612     @Override
 613     public ExtensionConverterProvider getExtensionConverterProvider() {
 614         return extensionConverterProvider;
 615     }
 616
 617     @Override
 618     public void setStatisticsContext(final StatisticsContext statisticsContext) {
 619         this.statisticsContext = statisticsContext;
 620     }
 621
 622     @Override
 623     public StatisticsContext getStatisticsContext() {
 624         return statisticsContext;
 625     }
 626
 627     @Override
 628     public synchronized void shutdownConnection() {
 629         LOG.debug("Shutdown method for node {}", nodeId);
 630         deviceState.setValid(false);
 631         if (DEVICE_CONTEXT_STATE.TERMINATION.equals(deviceCtxState)) {
 632             LOG.debug("DeviceCtx for Node {} is in termination process.", nodeId);
 633             return;
 634         }
 635         deviceCtxState = DEVICE_CONTEXT_STATE.TERMINATION;
 636
 637         if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
 638             LOG.debug("ConnectionCtx for Node {} is in RIP state.", deviceState.getNodeId());
 639             return;
 640         }
 641         /* Terminate Auxiliary Connection */
 642         for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
 643             LOG.debug("Closing auxiliary connection {}", connectionContext.getNodeId());
 644             connectionContext.closeConnection(false);
 645         }
 646         /* Terminate Primary Connection */
 647         getPrimaryConnectionContext().closeConnection(true);
 648         /* Close all Group Registry */
 649         deviceGroupRegistry.close();
 650         deviceFlowRegistry.close();
 651         deviceMeterRegistry.close();
 652     }
 653
 654     @Override
 655     public void setStatisticsRpcEnabled(boolean isStatisticsRpcEnabled) {
 656         this.isStatisticsRpcEnabled = isStatisticsRpcEnabled;
 657     }
 658
 659     @Override
 660     public DEVICE_CONTEXT_STATE getDeviceContextState() {
 661         return deviceCtxState;
 662     }
 663
 664     @Override
 665     public ListenableFuture<Void> shuttingDownDataStoreTransactions() {
 666         deviceState.setValid(false);
 667         return transactionChainManager.shuttingDown();
 668     }
 669
 670     @VisibleForTesting
 671     TransactionChainManager getTransactionChainManager() {
 672         return this.transactionChainManager;
 673     }
 674 }