openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.java

   1 /**
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.openflowplugin.impl.device;
   9
  10 import com.google.common.annotations.VisibleForTesting;
  11 import com.google.common.base.Function;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.base.Verify;
  14 import com.google.common.util.concurrent.AsyncFunction;
  15 import com.google.common.util.concurrent.FutureCallback;
  16 import com.google.common.util.concurrent.Futures;
  17 import com.google.common.util.concurrent.ListenableFuture;
  18 import io.netty.util.Timeout;
  19 import java.math.BigInteger;
  20 import java.util.HashMap;
  21 import java.util.List;
  22 import java.util.Map;
  23 import java.util.concurrent.ConcurrentHashMap;
  24 import java.util.concurrent.atomic.AtomicLong;
  25 import javax.annotation.CheckForNull;
  26 import javax.annotation.Nonnull;
  27 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  28 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  29 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  30 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  31 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  32 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueue;
  33 import org.opendaylight.openflowjava.protocol.api.keys.MessageTypeKey;
  34 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
  35 import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
  36 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  37 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  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.RpcContext;
  55 import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener;
  56 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
  57 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
  58 import org.opendaylight.openflowplugin.extension.api.ConvertorMessageFromOFJava;
  59 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
  60 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
  61 import org.opendaylight.openflowplugin.extension.api.exception.ConversionException;
  62 import org.opendaylight.openflowplugin.extension.api.path.MessagePath;
  63 import org.opendaylight.openflowplugin.impl.common.ItemLifeCycleSourceImpl;
  64 import org.opendaylight.openflowplugin.impl.common.NodeStaticReplyTranslatorUtil;
  65 import org.opendaylight.openflowplugin.impl.device.listener.MultiMsgCollectorImpl;
  66 import org.opendaylight.openflowplugin.impl.registry.flow.DeviceFlowRegistryImpl;
  67 import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory;
  68 import org.opendaylight.openflowplugin.impl.registry.group.DeviceGroupRegistryImpl;
  69 import org.opendaylight.openflowplugin.impl.registry.meter.DeviceMeterRegistryImpl;
  70 import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtils;
  71 import org.opendaylight.openflowplugin.impl.util.MdSalRegistrationUtils;
  72 import org.opendaylight.openflowplugin.openflow.md.core.session.SwitchConnectionCookieOFImpl;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.experimenter.message.service.rev151020.ExperimenterMessageFromDevBuilder;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorRef;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorBuilder;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.PortReason;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.Error;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterMessage;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemoved;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketIn;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortGrouping;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage;
  96 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.experimenter.core.ExperimenterDataOfChoice;
  97 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.experimenter.types.rev151020.experimenter.core.message.ExperimenterMessageOfChoice;
  98 import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
  99 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
 100 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
 101 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole;
 102 import org.opendaylight.yangtools.yang.binding.DataObject;
 103 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 104 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
 105 import org.slf4j.Logger;
 106 import org.slf4j.LoggerFactory;
 107
 108 /**
 109  *
 110  */
 111 public class DeviceContextImpl implements DeviceContext, ExtensionConverterProviderKeeper {
 112
 113     private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class);
 114
 115     // TODO: drain factor should be parametrized
 116     private static final float REJECTED_DRAIN_FACTOR = 0.25f;
 117     // TODO: low water mark factor should be parametrized
 118     private static final float LOW_WATERMARK_FACTOR = 0.75f;
 119     // TODO: high water mark factor should be parametrized
 120     private static final float HIGH_WATERMARK_FACTOR = 0.95f;
 121
 122     private final ConnectionContext primaryConnectionContext;
 123     private final DeviceState deviceState;
 124     private final DataBroker dataBroker;
 125     private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
 126     private final TransactionChainManager transactionChainManager;
 127     private final DeviceFlowRegistry deviceFlowRegistry;
 128     private final DeviceGroupRegistry deviceGroupRegistry;
 129     private final DeviceMeterRegistry deviceMeterRegistry;
 130     private final PacketInRateLimiter packetInLimiter;
 131     private final MessageSpy messageSpy;
 132     private final ItemLifeCycleKeeper flowLifeCycleKeeper;
 133     private NotificationPublishService notificationPublishService;
 134     private final OutboundQueue outboundQueueProvider;
 135     private Timeout barrierTaskTimeout;
 136     private final MessageTranslator<PortGrouping, FlowCapableNodeConnector> portStatusTranslator;
 137     private final MessageTranslator<PacketInMessage, PacketReceived> packetInTranslator;
 138     private final MessageTranslator<FlowRemoved, org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved> flowRemovedTranslator;
 139     private final TranslatorLibrary translatorLibrary;
 140     private final Map<Long, NodeConnectorRef> nodeConnectorCache;
 141     private final ItemLifeCycleRegistry itemLifeCycleSourceRegistry;
 142     private RpcContext rpcContext;
 143     private ExtensionConverterProvider extensionConverterProvider;
 144
 145     private final boolean switchFeaturesMandatory;
 146     private StatisticsContext statisticsContext;
 147
 148     private final DeviceInfo deviceInfo;
 149
 150     private volatile DEVICE_CONTEXT_STATE deviceCtxState;
 151     private boolean isStatisticsRpcEnabled;
 152
 153
 154     @VisibleForTesting
 155     DeviceContextImpl(@Nonnull final ConnectionContext primaryConnectionContext,
 156                       @Nonnull final DeviceState deviceState,
 157                       @Nonnull final DataBroker dataBroker,
 158                       @Nonnull final LifecycleConductor conductor,
 159                       @Nonnull final OutboundQueueProvider outboundQueueProvider,
 160                       @Nonnull final TranslatorLibrary translatorLibrary,
 161                       final boolean switchFeaturesMandatory) {
 162         this.switchFeaturesMandatory = switchFeaturesMandatory;
 163         this.primaryConnectionContext = Preconditions.checkNotNull(primaryConnectionContext);
 164         this.deviceState = Preconditions.checkNotNull(deviceState);
 165         this.dataBroker = Preconditions.checkNotNull(dataBroker);
 166         Preconditions.checkNotNull(conductor);
 167         this.outboundQueueProvider = Preconditions.checkNotNull(outboundQueueProvider);
 168         deviceInfo = primaryConnectionContext.getDeviceInfo();
 169         this.transactionChainManager = new TransactionChainManager(dataBroker, deviceInfo, conductor);
 170         auxiliaryConnectionContexts = new HashMap<>();
 171         deviceFlowRegistry = new DeviceFlowRegistryImpl();
 172         deviceGroupRegistry = new DeviceGroupRegistryImpl();
 173         deviceMeterRegistry = new DeviceMeterRegistryImpl();
 174         messageSpy = conductor.getMessageIntelligenceAgency();
 175
 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(deviceInfo.getVersion(), PortGrouping.class.getName()));
 183         packetInTranslator = translatorLibrary.lookupTranslator(
 184                 new TranslatorKey(deviceInfo.getVersion(), PacketIn.class.getName()));
 185         flowRemovedTranslator = translatorLibrary.lookupTranslator(
 186                 new TranslatorKey(deviceInfo.getVersion(), FlowRemoved.class.getName()));
 187
 188
 189         nodeConnectorCache = new ConcurrentHashMap<>();
 190
 191         itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl();
 192         flowLifeCycleKeeper = new ItemLifeCycleSourceImpl();
 193         itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper);
 194         deviceCtxState = DEVICE_CONTEXT_STATE.INITIALIZATION;
 195     }
 196
 197     /**
 198      * This method is called from {@link DeviceManagerImpl} only. So we could say "posthandshake process finish"
 199      * and we are able to set a scheduler for an automatic transaction submitting by time (0,5sec).
 200      */
 201     void initialSubmitTransaction() {
 202         transactionChainManager.initialSubmitWriteTransaction();
 203     }
 204
 205     @Override
 206     public Long reserveXidForDeviceMessage() {
 207         return outboundQueueProvider.reserveEntry();
 208     }
 209
 210     @Override
 211     public void addAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 212         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 213         auxiliaryConnectionContexts.put(connectionDistinguisher, connectionContext);
 214     }
 215
 216     private static SwitchConnectionDistinguisher createConnectionDistinguisher(final ConnectionContext connectionContext) {
 217         return new SwitchConnectionCookieOFImpl(connectionContext.getFeatures().getAuxiliaryId());
 218     }
 219
 220     @Override
 221     public void removeAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 222         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 223         LOG.debug("auxiliary connection dropped: {}, nodeId:{}", connectionContext.getConnectionAdapter()
 224                 .getRemoteAddress(), deviceInfo.getNodeId());
 225         auxiliaryConnectionContexts.remove(connectionDistinguisher);
 226     }
 227
 228     @Override
 229     public DeviceState getDeviceState() {
 230         return deviceState;
 231     }
 232
 233     @Override
 234     public DeviceInfo getDeviceInfo() {
 235         return this.deviceInfo;
 236     }
 237
 238     @Override
 239     public ReadOnlyTransaction getReadTransaction() {
 240         return dataBroker.newReadOnlyTransaction();
 241     }
 242
 243     @Override
 244     public ListenableFuture<Void> onClusterRoleChange(final OfpRole oldRole, @CheckForNull final OfpRole role) {
 245         LOG.trace("onClusterRoleChange {} for node:", role, deviceInfo.getNodeId());
 246         Preconditions.checkArgument(role != null);
 247         if (role.equals(oldRole)) {
 248             LOG.debug("Demanded role change for device {} is not changed. OldRole: {}, NewRole {}", deviceInfo.getNodeId(), oldRole, role);
 249             return Futures.immediateFuture(null);
 250         }
 251         if (OfpRole.BECOMEMASTER.equals(role)) {
 252             return onDeviceTakeClusterLeadership();
 253         } else if (OfpRole.BECOMESLAVE.equals(role)) {
 254             return onDeviceLostClusterLeadership();
 255         } else {
 256             LOG.warn("Unknown OFCluster Role {} for Node {}", role, deviceInfo.getNodeId());
 257             if (null != rpcContext) {
 258                 MdSalRegistrationUtils.unregisterServices(rpcContext);
 259             }
 260             return transactionChainManager.deactivateTransactionManager();
 261         }
 262     }
 263
 264     @Override
 265     public ListenableFuture<Void> onDeviceLostClusterLeadership() {
 266         LOG.trace("onDeviceLostClusterLeadership for node: {}", deviceInfo.getNodeId());
 267         if (null != rpcContext) {
 268             MdSalRegistrationUtils.registerSlaveServices(rpcContext, OfpRole.BECOMESLAVE);
 269         }
 270         return transactionChainManager.deactivateTransactionManager();
 271     }
 272
 273     @Override
 274     public ListenableFuture<Void> onDeviceTakeClusterLeadership() {
 275         LOG.trace("onDeviceTakeClusterLeadership for node: {}", deviceInfo.getNodeId());
 276         /* validation */
 277         if (statisticsContext == null) {
 278             final String errMsg = String.format("DeviceCtx %s is up but we are missing StatisticsContext", deviceInfo.getDatapathId());
 279             LOG.warn(errMsg);
 280             return Futures.immediateFailedFuture(new IllegalStateException(errMsg));
 281         }
 282         if (rpcContext == null) {
 283             final String errMsg = String.format("DeviceCtx %s is up but we are missing RpcContext", deviceInfo.getDatapathId());
 284             LOG.warn(errMsg);
 285             return Futures.immediateFailedFuture(new IllegalStateException(errMsg));
 286         }
 287         /* Routed RPC registration */
 288         MdSalRegistrationUtils.registerMasterServices(getRpcContext(), DeviceContextImpl.this, OfpRole.BECOMEMASTER);
 289
 290         if (isStatisticsRpcEnabled) {
 291             MdSalRegistrationUtils.registerStatCompatibilityServices(getRpcContext(), this,
 292                     notificationPublishService, new AtomicLong());
 293         }
 294
 295         /* Prepare init info collecting */
 296         getDeviceState().setDeviceSynchronized(false);
 297         transactionChainManager.activateTransactionManager();
 298         /* Init Collecting NodeInfo */
 299         final ListenableFuture<Void> initCollectingDeviceInfo = DeviceInitializationUtils.initializeNodeInformation(
 300                 DeviceContextImpl.this, switchFeaturesMandatory);
 301         /* Init Collecting StatInfo */
 302         final ListenableFuture<Boolean> statPollFuture = Futures.transform(initCollectingDeviceInfo,
 303                 new AsyncFunction<Void, Boolean>() {
 304
 305                     @Override
 306                     public ListenableFuture<Boolean> apply(@Nonnull final Void input) throws Exception {
 307                         getStatisticsContext().statListForCollectingInitialization();
 308                         return getStatisticsContext().gatherDynamicData();
 309                     }
 310                 });
 311
 312         return Futures.transform(statPollFuture, new Function<Boolean, Void>() {
 313
 314             @Override
 315             public Void apply(final Boolean input) {
 316                 if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
 317                     final String errMsg = String.format("We lost connection for Device %s, context has to be closed.",
 318                             getDeviceInfo().getNodeId());
 319                     LOG.warn(errMsg);
 320                     throw new IllegalStateException(errMsg);
 321                 }
 322                 if (!input) {
 323                     final String errMsg = String.format("Get Initial Device %s information fails",
 324                             getDeviceInfo().getNodeId());
 325                     LOG.warn(errMsg);
 326                     throw new IllegalStateException(errMsg);
 327                 }
 328                 LOG.debug("Get Initial Device {} information is successful", deviceInfo.getNodeId());
 329                 getDeviceState().setDeviceSynchronized(true);
 330                 initialSubmitTransaction();
 331                 getDeviceState().setStatisticsPollingEnabledProp(true);
 332                 return null;
 333             }
 334         });
 335     }
 336
 337     @Override
 338     public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
 339                                                           final InstanceIdentifier<T> path, final T data) throws Exception {
 340         transactionChainManager.writeToTransaction(store, path, data, false);
 341     }
 342
 343     @Override
 344     public <T extends DataObject> void writeToTransactionWithParentsSlow(LogicalDatastoreType store, InstanceIdentifier<T> path, T data) throws Exception {
 345         transactionChainManager.writeToTransaction(store, path, data, true);
 346     }
 347
 348     @Override
 349     public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) throws Exception {
 350         transactionChainManager.addDeleteOperationTotTxChain(store, path);
 351     }
 352
 353     @Override
 354     public boolean submitTransaction() {
 355         return transactionChainManager.submitWriteTransaction();
 356     }
 357
 358     @Override
 359     public ConnectionContext getPrimaryConnectionContext() {
 360         return primaryConnectionContext;
 361     }
 362
 363     @Override
 364     public ConnectionContext getAuxiliaryConnectiobContexts(final BigInteger cookie) {
 365         return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue()));
 366     }
 367
 368     @Override
 369     public DeviceFlowRegistry getDeviceFlowRegistry() {
 370         return deviceFlowRegistry;
 371     }
 372
 373     @Override
 374     public DeviceGroupRegistry getDeviceGroupRegistry() {
 375         return deviceGroupRegistry;
 376     }
 377
 378     @Override
 379     public DeviceMeterRegistry getDeviceMeterRegistry() {
 380         return deviceMeterRegistry;
 381     }
 382
 383     @Override
 384     public void processReply(final OfHeader ofHeader) {
 385         if (ofHeader instanceof Error) {
 386             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 387         } else {
 388             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 389         }
 390     }
 391
 392     @Override
 393     public void processReply(final Xid xid, final List<MultipartReply> ofHeaderList) {
 394         for (final MultipartReply multipartReply : ofHeaderList) {
 395             messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 396         }
 397     }
 398
 399     @Override
 400     public void processFlowRemovedMessage(final FlowRemoved flowRemoved) {
 401         final ItemLifecycleListener itemLifecycleListener = flowLifeCycleKeeper.getItemLifecycleListener();
 402         if (itemLifecycleListener != null) {
 403             //1. translate to general flow (table, priority, match, cookie)
 404             final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification =
 405                     flowRemovedTranslator.translate(flowRemoved, deviceInfo, null);
 406             //2. create registry key
 407             final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(flowRemovedNotification);
 408             //3. lookup flowId
 409             final FlowDescriptor flowDescriptor = deviceFlowRegistry.retrieveIdForFlow(flowRegKey);
 410             //4. if flowId present:
 411             if (flowDescriptor != null) {
 412                 // a) construct flow path
 413                 final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = getDeviceInfo().getNodeInstanceIdentifier()
 414                         .augmentation(FlowCapableNode.class)
 415                         .child(Table.class, flowDescriptor.getTableKey())
 416                         .child(Flow.class, new FlowKey(flowDescriptor.getFlowId()));
 417                 // b) notify listener
 418                 itemLifecycleListener.onRemoved(flowPath);
 419                 // c) trigger off a notification
 420                 notificationPublishService.offerNotification(flowRemovedNotification);
 421             } else {
 422                 LOG.debug("flow id not found: nodeId={} tableId={}, priority={}",
 423                         getDeviceInfo().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority());
 424             }
 425         }
 426     }
 427
 428     @Override
 429     public void processPortStatusMessage(final PortStatusMessage portStatus) {
 430         messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 431         final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatus, deviceInfo, null);
 432
 433         final KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> iiToNodeConnector = provideIIToNodeConnector(portStatus.getPortNo(), portStatus.getVersion());
 434         try {
 435             if (portStatus.getReason().equals(PortReason.OFPPRADD) || portStatus.getReason().equals(PortReason.OFPPRMODIFY)) {
 436                 // because of ADD status node connector has to be created
 437                 final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey());
 438                 nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build());
 439                 nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector);
 440                 writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build());
 441             } else if (portStatus.getReason().equals(PortReason.OFPPRDELETE)) {
 442                 addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
 443             }
 444             submitTransaction();
 445         } catch (final Exception e) {
 446             LOG.warn("Error processing port status message: {}", e.getMessage());
 447         }
 448     }
 449
 450     private KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> provideIIToNodeConnector(final long portNo, final short version) {
 451         final InstanceIdentifier<Node> iiToNodes = deviceInfo.getNodeInstanceIdentifier();
 452         final BigInteger dataPathId = deviceInfo.getDatapathId();
 453         final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), portNo, version);
 454         return iiToNodes.child(NodeConnector.class, new NodeConnectorKey(nodeConnectorId));
 455     }
 456
 457     @Override
 458     public void processPacketInMessage(final PacketInMessage packetInMessage) {
 459         messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH);
 460         final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter();
 461         final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, deviceInfo, null);
 462
 463         if (packetReceived == null) {
 464             LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress());
 465             messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE);
 466             return;
 467         } else {
 468             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_OUT_SUCCESS);
 469         }
 470
 471         if (!packetInLimiter.acquirePermit()) {
 472             LOG.debug("Packet limited");
 473             // TODO: save packet into emergency slot if possible
 474             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PACKET_IN_LIMIT_REACHED_AND_DROPPED);
 475             return;
 476         }
 477
 478         final ListenableFuture<?> offerNotification = notificationPublishService.offerNotification(packetReceived);
 479         if (NotificationPublishService.REJECTED.equals(offerNotification)) {
 480             LOG.debug("notification offer rejected");
 481             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 482             packetInLimiter.drainLowWaterMark();
 483             packetInLimiter.releasePermit();
 484             return;
 485         }
 486
 487         Futures.addCallback(offerNotification, new FutureCallback<Object>() {
 488             @Override
 489             public void onSuccess(final Object result) {
 490                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 491                 packetInLimiter.releasePermit();
 492             }
 493
 494             @Override
 495             public void onFailure(final Throwable t) {
 496                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 497                 LOG.debug("notification offer failed: {}", t.getMessage());
 498                 LOG.trace("notification offer failed..", t);
 499                 packetInLimiter.releasePermit();
 500             }
 501         });
 502     }
 503
 504     @Override
 505     public void processExperimenterMessage(final ExperimenterMessage notification) {
 506         // lookup converter
 507         final ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice();
 508         final MessageTypeKey<? extends ExperimenterDataOfChoice> key = new MessageTypeKey<>(
 509                 deviceInfo.getVersion(),
 510                 (Class<? extends ExperimenterDataOfChoice>) vendorData.getImplementedInterface());
 511         final ConvertorMessageFromOFJava<ExperimenterDataOfChoice, MessagePath> messageConverter = extensionConverterProvider.getMessageConverter(key);
 512         if (messageConverter == null) {
 513             LOG.warn("custom converter for {}[OF:{}] not found",
 514                     notification.getExperimenterDataOfChoice().getImplementedInterface(),
 515                     deviceInfo.getVersion());
 516             return;
 517         }
 518         // build notification
 519         final ExperimenterMessageOfChoice messageOfChoice;
 520         try {
 521             messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION);
 522             final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new ExperimenterMessageFromDevBuilder()
 523                 .setNode(new NodeRef(deviceInfo.getNodeInstanceIdentifier()))
 524                     .setExperimenterMessageOfChoice(messageOfChoice);
 525             // publish
 526             notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
 527         } catch (final ConversionException e) {
 528             LOG.error("Conversion of experimenter notification failed", e);
 529         }
 530     }
 531
 532     @Override
 533     public TranslatorLibrary oook() {
 534         return translatorLibrary;
 535     }
 536
 537     @Override
 538     public synchronized void close() {
 539         LOG.debug("closing deviceContext: {}, nodeId:{}",
 540                 getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress(),
 541                 getDeviceInfo().getNodeId());
 542         // NOOP
 543         throw new UnsupportedOperationException("Autocloseble.close will be removed soon");
 544     }
 545
 546     @Override
 547     public void setCurrentBarrierTimeout(final Timeout timeout) {
 548         barrierTaskTimeout = timeout;
 549     }
 550
 551     @Override
 552     public Timeout getBarrierTaskTimeout() {
 553         return barrierTaskTimeout;
 554     }
 555
 556     @Override
 557     public void setNotificationPublishService(final NotificationPublishService notificationPublishService) {
 558         this.notificationPublishService = notificationPublishService;
 559     }
 560
 561     @Override
 562     public MessageSpy getMessageSpy() {
 563         return messageSpy;
 564     }
 565
 566     @Override
 567     public void onPublished() {
 568         Verify.verify(DEVICE_CONTEXT_STATE.INITIALIZATION.equals(deviceCtxState));
 569         deviceCtxState = DEVICE_CONTEXT_STATE.WORKING;
 570         primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 571         for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) {
 572             switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 573         }
 574     }
 575
 576     @Override
 577     public MultiMsgCollector getMultiMsgCollector(final RequestContext<List<MultipartReply>> requestContext) {
 578         return new MultiMsgCollectorImpl(this, requestContext);
 579     }
 580
 581     @Override
 582     public NodeConnectorRef lookupNodeConnectorRef(final Long portNumber) {
 583         return nodeConnectorCache.get(portNumber);
 584     }
 585
 586     @Override
 587     public void storeNodeConnectorRef(@Nonnull final Long portNumber, @Nonnull final NodeConnectorRef nodeConnectorRef) {
 588         nodeConnectorCache.put(
 589                 Preconditions.checkNotNull(portNumber),
 590                 Preconditions.checkNotNull(nodeConnectorRef));
 591     }
 592
 593     @Override
 594     public void updatePacketInRateLimit(final long upperBound) {
 595         packetInLimiter.changeWaterMarks((int) (LOW_WATERMARK_FACTOR * upperBound), (int) (HIGH_WATERMARK_FACTOR * upperBound));
 596     }
 597
 598     @Override
 599     public ItemLifeCycleRegistry getItemLifeCycleSourceRegistry() {
 600         return itemLifeCycleSourceRegistry;
 601     }
 602
 603     @Override
 604     public void setRpcContext(final RpcContext rpcContext) {
 605         this.rpcContext = rpcContext;
 606     }
 607
 608     @Override
 609     public RpcContext getRpcContext() {
 610         return rpcContext;
 611     }
 612
 613     @Override
 614     public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
 615         this.extensionConverterProvider = extensionConverterProvider;
 616     }
 617
 618     @Override
 619     public ExtensionConverterProvider getExtensionConverterProvider() {
 620         return extensionConverterProvider;
 621     }
 622
 623     @Override
 624     public void setStatisticsContext(final StatisticsContext statisticsContext) {
 625         this.statisticsContext = statisticsContext;
 626     }
 627
 628     @Override
 629     public StatisticsContext getStatisticsContext() {
 630         return statisticsContext;
 631     }
 632
 633     @Override
 634     public synchronized void shutdownConnection() {
 635         LOG.debug("Shutdown method for node {}", deviceInfo.getNodeId());
 636         deviceState.setValid(false);
 637         if (DEVICE_CONTEXT_STATE.TERMINATION.equals(deviceCtxState)) {
 638             LOG.debug("DeviceCtx for Node {} is in termination process.", deviceInfo.getNodeId());
 639             return;
 640         }
 641         deviceCtxState = DEVICE_CONTEXT_STATE.TERMINATION;
 642
 643         if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
 644             LOG.debug("ConnectionCtx for Node {} is in RIP state.", deviceInfo.getNodeId());
 645             return;
 646         }
 647         /* Terminate Auxiliary Connection */
 648         for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
 649             LOG.debug("Closing auxiliary connection {}", connectionContext.getNodeId());
 650             connectionContext.closeConnection(false);
 651         }
 652         /* Terminate Primary Connection */
 653         getPrimaryConnectionContext().closeConnection(true);
 654         /* Close all Group Registry */
 655         deviceGroupRegistry.close();
 656         deviceFlowRegistry.close();
 657         deviceMeterRegistry.close();
 658     }
 659
 660     @Override
 661     public void setStatisticsRpcEnabled(boolean isStatisticsRpcEnabled) {
 662         this.isStatisticsRpcEnabled = isStatisticsRpcEnabled;
 663     }
 664
 665     @Override
 666     public DEVICE_CONTEXT_STATE getDeviceContextState() {
 667         return deviceCtxState;
 668     }
 669
 670     @Override
 671     public ListenableFuture<Void> shuttingDownDataStoreTransactions() {
 672         deviceState.setValid(false);
 673         return transactionChainManager.shuttingDown();
 674     }
 675
 676     @VisibleForTesting
 677     TransactionChainManager getTransactionChainManager() {
 678         return this.transactionChainManager;
 679     }
 680 }