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.Preconditions;
  12 import com.google.common.base.Verify;
  13 import com.google.common.util.concurrent.FutureCallback;
  14 import com.google.common.util.concurrent.Futures;
  15 import com.google.common.util.concurrent.ListenableFuture;
  16 import io.netty.util.Timeout;
  17 import java.math.BigInteger;
  18 import java.util.HashMap;
  19 import java.util.List;
  20 import java.util.Map;
  21 import java.util.concurrent.ExecutionException;
  22 import java.util.concurrent.TimeUnit;
  23 import java.util.concurrent.atomic.AtomicInteger;
  24 import javax.annotation.Nonnull;
  25 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  26 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  27 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  28 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  29 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainClosedException;
  30 import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
  31 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  32 import org.opendaylight.openflowjava.protocol.api.keys.MessageTypeKey;
  33 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
  34 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  35 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  36 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
  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.ClusterInitializationPhaseHandler;
  43 import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
  44 import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleService;
  45 import org.opendaylight.openflowplugin.api.openflow.md.core.SwitchConnectionDistinguisher;
  46 import org.opendaylight.openflowplugin.api.openflow.md.core.TranslatorKey;
  47 import org.opendaylight.openflowplugin.api.openflow.registry.ItemLifeCycleRegistry;
  48 import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
  49 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor;
  50 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowRegistryKey;
  51 import org.opendaylight.openflowplugin.api.openflow.registry.group.DeviceGroupRegistry;
  52 import org.opendaylight.openflowplugin.api.openflow.registry.meter.DeviceMeterRegistry;
  53 import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleKeeper;
  54 import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener;
  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.util.DeviceInitializationUtils;
  69 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
  70 import org.opendaylight.openflowplugin.openflow.md.core.session.SwitchConnectionCookieOFImpl;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.experimenter.message.service.rev151020.ExperimenterMessageFromDevBuilder;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorBuilder;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.PortReason;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.Error;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterMessage;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemoved;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketIn;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortGrouping;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.experimenter.core.ExperimenterDataOfChoice;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.experimenter.types.rev151020.experimenter.core.message.ExperimenterMessageOfChoice;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
  96 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
  97 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
  98 import org.opendaylight.yangtools.yang.binding.DataObject;
  99 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 100 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
 101 import org.slf4j.Logger;
 102 import org.slf4j.LoggerFactory;
 103
 104 public class DeviceContextImpl implements DeviceContext, ExtensionConverterProviderKeeper{
 105
 106     private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class);
 107
 108     // TODO: drain factor should be parametrized
 109     private static final float REJECTED_DRAIN_FACTOR = 0.25f;
 110     // TODO: low water mark factor should be parametrized
 111     private static final float LOW_WATERMARK_FACTOR = 0.75f;
 112     // TODO: high water mark factor should be parametrized
 113     private static final float HIGH_WATERMARK_FACTOR = 0.95f;
 114     private boolean initialized;
 115     private static final Long RETRY_DELAY = 100L;
 116     private static final int RETRY_COUNT = 3;
 117
 118     private ConnectionContext primaryConnectionContext;
 119     private final DeviceState deviceState;
 120     private final DataBroker dataBroker;
 121     private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
 122     private TransactionChainManager transactionChainManager;
 123     private DeviceFlowRegistry deviceFlowRegistry;
 124     private DeviceGroupRegistry deviceGroupRegistry;
 125     private DeviceMeterRegistry deviceMeterRegistry;
 126     private final PacketInRateLimiter packetInLimiter;
 127     private final MessageSpy messageSpy;
 128     private final ItemLifeCycleKeeper flowLifeCycleKeeper;
 129     private NotificationPublishService notificationPublishService;
 130     private Timeout barrierTaskTimeout;
 131     private final MessageTranslator<PortGrouping, FlowCapableNodeConnector> portStatusTranslator;
 132     private final MessageTranslator<PacketInMessage, PacketReceived> packetInTranslator;
 133     private final MessageTranslator<FlowRemoved, org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved> flowRemovedTranslator;
 134     private final TranslatorLibrary translatorLibrary;
 135     private final ItemLifeCycleRegistry itemLifeCycleSourceRegistry;
 136     private ExtensionConverterProvider extensionConverterProvider;
 137     private final DeviceManager deviceManager;
 138     private boolean skipTableFeatures;
 139     private boolean switchFeaturesMandatory;
 140     private final DeviceInfo deviceInfo;
 141     private final ConvertorExecutor convertorExecutor;
 142     private volatile CONTEXT_STATE state;
 143     private ClusterInitializationPhaseHandler clusterInitializationPhaseHandler;
 144
 145     DeviceContextImpl(
 146             @Nonnull final ConnectionContext primaryConnectionContext,
 147             @Nonnull final DataBroker dataBroker,
 148             @Nonnull final MessageSpy messageSpy,
 149             @Nonnull final TranslatorLibrary translatorLibrary,
 150             @Nonnull final DeviceManager manager,
 151             final ConvertorExecutor convertorExecutor,
 152             final boolean skipTableFeatures) {
 153         this.primaryConnectionContext = primaryConnectionContext;
 154         this.deviceInfo = primaryConnectionContext.getDeviceInfo();
 155         this.deviceState = new DeviceStateImpl();
 156         this.dataBroker = dataBroker;
 157         this.auxiliaryConnectionContexts = new HashMap<>();
 158         this.messageSpy = Preconditions.checkNotNull(messageSpy);
 159         this.deviceManager = manager;
 160
 161         this.packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(),
 162                 /*initial*/ 1000, /*initial*/2000, this.messageSpy, REJECTED_DRAIN_FACTOR);
 163
 164         this.translatorLibrary = translatorLibrary;
 165         this.portStatusTranslator = translatorLibrary.lookupTranslator(
 166                 new TranslatorKey(deviceInfo.getVersion(), PortGrouping.class.getName()));
 167         this.packetInTranslator = translatorLibrary.lookupTranslator(
 168                 new TranslatorKey(deviceInfo.getVersion(), PacketIn.class.getName()));
 169         this.flowRemovedTranslator = translatorLibrary.lookupTranslator(
 170                 new TranslatorKey(deviceInfo.getVersion(), FlowRemoved.class.getName()));
 171
 172         this.itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl();
 173         this.flowLifeCycleKeeper = new ItemLifeCycleSourceImpl();
 174         this.itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper);
 175         this.state = CONTEXT_STATE.INITIALIZATION;
 176         this.convertorExecutor = convertorExecutor;
 177         this.skipTableFeatures = skipTableFeatures;
 178         this.initialized = false;
 179     }
 180
 181     @Override
 182     public void initialSubmitTransaction() {
 183         if (initialized) {
 184             transactionChainManager.initialSubmitWriteTransaction();
 185         }
 186     }
 187
 188     @Override
 189     public void addAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 190         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 191         auxiliaryConnectionContexts.put(connectionDistinguisher, connectionContext);
 192     }
 193
 194     private static SwitchConnectionDistinguisher createConnectionDistinguisher(final ConnectionContext connectionContext) {
 195         return new SwitchConnectionCookieOFImpl(connectionContext.getFeatures().getAuxiliaryId());
 196     }
 197
 198     @Override
 199     public void removeAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 200         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 201         LOG.debug("auxiliary connection dropped: {}, nodeId:{}", connectionContext.getConnectionAdapter()
 202                 .getRemoteAddress(), getDeviceInfo().getLOGValue());
 203         auxiliaryConnectionContexts.remove(connectionDistinguisher);
 204     }
 205
 206     @Override
 207     public DeviceState getDeviceState() {
 208         return deviceState;
 209     }
 210
 211     @Override
 212     public ReadOnlyTransaction getReadTransaction() {
 213         return dataBroker.newReadOnlyTransaction();
 214     }
 215
 216     @Override
 217     public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
 218                                                           final InstanceIdentifier<T> path,
 219                                                           final T data){
 220         if (initialized) {
 221             transactionChainManager.writeToTransaction(store, path, data, false);
 222         }
 223     }
 224
 225     @Override
 226     public <T extends DataObject> void writeToTransactionWithParentsSlow(final LogicalDatastoreType store,
 227                                                                          final InstanceIdentifier<T> path,
 228                                                                          final T data){
 229         if (initialized) {
 230             transactionChainManager.writeToTransaction(store, path, data, true);
 231         }
 232     }
 233
 234     @Override
 235     public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) throws TransactionChainClosedException {
 236         if (initialized) {
 237             transactionChainManager.addDeleteOperationTotTxChain(store, path);
 238         }
 239     }
 240
 241     @Override
 242     public boolean submitTransaction() {
 243         return initialized && transactionChainManager.submitWriteTransaction();
 244     }
 245
 246     @Override
 247     public ConnectionContext getPrimaryConnectionContext() {
 248         return primaryConnectionContext;
 249     }
 250
 251     @Override
 252     public ConnectionContext getAuxiliaryConnectionContexts(final BigInteger cookie) {
 253         return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue()));
 254     }
 255
 256     @Override
 257     public DeviceFlowRegistry getDeviceFlowRegistry() {
 258         return deviceFlowRegistry;
 259     }
 260
 261     @Override
 262     public DeviceGroupRegistry getDeviceGroupRegistry() {
 263         return deviceGroupRegistry;
 264     }
 265
 266     @Override
 267     public DeviceMeterRegistry getDeviceMeterRegistry() {
 268         return deviceMeterRegistry;
 269     }
 270
 271     @Override
 272     public void processReply(final OfHeader ofHeader) {
 273         if (ofHeader instanceof Error) {
 274             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 275         } else {
 276             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 277         }
 278     }
 279
 280     @Override
 281     public void processReply(final Xid xid, final List<MultipartReply> ofHeaderList) {
 282         for (final MultipartReply multipartReply : ofHeaderList) {
 283             messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 284         }
 285     }
 286
 287     @Override
 288     public void processFlowRemovedMessage(final FlowRemoved flowRemoved) {
 289         //1. translate to general flow (table, priority, match, cookie)
 290         final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification =
 291                 flowRemovedTranslator.translate(flowRemoved, deviceInfo, null);
 292
 293         if(!deviceManager.getIsNotificationFlowRemovedOff()) {
 294             // Trigger off a notification
 295             notificationPublishService.offerNotification(flowRemovedNotification);
 296         } else if(LOG.isDebugEnabled()) {
 297             LOG.debug("For nodeId={} isNotificationFlowRemovedOff={}", getDeviceInfo().getLOGValue(), deviceManager.getIsNotificationFlowRemovedOff());
 298         }
 299
 300         final ItemLifecycleListener itemLifecycleListener = flowLifeCycleKeeper.getItemLifecycleListener();
 301         if (itemLifecycleListener != null) {
 302             //2. create registry key
 303             final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(flowRemovedNotification);
 304             //3. lookup flowId
 305             final FlowDescriptor flowDescriptor = deviceFlowRegistry.retrieveIdForFlow(flowRegKey);
 306             //4. if flowId present:
 307             if (flowDescriptor != null) {
 308                 // a) construct flow path
 309                 final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = getDeviceInfo().getNodeInstanceIdentifier()
 310                         .augmentation(FlowCapableNode.class)
 311                         .child(Table.class, flowDescriptor.getTableKey())
 312                         .child(Flow.class, new FlowKey(flowDescriptor.getFlowId()));
 313                 // b) notify listener
 314                 itemLifecycleListener.onRemoved(flowPath);
 315             } else {
 316                 LOG.debug("flow id not found: nodeId={} tableId={}, priority={}",
 317                         getDeviceInfo().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority());
 318             }
 319         }
 320     }
 321
 322     @Override
 323     public void processPortStatusMessage(final PortStatusMessage portStatus) {
 324         messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 325         try {
 326             updatePortStatusMessage(portStatus);
 327         } catch (final Exception e) {
 328             LOG.warn("Error processing port status message for port {} on device {} : {}", portStatus.getPortNo().toString(),
 329                     getDeviceInfo().getNodeId().toString(), e);
 330             retryProcessPortStatusMessage(portStatus,RETRY_COUNT);
 331         }
 332     }
 333
 334     private void retryProcessPortStatusMessage(final PortStatusMessage portStatus, final int retryCount){
 335         Thread thread = new Thread(new PortStatusRetryRunnable(portStatus, retryCount));
 336         thread.start();
 337     }
 338
 339     private class PortStatusRetryRunnable implements Runnable {
 340         private final PortStatusMessage portStatusMessage;
 341         private final int retryCount;
 342         private int count ;
 343
 344         PortStatusRetryRunnable (final PortStatusMessage portStatus,final int retry){
 345             portStatusMessage = portStatus;
 346             retryCount = retry;
 347             count = 0;
 348         }
 349         @Override
 350         public void run() {
 351             while(count < retryCount) {
 352                 try {
 353                     TimeUnit.MILLISECONDS.sleep(RETRY_DELAY);
 354                 } catch (InterruptedException e) {
 355                     LOG.warn("Caught interrupted exception while sleeping during a port retry task");
 356                 }
 357
 358                 try {
 359                     updatePortStatusMessage(portStatusMessage);
 360                     break;
 361                 } catch (final Exception e) {
 362                     LOG.warn("Error processing port status message for {} on port {} on device {} : {}",count,
 363                             portStatusMessage.getPortNo().toString(),getDeviceInfo().getNodeId().toString(), e);
 364                     count ++ ;
 365                 }
 366             }
 367             LOG.warn("Failed to update port status for port {} on device {} even after 3 retries",
 368                     portStatusMessage.getPortNo().toString(),getDeviceInfo().getNodeId().toString());
 369         }
 370     }
 371
 372     void updatePortStatusMessage(final PortStatusMessage portStatusMessage){
 373         final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatusMessage,
 374                 getDeviceInfo(), null);
 375
 376         final KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> iiToNodeConnector =
 377                 provideIIToNodeConnector(portStatusMessage.getPortNo(), portStatusMessage.getVersion());
 378
 379         if (portStatusMessage.getReason().equals(PortReason.OFPPRADD) || portStatusMessage.getReason().
 380                 equals(PortReason.OFPPRMODIFY)) {
 381             // because of ADD status node connector has to be created
 382             final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey());
 383             nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class,
 384                     new FlowCapableNodeConnectorStatisticsDataBuilder().build());
 385             nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector);
 386             writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build());
 387         } else if (portStatusMessage.getReason().equals(PortReason.OFPPRDELETE)) {
 388             addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
 389         }
 390         submitTransaction();
 391     }
 392
 393     private KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> provideIIToNodeConnector(final long portNo, final short version) {
 394         final InstanceIdentifier<Node> iiToNodes = getDeviceInfo().getNodeInstanceIdentifier();
 395         final BigInteger dataPathId = getDeviceInfo().getDatapathId();
 396         final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), portNo, version);
 397         return iiToNodes.child(NodeConnector.class, new NodeConnectorKey(nodeConnectorId));
 398     }
 399
 400     @Override
 401     public void processPacketInMessage(final PacketInMessage packetInMessage) {
 402         messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH);
 403         final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter();
 404         final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, getDeviceInfo(), null);
 405
 406         if (packetReceived == null) {
 407             LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress());
 408             messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE);
 409             return;
 410         } else {
 411             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_OUT_SUCCESS);
 412         }
 413
 414         if (!packetInLimiter.acquirePermit()) {
 415             LOG.debug("Packet limited");
 416             // TODO: save packet into emergency slot if possible
 417             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PACKET_IN_LIMIT_REACHED_AND_DROPPED);
 418             return;
 419         }
 420
 421         final ListenableFuture<?> offerNotification = notificationPublishService.offerNotification(packetReceived);
 422         if (NotificationPublishService.REJECTED.equals(offerNotification)) {
 423             LOG.debug("notification offer rejected");
 424             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 425             packetInLimiter.drainLowWaterMark();
 426             packetInLimiter.releasePermit();
 427             return;
 428         }
 429
 430         Futures.addCallback(offerNotification, new FutureCallback<Object>() {
 431             @Override
 432             public void onSuccess(final Object result) {
 433                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 434                 packetInLimiter.releasePermit();
 435             }
 436
 437             @Override
 438             public void onFailure(final Throwable t) {
 439                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 440                 LOG.debug("notification offer failed: {}", t.getMessage());
 441                 LOG.trace("notification offer failed..", t);
 442                 packetInLimiter.releasePermit();
 443             }
 444         });
 445     }
 446
 447     @Override
 448     public void processExperimenterMessage(final ExperimenterMessage notification) {
 449         // lookup converter
 450         final ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice();
 451         final MessageTypeKey<? extends ExperimenterDataOfChoice> key = new MessageTypeKey<>(
 452                 getDeviceInfo().getVersion(),
 453                 (Class<? extends ExperimenterDataOfChoice>) vendorData.getImplementedInterface());
 454         final ConvertorMessageFromOFJava<ExperimenterDataOfChoice, MessagePath> messageConverter = extensionConverterProvider.getMessageConverter(key);
 455         if (messageConverter == null) {
 456             LOG.warn("custom converter for {}[OF:{}] not found",
 457                     notification.getExperimenterDataOfChoice().getImplementedInterface(),
 458                     getDeviceInfo().getVersion());
 459             return;
 460         }
 461         // build notification
 462         final ExperimenterMessageOfChoice messageOfChoice;
 463         try {
 464             messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION);
 465             final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new ExperimenterMessageFromDevBuilder()
 466                     .setNode(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier()))
 467                     .setExperimenterMessageOfChoice(messageOfChoice);
 468             // publish
 469             notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
 470         } catch (final ConversionException e) {
 471             LOG.error("Conversion of experimenter notification failed", e);
 472         }
 473     }
 474
 475     @Override
 476     public TranslatorLibrary oook() {
 477         return translatorLibrary;
 478     }
 479
 480     @Override
 481     public synchronized void close() {
 482         LOG.debug("closing deviceContext: {}, nodeId:{}",
 483                 getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress(),
 484                 getDeviceInfo().getLOGValue());
 485         // NOOP
 486         throw new UnsupportedOperationException("Autocloseble.close will be removed soon");
 487     }
 488
 489     @Override
 490     public void setCurrentBarrierTimeout(final Timeout timeout) {
 491         barrierTaskTimeout = timeout;
 492     }
 493
 494     @Override
 495     public Timeout getBarrierTaskTimeout() {
 496         return barrierTaskTimeout;
 497     }
 498
 499     @Override
 500     public void setNotificationPublishService(final NotificationPublishService notificationPublishService) {
 501         this.notificationPublishService = notificationPublishService;
 502     }
 503
 504     @Override
 505     public MessageSpy getMessageSpy() {
 506         return messageSpy;
 507     }
 508
 509     @Override
 510     public void onPublished() {
 511         Verify.verify(CONTEXT_STATE.INITIALIZATION.equals(getState()));
 512         setState(CONTEXT_STATE.WORKING);
 513         primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 514         for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) {
 515             switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 516         }
 517     }
 518
 519     @Override
 520     public MultiMsgCollector getMultiMsgCollector(final RequestContext<List<MultipartReply>> requestContext) {
 521         return new MultiMsgCollectorImpl(this, requestContext);
 522     }
 523
 524     @Override
 525     public void updatePacketInRateLimit(final long upperBound) {
 526         packetInLimiter.changeWaterMarks((int) (LOW_WATERMARK_FACTOR * upperBound), (int) (HIGH_WATERMARK_FACTOR * upperBound));
 527     }
 528
 529     @Override
 530     public ItemLifeCycleRegistry getItemLifeCycleSourceRegistry() {
 531         return itemLifeCycleSourceRegistry;
 532     }
 533
 534     @Override
 535     public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
 536         this.extensionConverterProvider = extensionConverterProvider;
 537     }
 538
 539     @Override
 540     public ExtensionConverterProvider getExtensionConverterProvider() {
 541         return extensionConverterProvider;
 542     }
 543
 544     @Override
 545     public synchronized void shutdownConnection() {
 546         if (LOG.isDebugEnabled()) {
 547             LOG.debug("Shutdown method for node {}", getDeviceInfo().getLOGValue());
 548         }
 549         if (CONTEXT_STATE.TERMINATION.equals(getState())) {
 550             LOG.debug("DeviceCtx for Node {} is in termination process.", getDeviceInfo().getLOGValue());
 551             return;
 552         }
 553
 554         if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
 555             LOG.debug("ConnectionCtx for Node {} is in RIP state.", getDeviceInfo().getLOGValue());
 556             return;
 557         }
 558
 559         // Terminate Auxiliary Connection
 560         for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
 561             LOG.debug("Closing auxiliary connection {}", connectionContext.getNodeId());
 562             connectionContext.closeConnection(false);
 563         }
 564
 565         // Terminate Primary Connection
 566         getPrimaryConnectionContext().closeConnection(true);
 567
 568         // Close all datastore registries
 569         if (initialized) {
 570             deviceGroupRegistry.close();
 571             deviceFlowRegistry.close();
 572             deviceMeterRegistry.close();
 573         }
 574     }
 575
 576     @Override
 577     public ListenableFuture<Void> shuttingDownDataStoreTransactions() {
 578         return initialized
 579                 ? this.transactionChainManager.shuttingDown()
 580                 : Futures.immediateFuture(null);
 581     }
 582
 583     @VisibleForTesting
 584     TransactionChainManager getTransactionChainManager() {
 585         return this.transactionChainManager;
 586     }
 587
 588     @Override
 589     public void setSwitchFeaturesMandatory(boolean switchFeaturesMandatory) {
 590         this.switchFeaturesMandatory = switchFeaturesMandatory;
 591     }
 592
 593     @Override
 594     public CONTEXT_STATE getState() {
 595         return this.state;
 596     }
 597
 598     @Override
 599     public void setState(CONTEXT_STATE state) {
 600         this.state = state;
 601     }
 602
 603     @Override
 604     public ListenableFuture<Void> stopClusterServices(boolean deviceDisconnected) {
 605         return initialized
 606                 ? this.transactionChainManager.deactivateTransactionManager()
 607                 : Futures.immediateFuture(null);
 608     }
 609
 610     @Override
 611     public ServiceGroupIdentifier getServiceIdentifier() {
 612         return this.deviceInfo.getServiceIdentifier();
 613     }
 614
 615     @Override
 616     public DeviceInfo getDeviceInfo() {
 617         return this.deviceInfo;
 618     }
 619
 620     @Override
 621     public void putLifecycleServiceIntoTxChainManager(final LifecycleService lifecycleService){
 622         if (initialized) {
 623             this.transactionChainManager.setLifecycleService(lifecycleService);
 624         }
 625     }
 626
 627     @Override
 628     public void replaceConnectionContext(final ConnectionContext connectionContext){
 629         // Act like we are initializing the context
 630         setState(CONTEXT_STATE.INITIALIZATION);
 631         this.primaryConnectionContext = connectionContext;
 632         this.onPublished();
 633     }
 634
 635     @Override
 636     public boolean isSkipTableFeatures() {
 637         return this.skipTableFeatures;
 638     }
 639
 640     @Override
 641     public void setLifecycleInitializationPhaseHandler(final ClusterInitializationPhaseHandler handler) {
 642         this.clusterInitializationPhaseHandler = handler;
 643     }
 644
 645     @Override
 646     public boolean onContextInstantiateService(final ConnectionContext connectionContext) {
 647
 648         if (getPrimaryConnectionContext().getConnectionState().equals(ConnectionContext.CONNECTION_STATE.RIP)) {
 649             LOG.warn("Connection on device {} was interrupted, will stop starting master services.", deviceInfo.getLOGValue());
 650             return false;
 651         }
 652
 653         LOG.info("Starting device context cluster services for node {}", deviceInfo.getLOGValue());
 654
 655         lazyTransactionManagerInitialization();
 656
 657         this.transactionChainManager.activateTransactionManager();
 658
 659         try {
 660             DeviceInitializationUtils.initializeNodeInformation(this, switchFeaturesMandatory, this.convertorExecutor);
 661         } catch (ExecutionException | InterruptedException e) {
 662             LOG.warn("Device {} cannot be initialized: ", deviceInfo.getLOGValue(), e);
 663             return false;
 664         }
 665
 666         return this.clusterInitializationPhaseHandler.onContextInstantiateService(getPrimaryConnectionContext());
 667     }
 668
 669     @VisibleForTesting
 670     void lazyTransactionManagerInitialization() {
 671         if (!this.initialized) {
 672             if (LOG.isDebugEnabled()) {
 673                 LOG.debug("Transaction chain manager for node {} created", deviceInfo.getLOGValue());
 674             }
 675             this.transactionChainManager = new TransactionChainManager(dataBroker, deviceInfo);
 676             this.deviceFlowRegistry = new DeviceFlowRegistryImpl(dataBroker, deviceInfo.getNodeInstanceIdentifier());
 677             this.deviceGroupRegistry = new DeviceGroupRegistryImpl();
 678             this.deviceMeterRegistry = new DeviceMeterRegistryImpl();
 679             this.initialized = true;
 680         }
 681     }
 682 }