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 javax.annotation.CheckForNull;
  11 import javax.annotation.Nonnull;
  12 import com.google.common.annotations.VisibleForTesting;
  13 import com.google.common.base.Preconditions;
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import com.google.common.util.concurrent.Futures;
  16 import com.google.common.util.concurrent.ListenableFuture;
  17 import io.netty.util.HashedWheelTimer;
  18 import io.netty.util.Timeout;
  19 import java.math.BigInteger;
  20 import java.util.Collection;
  21 import java.util.Collections;
  22 import java.util.HashMap;
  23 import java.util.HashSet;
  24 import java.util.LinkedList;
  25 import java.util.List;
  26 import java.util.Map;
  27 import java.util.concurrent.ConcurrentHashMap;
  28
  29 import com.google.common.annotations.VisibleForTesting;
  30 import com.google.common.base.Preconditions;
  31 import com.google.common.util.concurrent.CheckedFuture;
  32 import com.google.common.util.concurrent.FutureCallback;
  33 import com.google.common.util.concurrent.Futures;
  34 import com.google.common.util.concurrent.ListenableFuture;
  35 import io.netty.util.HashedWheelTimer;
  36 import io.netty.util.Timeout;
  37 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  38 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  39 import org.opendaylight.controller.md.sal.binding.api.NotificationService;
  40 import org.opendaylight.controller.md.sal.binding.api.ReadTransaction;
  41 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  42 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  43 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueue;
  44 import org.opendaylight.openflowjava.protocol.api.keys.MessageTypeKey;
  45 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
  46 import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
  47 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  48 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
  49 import org.opendaylight.openflowplugin.api.openflow.device.MessageTranslator;
  50 import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
  51 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
  52 import org.opendaylight.openflowplugin.api.openflow.device.Xid;
  53 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceContextClosedHandler;
  54 import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
  55 import org.opendaylight.openflowplugin.api.openflow.md.core.SwitchConnectionDistinguisher;
  56 import org.opendaylight.openflowplugin.api.openflow.md.core.TranslatorKey;
  57 import org.opendaylight.openflowplugin.api.openflow.registry.ItemLifeCycleRegistry;
  58 import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
  59 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor;
  60 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowRegistryKey;
  61 import org.opendaylight.openflowplugin.api.openflow.registry.group.DeviceGroupRegistry;
  62 import org.opendaylight.openflowplugin.api.openflow.registry.meter.DeviceMeterRegistry;
  63 import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleKeeper;
  64 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
  65 import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener;
  66 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
  67 import org.opendaylight.openflowplugin.extension.api.ConvertorMessageFromOFJava;
  68 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
  69 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
  70 import org.opendaylight.openflowplugin.extension.api.exception.ConversionException;
  71 import org.opendaylight.openflowplugin.extension.api.path.MessagePath;
  72 import org.opendaylight.openflowplugin.impl.common.ItemLifeCycleSourceImpl;
  73 import org.opendaylight.openflowplugin.impl.common.NodeStaticReplyTranslatorUtil;
  74 import org.opendaylight.openflowplugin.impl.device.listener.MultiMsgCollectorImpl;
  75 import org.opendaylight.openflowplugin.impl.registry.flow.DeviceFlowRegistryImpl;
  76 import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory;
  77 import org.opendaylight.openflowplugin.impl.registry.group.DeviceGroupRegistryImpl;
  78 import org.opendaylight.openflowplugin.impl.registry.meter.DeviceMeterRegistryImpl;
  79 import org.opendaylight.openflowplugin.openflow.md.core.session.SwitchConnectionCookieOFImpl;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.experimenter.message.service.rev151020.ExperimenterMessageFromDevBuilder;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorRef;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorBuilder;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.PortReason;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.Error;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterMessage;
  96 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemoved;
  97 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply;
  98 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  99 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketIn;
 100 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage;
 101 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortGrouping;
 102 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage;
 103 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.experimenter.core.ExperimenterDataOfChoice;
 104 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.experimenter.types.rev151020.experimenter.core.message.ExperimenterMessageOfChoice;
 105 import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
 106 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
 107 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
 108 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole;
 109 import org.opendaylight.yangtools.yang.binding.DataObject;
 110 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 111 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
 112 import org.slf4j.Logger;
 113 import org.slf4j.LoggerFactory;
 114
 115 /**
 116  *
 117  */
 118 public class DeviceContextImpl implements DeviceContext, ExtensionConverterProviderKeeper {
 119
 120     private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class);
 121
 122     // TODO: drain factor should be parametrized
 123     public static final float REJECTED_DRAIN_FACTOR = 0.25f;
 124     // TODO: low water mark factor should be parametrized
 125     private static final float LOW_WATERMARK_FACTOR = 0.75f;
 126     // TODO: high water mark factor should be parametrized
 127     private static final float HIGH_WATERMARK_FACTOR = 0.95f;
 128
 129     private final ConnectionContext primaryConnectionContext;
 130     private final DeviceState deviceState;
 131     private final DataBroker dataBroker;
 132     private final HashedWheelTimer hashedWheelTimer;
 133     private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
 134     private final TransactionChainManager transactionChainManager;
 135     private final DeviceFlowRegistry deviceFlowRegistry;
 136     private final DeviceGroupRegistry deviceGroupRegistry;
 137     private final DeviceMeterRegistry deviceMeterRegistry;
 138     private final Collection<DeviceContextClosedHandler> closeHandlers = new HashSet<>();
 139     private final PacketInRateLimiter packetInLimiter;
 140     private final MessageSpy messageSpy;
 141     private final ItemLifeCycleKeeper flowLifeCycleKeeper;
 142     private NotificationPublishService notificationPublishService;
 143     private NotificationService notificationService;
 144     private final OutboundQueue outboundQueueProvider;
 145     private Timeout barrierTaskTimeout;
 146     private final MessageTranslator<PortGrouping, FlowCapableNodeConnector> portStatusTranslator;
 147     private final MessageTranslator<PacketInMessage, PacketReceived> packetInTranslator;
 148     private final MessageTranslator<FlowRemoved, org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved> flowRemovedTranslator;
 149     private final TranslatorLibrary translatorLibrary;
 150     private final Map<Long, NodeConnectorRef> nodeConnectorCache;
 151     private ItemLifeCycleRegistry itemLifeCycleSourceRegistry;
 152     private RpcContext rpcContext;
 153     private ExtensionConverterProvider extensionConverterProvider;
 154
 155
 156     @VisibleForTesting
 157     DeviceContextImpl(@Nonnull final ConnectionContext primaryConnectionContext,
 158                       @Nonnull final DeviceState deviceState,
 159                       @Nonnull final DataBroker dataBroker,
 160                       @Nonnull final HashedWheelTimer hashedWheelTimer,
 161                       @Nonnull final MessageSpy _messageSpy,
 162                       @Nonnull final OutboundQueueProvider outboundQueueProvider,
 163             @Nonnull final TranslatorLibrary translatorLibrary) {
 164         this.primaryConnectionContext = Preconditions.checkNotNull(primaryConnectionContext);
 165         this.deviceState = Preconditions.checkNotNull(deviceState);
 166         this.dataBroker = Preconditions.checkNotNull(dataBroker);
 167         this.hashedWheelTimer = Preconditions.checkNotNull(hashedWheelTimer);
 168         this.outboundQueueProvider = Preconditions.checkNotNull(outboundQueueProvider);
 169         primaryConnectionContext.setDeviceDisconnectedHandler(DeviceContextImpl.this);
 170         this.transactionChainManager = new TransactionChainManager(dataBroker, deviceState);
 171         auxiliaryConnectionContexts = new HashMap<>();
 172         deviceFlowRegistry = new DeviceFlowRegistryImpl();
 173         deviceGroupRegistry = new DeviceGroupRegistryImpl();
 174         deviceMeterRegistry = new DeviceMeterRegistryImpl();
 175         messageSpy = _messageSpy;
 176
 177         packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(),
 178                 /*initial*/ 1000, /*initial*/2000, messageSpy, REJECTED_DRAIN_FACTOR);
 179
 180         this.translatorLibrary = translatorLibrary;
 181         portStatusTranslator = translatorLibrary.lookupTranslator(
 182                 new TranslatorKey(deviceState.getVersion(), PortGrouping.class.getName()));
 183         packetInTranslator = translatorLibrary.lookupTranslator(
 184                 new TranslatorKey(deviceState.getVersion(), PacketIn.class.getName()));
 185         flowRemovedTranslator = translatorLibrary.lookupTranslator(
 186                 new TranslatorKey(deviceState.getVersion(), FlowRemoved.class.getName()));
 187
 188
 189         nodeConnectorCache = new ConcurrentHashMap<>();
 190
 191         itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl();
 192         flowLifeCycleKeeper = new ItemLifeCycleSourceImpl();
 193         itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper);
 194     }
 195
 196     /**
 197      * This method is called from {@link DeviceManagerImpl} only. So we could say "posthandshake process finish"
 198      * and we are able to set a scheduler for an automatic transaction submitting by time (0,5sec).
 199      */
 200     void initialSubmitTransaction() {
 201         transactionChainManager.initialSubmitWriteTransaction();
 202     }
 203
 204     @Override
 205     public Long getReservedXid() {
 206         return outboundQueueProvider.reserveEntry();
 207     }
 208
 209     @Override
 210     public void addAuxiliaryConenctionContext(final ConnectionContext connectionContext) {
 211         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 212         auxiliaryConnectionContexts.put(connectionDistinguisher, connectionContext);
 213     }
 214
 215     private static SwitchConnectionDistinguisher createConnectionDistinguisher(final ConnectionContext connectionContext) {
 216         return new SwitchConnectionCookieOFImpl(connectionContext.getFeatures().getAuxiliaryId());
 217     }
 218
 219     @Override
 220     public void removeAuxiliaryConenctionContext(final ConnectionContext connectionContext) {
 221         // TODO Auto-generated method stub
 222     }
 223
 224     @Override
 225     public DeviceState getDeviceState() {
 226         return deviceState;
 227     }
 228
 229     @Override
 230     public ReadTransaction getReadTransaction() {
 231         return dataBroker.newReadOnlyTransaction();
 232     }
 233
 234     @Override
 235     public void onClusterRoleChange(@CheckForNull final OfpRole role) {
 236         LOG.debug("onClusterRoleChange {} for node:", role, deviceState.getNodeId());
 237         Preconditions.checkArgument(role != null);
 238         if (OfpRole.BECOMEMASTER.equals(role)) {
 239             transactionChainManager.activateTransactionManager();
 240         } else if (OfpRole.BECOMESLAVE.equals(role)) {
 241             transactionChainManager.deactivateTransactionManager();
 242         } else {
 243             LOG.warn("Unknow OFCluster Role {} for Node {}", role, deviceState.getNodeId());
 244         }
 245     }
 246
 247     @Override
 248     public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
 249                                                           final InstanceIdentifier<T> path, final T data) {
 250         transactionChainManager.writeToTransaction(store, path, data);
 251     }
 252
 253     @Override
 254     public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) {
 255         transactionChainManager.addDeleteOperationTotTxChain(store, path);
 256     }
 257
 258     @Override
 259     public boolean submitTransaction() {
 260         return transactionChainManager.submitWriteTransaction();
 261     }
 262
 263     @Override
 264     public ConnectionContext getPrimaryConnectionContext() {
 265         return primaryConnectionContext;
 266     }
 267
 268     @Override
 269     public ConnectionContext getAuxiliaryConnectiobContexts(final BigInteger cookie) {
 270         return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue()));
 271     }
 272
 273     @Override
 274     public DeviceFlowRegistry getDeviceFlowRegistry() {
 275         return deviceFlowRegistry;
 276     }
 277
 278     @Override
 279     public DeviceGroupRegistry getDeviceGroupRegistry() {
 280         return deviceGroupRegistry;
 281     }
 282
 283     @Override
 284     public DeviceMeterRegistry getDeviceMeterRegistry() {
 285         return deviceMeterRegistry;
 286     }
 287
 288     @Override
 289     public void processReply(final OfHeader ofHeader) {
 290         if (ofHeader instanceof Error) {
 291             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 292         } else {
 293             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 294         }
 295     }
 296
 297     @Override
 298     public void processReply(final Xid xid, final List<MultipartReply> ofHeaderList) {
 299         for (final MultipartReply multipartReply : ofHeaderList) {
 300             messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 301         }
 302     }
 303
 304     @Override
 305     public void processFlowRemovedMessage(final FlowRemoved flowRemoved) {
 306         final ItemLifecycleListener itemLifecycleListener = flowLifeCycleKeeper.getItemLifecycleListener();
 307         if (itemLifecycleListener != null) {
 308             //1. translate to general flow (table, priority, match, cookie)
 309             final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification =
 310                     flowRemovedTranslator.translate(flowRemoved, this, null);
 311             //2. create registry key
 312             FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(flowRemovedNotification);
 313             //3. lookup flowId
 314             final FlowDescriptor flowDescriptor = deviceFlowRegistry.retrieveIdForFlow(flowRegKey);
 315             //4. if flowId present:
 316             if (flowDescriptor != null) {
 317                 // a) construct flow path
 318                 KeyedInstanceIdentifier<Flow, FlowKey> flowPath = getDeviceState().getNodeInstanceIdentifier()
 319                         .augmentation(FlowCapableNode.class)
 320                         .child(Table.class, flowDescriptor.getTableKey())
 321                         .child(Flow.class, new FlowKey(flowDescriptor.getFlowId()));
 322                 // b) notify listener
 323                 itemLifecycleListener.onRemoved(flowPath);
 324             } else {
 325                 LOG.debug("flow id not found: nodeId={} tableId={}, priority={}",
 326                         getDeviceState().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority());
 327             }
 328         }
 329     }
 330
 331     @Override
 332     public void processPortStatusMessage(final PortStatusMessage portStatus) {
 333         messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 334         final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatus, this, null);
 335
 336         final KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> iiToNodeConnector = provideIIToNodeConnector(portStatus.getPortNo(), portStatus.getVersion());
 337         if (portStatus.getReason().equals(PortReason.OFPPRADD) || portStatus.getReason().equals(PortReason.OFPPRMODIFY)) {
 338             // because of ADD status node connector has to be created
 339             final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey());
 340             nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build());
 341             nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector);
 342             writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build());
 343         } else if (portStatus.getReason().equals(PortReason.OFPPRDELETE)) {
 344             addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
 345         }
 346         submitTransaction();
 347     }
 348
 349     private KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> provideIIToNodeConnector(final long portNo, final short version) {
 350         final InstanceIdentifier<Node> iiToNodes = deviceState.getNodeInstanceIdentifier();
 351         final BigInteger dataPathId = deviceState.getFeatures().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, this, null);
 361
 362         if (packetReceived == null) {
 363             LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress());
 364             messageSpy.spyMessage(packetReceived.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<? extends Object> 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(ExperimenterMessage notification) {
 405         // lookup converter
 406         ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice();
 407         MessageTypeKey<? extends ExperimenterDataOfChoice> key = new MessageTypeKey<>(
 408                 deviceState.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                     deviceState.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(deviceState.getNodeInstanceIdentifier()))
 423                     .setExperimenterMessageOfChoice(messageOfChoice);
 424             // publish
 425             notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
 426         } catch (ConversionException e) {
 427             LOG.warn("Conversion of experimenter notification failed", e);
 428         }
 429     }
 430
 431     @Override
 432     public TranslatorLibrary oook() {
 433         return translatorLibrary;
 434     }
 435
 436     @Override
 437     public HashedWheelTimer getTimer() {
 438         return hashedWheelTimer;
 439     }
 440
 441     @Override
 442     public void close() {
 443         LOG.debug("closing deviceContext: {}, nodeId:{}",
 444                 getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress(),
 445                 getDeviceState().getNodeId());
 446
 447         tearDown();
 448
 449         primaryConnectionContext.closeConnection(false);
 450     }
 451
 452     private synchronized void tearDown() {
 453         if (deviceState.isValid()) {
 454             deviceState.setValid(false);
 455
 456             for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
 457                 connectionContext.closeConnection(false);
 458             }
 459
 460             LOG.info("Closing transaction chain manager without cleaning inventory operational");
 461             transactionChainManager.close();
 462
 463             deviceGroupRegistry.close();
 464             deviceFlowRegistry.close();
 465             deviceMeterRegistry.close();
 466
 467             final LinkedList<DeviceContextClosedHandler> reversedCloseHandlers = new LinkedList<>(closeHandlers);
 468             Collections.reverse(reversedCloseHandlers);
 469             for (final DeviceContextClosedHandler deviceContextClosedHandler : reversedCloseHandlers) {
 470                 deviceContextClosedHandler.onDeviceContextClosed(this);
 471             }
 472             closeHandlers.clear();
 473         }
 474     }
 475
 476     @Override
 477     public void onDeviceDisconnected(final ConnectionContext connectionContext) {
 478         if (getPrimaryConnectionContext().equals(connectionContext)) {
 479             try {
 480                 tearDown();
 481             } catch (final Exception e) {
 482                 LOG.trace("Error closing device context.");
 483             }
 484         } else {
 485             LOG.debug("auxiliary connection dropped: {}, nodeId:{}",
 486                     connectionContext.getConnectionAdapter().getRemoteAddress(),
 487                     getDeviceState().getNodeId());
 488             final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 489             auxiliaryConnectionContexts.remove(connectionDistinguisher);
 490         }
 491     }
 492
 493     @Override
 494     public void setCurrentBarrierTimeout(final Timeout timeout) {
 495         barrierTaskTimeout = timeout;
 496     }
 497
 498     @Override
 499     public Timeout getBarrierTaskTimeout() {
 500         return barrierTaskTimeout;
 501     }
 502
 503     @Override
 504     public void setNotificationService(final NotificationService notificationServiceParam) {
 505         notificationService = notificationServiceParam;
 506     }
 507
 508     @Override
 509     public void setNotificationPublishService(final NotificationPublishService notificationPublishService) {
 510         this.notificationPublishService = notificationPublishService;
 511     }
 512
 513     @Override
 514     public MessageSpy getMessageSpy() {
 515         return messageSpy;
 516     }
 517
 518     @Override
 519     public void addDeviceContextClosedHandler(final DeviceContextClosedHandler deviceContextClosedHandler) {
 520         closeHandlers.add(deviceContextClosedHandler);
 521     }
 522
 523     @Override
 524     public void onPublished() {
 525         primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 526         for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) {
 527             switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 528         }
 529     }
 530
 531     @Override
 532     public MultiMsgCollector getMultiMsgCollector(final RequestContext<List<MultipartReply>> requestContext) {
 533         return new MultiMsgCollectorImpl(this, requestContext);
 534     }
 535
 536     @Override
 537     public NodeConnectorRef lookupNodeConnectorRef(Long portNumber) {
 538         return nodeConnectorCache.get(portNumber);
 539     }
 540
 541     @Override
 542     public void storeNodeConnectorRef(final Long portNumber, final NodeConnectorRef nodeConnectorRef) {
 543         nodeConnectorCache.put(
 544                 Preconditions.checkNotNull(portNumber),
 545                 Preconditions.checkNotNull(nodeConnectorRef));
 546     }
 547
 548     @Override
 549     public void updatePacketInRateLimit(long upperBound) {
 550         packetInLimiter.changeWaterMarks((int) (LOW_WATERMARK_FACTOR * upperBound), (int) (HIGH_WATERMARK_FACTOR * upperBound));
 551     }
 552
 553     @Override
 554     public ItemLifeCycleRegistry getItemLifeCycleSourceRegistry() {
 555         return itemLifeCycleSourceRegistry;
 556     }
 557
 558     @Override
 559     public void setRpcContext(RpcContext rpcContext) {
 560         this.rpcContext = rpcContext;
 561     }
 562
 563     @Override
 564     public RpcContext getRpcContext() {
 565         return rpcContext;
 566     }
 567
 568     @Override
 569     public void setExtensionConverterProvider(ExtensionConverterProvider extensionConverterProvider) {
 570         this.extensionConverterProvider = extensionConverterProvider;
 571     }
 572
 573     @Override
 574     public ExtensionConverterProvider getExtensionConverterProvider() {
 575         return extensionConverterProvider;
 576     }
 577 }