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 javax.annotation.Nonnull;
  23 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  24 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  25 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  26 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  27 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainClosedException;
  28 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  29 import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
  30 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  31 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueue;
  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.connection.OutboundQueueProvider;
  35 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  36 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  37 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
  38 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
  39 import org.opendaylight.openflowplugin.api.openflow.device.MessageTranslator;
  40 import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
  41 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
  42 import org.opendaylight.openflowplugin.api.openflow.device.Xid;
  43 import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
  44 import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleConductor;
  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 /**
 105  *
 106  */
 107 public class DeviceContextImpl implements DeviceContext, ExtensionConverterProviderKeeper{
 108
 109     private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class);
 110
 111     // TODO: drain factor should be parametrized
 112     private static final float REJECTED_DRAIN_FACTOR = 0.25f;
 113     // TODO: low water mark factor should be parametrized
 114     private static final float LOW_WATERMARK_FACTOR = 0.75f;
 115     // TODO: high water mark factor should be parametrized
 116     private static final float HIGH_WATERMARK_FACTOR = 0.95f;
 117
 118     private final ConnectionContext primaryConnectionContext;
 119     private final DeviceState deviceState;
 120     private final DataBroker dataBroker;
 121     private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
 122     private final TransactionChainManager transactionChainManager;
 123     private final DeviceFlowRegistry deviceFlowRegistry;
 124     private final DeviceGroupRegistry deviceGroupRegistry;
 125     private final DeviceMeterRegistry deviceMeterRegistry;
 126     private final PacketInRateLimiter packetInLimiter;
 127     private final MessageSpy messageSpy;
 128     private final ItemLifeCycleKeeper flowLifeCycleKeeper;
 129     private NotificationPublishService notificationPublishService;
 130     private final OutboundQueue outboundQueueProvider;
 131     private Timeout barrierTaskTimeout;
 132     private final MessageTranslator<PortGrouping, FlowCapableNodeConnector> portStatusTranslator;
 133     private final MessageTranslator<PacketInMessage, PacketReceived> packetInTranslator;
 134     private final MessageTranslator<FlowRemoved, org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved> flowRemovedTranslator;
 135     private final TranslatorLibrary translatorLibrary;
 136     private final ItemLifeCycleRegistry itemLifeCycleSourceRegistry;
 137     private ExtensionConverterProvider extensionConverterProvider;
 138     private final DeviceManager deviceManager;
 139
 140     private boolean switchFeaturesMandatory;
 141     private final DeviceInfo deviceInfo;
 142     private final ConvertorExecutor convertorExecutor;
 143     private volatile CONTEXT_STATE state;
 144
 145     public DeviceContextImpl(@Nonnull final ConnectionContext primaryConnectionContext,
 146                              @Nonnull final DeviceState deviceState,
 147                              @Nonnull final DataBroker dataBroker,
 148                              @Nonnull final LifecycleConductor conductor,
 149                              @Nonnull final OutboundQueueProvider outboundQueueProvider,
 150                              @Nonnull final TranslatorLibrary translatorLibrary,
 151                              final DeviceManager manager,
 152                              @Nonnull final DeviceInfo deviceInfo,
 153                              final ConvertorExecutor convertorExecutor) {
 154         this.primaryConnectionContext = Preconditions.checkNotNull(primaryConnectionContext);
 155         this.deviceState = Preconditions.checkNotNull(deviceState);
 156         this.dataBroker = Preconditions.checkNotNull(dataBroker);
 157         Preconditions.checkNotNull(conductor);
 158         this.outboundQueueProvider = Preconditions.checkNotNull(outboundQueueProvider);
 159         this.transactionChainManager = new TransactionChainManager(dataBroker, deviceInfo , conductor);
 160         auxiliaryConnectionContexts = new HashMap<>();
 161         deviceFlowRegistry = new DeviceFlowRegistryImpl(dataBroker, deviceInfo.getNodeInstanceIdentifier());
 162         deviceGroupRegistry = new DeviceGroupRegistryImpl();
 163         deviceMeterRegistry = new DeviceMeterRegistryImpl();
 164         messageSpy = conductor.getMessageIntelligenceAgency();
 165         this.deviceManager = Preconditions.checkNotNull(manager);
 166
 167         packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(),
 168                 /*initial*/ 1000, /*initial*/2000, messageSpy, REJECTED_DRAIN_FACTOR);
 169
 170         this.translatorLibrary = translatorLibrary;
 171         portStatusTranslator = translatorLibrary.lookupTranslator(
 172                 new TranslatorKey(deviceInfo.getVersion(), PortGrouping.class.getName()));
 173         packetInTranslator = translatorLibrary.lookupTranslator(
 174                 new TranslatorKey(deviceInfo.getVersion(), PacketIn.class.getName()));
 175         flowRemovedTranslator = translatorLibrary.lookupTranslator(
 176                 new TranslatorKey(deviceInfo.getVersion(), FlowRemoved.class.getName()));
 177
 178         itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl();
 179         flowLifeCycleKeeper = new ItemLifeCycleSourceImpl();
 180         itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper);
 181         this.deviceInfo = deviceInfo;
 182         this.state = CONTEXT_STATE.INITIALIZATION;
 183         this.convertorExecutor = convertorExecutor;
 184     }
 185
 186     /**
 187      * This method is called from {@link DeviceManagerImpl} only. So we could say "posthandshake process finish"
 188      * and we are able to set a scheduler for an automatic transaction submitting by time (0,5sec).
 189      */
 190     void initialSubmitTransaction() {
 191         transactionChainManager.initialSubmitWriteTransaction();
 192     }
 193
 194     @Override
 195     public Long reserveXidForDeviceMessage() {
 196         return outboundQueueProvider.reserveEntry();
 197     }
 198
 199     @Override
 200     public void addAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 201         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 202         auxiliaryConnectionContexts.put(connectionDistinguisher, connectionContext);
 203     }
 204
 205     private static SwitchConnectionDistinguisher createConnectionDistinguisher(final ConnectionContext connectionContext) {
 206         return new SwitchConnectionCookieOFImpl(connectionContext.getFeatures().getAuxiliaryId());
 207     }
 208
 209     @Override
 210     public void removeAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 211         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 212         LOG.debug("auxiliary connection dropped: {}, nodeId:{}", connectionContext.getConnectionAdapter()
 213                 .getRemoteAddress(), getDeviceInfo().getNodeId());
 214         auxiliaryConnectionContexts.remove(connectionDistinguisher);
 215     }
 216
 217     @Override
 218     public DeviceState getDeviceState() {
 219         return deviceState;
 220     }
 221
 222     @Override
 223     public ReadOnlyTransaction getReadTransaction() {
 224         return dataBroker.newReadOnlyTransaction();
 225     }
 226
 227     @Override
 228     public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
 229                                                           final InstanceIdentifier<T> path,
 230                                                           final T data){
 231         transactionChainManager.writeToTransaction(store, path, data, false);
 232     }
 233
 234     @Override
 235     public <T extends DataObject> void writeToTransactionWithParentsSlow(final LogicalDatastoreType store,
 236                                                                          final InstanceIdentifier<T> path,
 237                                                                          final T data){
 238         transactionChainManager.writeToTransaction(store, path, data, true);
 239     }
 240
 241     @Override
 242     public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) throws TransactionChainClosedException {
 243         transactionChainManager.addDeleteOperationTotTxChain(store, path);
 244     }
 245
 246     @Override
 247     public boolean submitTransaction() {
 248         return transactionChainManager.submitWriteTransaction();
 249     }
 250
 251     @Override
 252     public ConnectionContext getPrimaryConnectionContext() {
 253         return primaryConnectionContext;
 254     }
 255
 256     @Override
 257     public ConnectionContext getAuxiliaryConnectiobContexts(final BigInteger cookie) {
 258         return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue()));
 259     }
 260
 261     @Override
 262     public DeviceFlowRegistry getDeviceFlowRegistry() {
 263         return deviceFlowRegistry;
 264     }
 265
 266     @Override
 267     public DeviceGroupRegistry getDeviceGroupRegistry() {
 268         return deviceGroupRegistry;
 269     }
 270
 271     @Override
 272     public DeviceMeterRegistry getDeviceMeterRegistry() {
 273         return deviceMeterRegistry;
 274     }
 275
 276     @Override
 277     public void processReply(final OfHeader ofHeader) {
 278         if (ofHeader instanceof Error) {
 279             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 280         } else {
 281             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 282         }
 283     }
 284
 285     @Override
 286     public void processReply(final Xid xid, final List<MultipartReply> ofHeaderList) {
 287         for (final MultipartReply multipartReply : ofHeaderList) {
 288             messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 289         }
 290     }
 291
 292     @Override
 293     public void processFlowRemovedMessage(final FlowRemoved flowRemoved) {
 294         //1. translate to general flow (table, priority, match, cookie)
 295         final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification =
 296                 flowRemovedTranslator.translate(flowRemoved, deviceInfo, null);
 297
 298         if(!deviceManager.getIsNotificationFlowRemovedOff()) {
 299             // Trigger off a notification
 300             notificationPublishService.offerNotification(flowRemovedNotification);
 301         } else if(LOG.isDebugEnabled()) {
 302             LOG.debug("For nodeId={} isNotificationFlowRemovedOff={}", getDeviceInfo().getNodeId(), deviceManager.getIsNotificationFlowRemovedOff());
 303         }
 304
 305         final ItemLifecycleListener itemLifecycleListener = flowLifeCycleKeeper.getItemLifecycleListener();
 306         if (itemLifecycleListener != null) {
 307             //2. create registry key
 308             final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(flowRemovedNotification);
 309             //3. lookup flowId
 310             final FlowDescriptor flowDescriptor = deviceFlowRegistry.retrieveIdForFlow(flowRegKey);
 311             //4. if flowId present:
 312             if (flowDescriptor != null) {
 313                 // a) construct flow path
 314                 final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = getDeviceInfo().getNodeInstanceIdentifier()
 315                         .augmentation(FlowCapableNode.class)
 316                         .child(Table.class, flowDescriptor.getTableKey())
 317                         .child(Flow.class, new FlowKey(flowDescriptor.getFlowId()));
 318                 // b) notify listener
 319                 itemLifecycleListener.onRemoved(flowPath);
 320             } else {
 321                 LOG.debug("flow id not found: nodeId={} tableId={}, priority={}",
 322                         getDeviceInfo().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority());
 323             }
 324         }
 325     }
 326
 327     @Override
 328     public void processPortStatusMessage(final PortStatusMessage portStatus) {
 329         messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 330         final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatus, getDeviceInfo(), null);
 331
 332         final KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> iiToNodeConnector = provideIIToNodeConnector(portStatus.getPortNo(), portStatus.getVersion());
 333         try {
 334             if (portStatus.getReason().equals(PortReason.OFPPRADD) || portStatus.getReason().equals(PortReason.OFPPRMODIFY)) {
 335                 // because of ADD status node connector has to be created
 336                 final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey());
 337                 nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build());
 338                 nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector);
 339                 writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build());
 340             } else if (portStatus.getReason().equals(PortReason.OFPPRDELETE)) {
 341                 addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
 342             }
 343             submitTransaction();
 344         } catch (final Exception e) {
 345             LOG.warn("Error processing port status message: ", e);
 346         }
 347     }
 348
 349     private KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> provideIIToNodeConnector(final long portNo, final short version) {
 350         final InstanceIdentifier<Node> iiToNodes = getDeviceInfo().getNodeInstanceIdentifier();
 351         final BigInteger dataPathId = getDeviceInfo().getDatapathId();
 352         final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), portNo, version);
 353         return iiToNodes.child(NodeConnector.class, new NodeConnectorKey(nodeConnectorId));
 354     }
 355
 356     @Override
 357     public void processPacketInMessage(final PacketInMessage packetInMessage) {
 358         messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH);
 359         final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter();
 360         final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, getDeviceInfo(), null);
 361
 362         if (packetReceived == null) {
 363             LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress());
 364             messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE);
 365             return;
 366         } else {
 367             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_OUT_SUCCESS);
 368         }
 369
 370         if (!packetInLimiter.acquirePermit()) {
 371             LOG.debug("Packet limited");
 372             // TODO: save packet into emergency slot if possible
 373             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PACKET_IN_LIMIT_REACHED_AND_DROPPED);
 374             return;
 375         }
 376
 377         final ListenableFuture<?> offerNotification = notificationPublishService.offerNotification(packetReceived);
 378         if (NotificationPublishService.REJECTED.equals(offerNotification)) {
 379             LOG.debug("notification offer rejected");
 380             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 381             packetInLimiter.drainLowWaterMark();
 382             packetInLimiter.releasePermit();
 383             return;
 384         }
 385
 386         Futures.addCallback(offerNotification, new FutureCallback<Object>() {
 387             @Override
 388             public void onSuccess(final Object result) {
 389                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 390                 packetInLimiter.releasePermit();
 391             }
 392
 393             @Override
 394             public void onFailure(final Throwable t) {
 395                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 396                 LOG.debug("notification offer failed: {}", t.getMessage());
 397                 LOG.trace("notification offer failed..", t);
 398                 packetInLimiter.releasePermit();
 399             }
 400         });
 401     }
 402
 403     @Override
 404     public void processExperimenterMessage(final ExperimenterMessage notification) {
 405         // lookup converter
 406         final ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice();
 407         final MessageTypeKey<? extends ExperimenterDataOfChoice> key = new MessageTypeKey<>(
 408                 getDeviceInfo().getVersion(),
 409                 (Class<? extends ExperimenterDataOfChoice>) vendorData.getImplementedInterface());
 410         final ConvertorMessageFromOFJava<ExperimenterDataOfChoice, MessagePath> messageConverter = extensionConverterProvider.getMessageConverter(key);
 411         if (messageConverter == null) {
 412             LOG.warn("custom converter for {}[OF:{}] not found",
 413                     notification.getExperimenterDataOfChoice().getImplementedInterface(),
 414                     getDeviceInfo().getVersion());
 415             return;
 416         }
 417         // build notification
 418         final ExperimenterMessageOfChoice messageOfChoice;
 419         try {
 420             messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION);
 421             final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new ExperimenterMessageFromDevBuilder()
 422                 .setNode(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier()))
 423                     .setExperimenterMessageOfChoice(messageOfChoice);
 424             // publish
 425             notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
 426         } catch (final ConversionException e) {
 427             LOG.error("Conversion of experimenter notification failed", e);
 428         }
 429     }
 430
 431     @Override
 432     public TranslatorLibrary oook() {
 433         return translatorLibrary;
 434     }
 435
 436     @Override
 437     public synchronized void close() {
 438         LOG.debug("closing deviceContext: {}, nodeId:{}",
 439                 getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress(),
 440                 getDeviceInfo().getNodeId());
 441         // NOOP
 442         throw new UnsupportedOperationException("Autocloseble.close will be removed soon");
 443     }
 444
 445     @Override
 446     public void setCurrentBarrierTimeout(final Timeout timeout) {
 447         barrierTaskTimeout = timeout;
 448     }
 449
 450     @Override
 451     public Timeout getBarrierTaskTimeout() {
 452         return barrierTaskTimeout;
 453     }
 454
 455     @Override
 456     public void setNotificationPublishService(final NotificationPublishService notificationPublishService) {
 457         this.notificationPublishService = notificationPublishService;
 458     }
 459
 460     @Override
 461     public MessageSpy getMessageSpy() {
 462         return messageSpy;
 463     }
 464
 465     @Override
 466     public void onPublished() {
 467         Verify.verify(CONTEXT_STATE.INITIALIZATION.equals(getState()));
 468         setState(CONTEXT_STATE.WORKING);
 469         primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 470         for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) {
 471             switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 472         }
 473     }
 474
 475     @Override
 476     public MultiMsgCollector getMultiMsgCollector(final RequestContext<List<MultipartReply>> requestContext) {
 477         return new MultiMsgCollectorImpl(this, requestContext);
 478     }
 479
 480     @Override
 481     public void updatePacketInRateLimit(final long upperBound) {
 482         packetInLimiter.changeWaterMarks((int) (LOW_WATERMARK_FACTOR * upperBound), (int) (HIGH_WATERMARK_FACTOR * upperBound));
 483     }
 484
 485     @Override
 486     public ItemLifeCycleRegistry getItemLifeCycleSourceRegistry() {
 487         return itemLifeCycleSourceRegistry;
 488     }
 489
 490     @Override
 491     public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
 492         this.extensionConverterProvider = extensionConverterProvider;
 493     }
 494
 495     @Override
 496     public ExtensionConverterProvider getExtensionConverterProvider() {
 497         return extensionConverterProvider;
 498     }
 499
 500     @Override
 501     public synchronized void shutdownConnection() {
 502         LOG.debug("Shutdown method for node {}", getDeviceInfo().getNodeId());
 503         if (CONTEXT_STATE.TERMINATION.equals(getState())) {
 504             LOG.debug("DeviceCtx for Node {} is in termination process.", getDeviceInfo().getNodeId());
 505             return;
 506         }
 507         setState(CONTEXT_STATE.TERMINATION);
 508
 509         if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
 510             LOG.debug("ConnectionCtx for Node {} is in RIP state.", getDeviceInfo().getNodeId());
 511             return;
 512         }
 513         /* Terminate Auxiliary Connection */
 514         for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
 515             LOG.debug("Closing auxiliary connection {}", connectionContext.getNodeId());
 516             connectionContext.closeConnection(false);
 517         }
 518         /* Terminate Primary Connection */
 519         getPrimaryConnectionContext().closeConnection(true);
 520         /* Close all Group Registry */
 521         deviceGroupRegistry.close();
 522         deviceFlowRegistry.close();
 523         deviceMeterRegistry.close();
 524     }
 525
 526     @Override
 527     public ListenableFuture<Void> shuttingDownDataStoreTransactions() {
 528         return transactionChainManager.shuttingDown();
 529     }
 530
 531     @VisibleForTesting
 532     TransactionChainManager getTransactionChainManager() {
 533         return this.transactionChainManager;
 534     }
 535
 536     @Override
 537     public void setSwitchFeaturesMandatory(boolean switchFeaturesMandatory) {
 538         this.switchFeaturesMandatory = switchFeaturesMandatory;
 539     }
 540
 541     @Override
 542     public void registerClusterSingletonServices(ClusterSingletonServiceProvider singletonServiceProvider) {
 543
 544     }
 545
 546     @Override
 547     public CONTEXT_STATE getState() {
 548         return this.state;
 549     }
 550
 551     @Override
 552     public void setState(CONTEXT_STATE state) {
 553         this.state = state;
 554     }
 555
 556     @Override
 557     public void startupClusterServices() throws ExecutionException, InterruptedException {
 558         LOG.debug("Initializing transaction chain manager for node {}", getDeviceInfo().getNodeId());
 559         this.transactionChainManager.activateTransactionManager();
 560         LOG.debug("Waiting to get node {} information", getDeviceInfo().getNodeId());
 561         DeviceInitializationUtils.initializeNodeInformation(this, switchFeaturesMandatory, this.convertorExecutor).get();
 562     }
 563
 564     @Override
 565     public ListenableFuture<Void> stopClusterServices() {
 566         return this.transactionChainManager.deactivateTransactionManager();
 567     }
 568
 569     @Override
 570     public ServiceGroupIdentifier getServiceIdentifier() {
 571         return this.deviceInfo.getServiceIdentifier();
 572     }
 573
 574     @Override
 575     public DeviceInfo getDeviceInfo() {
 576         return this.deviceInfo;
 577     }
 578 }