openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceManagerImpl.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.base.Preconditions;
  11 import com.google.common.util.concurrent.AsyncFunction;
  12 import com.google.common.util.concurrent.FutureCallback;
  13 import com.google.common.util.concurrent.Futures;
  14 import com.google.common.util.concurrent.ListenableFuture;
  15 import io.netty.util.HashedWheelTimer;
  16 import java.math.BigInteger;
  17 import java.util.ArrayList;
  18 import java.util.Arrays;
  19 import java.util.Collection;
  20 import java.util.Collections;
  21 import java.util.Iterator;
  22 import java.util.List;
  23 import java.util.concurrent.ExecutionException;
  24 import java.util.concurrent.ScheduledThreadPoolExecutor;
  25 import java.util.concurrent.TimeUnit;
  26 import javax.annotation.CheckForNull;
  27 import javax.annotation.Nonnull;
  28 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  29 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  30 import org.opendaylight.controller.md.sal.binding.api.NotificationService;
  31 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  32 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  33 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  34 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueue;
  35 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandlerRegistration;
  36 import org.opendaylight.openflowplugin.api.ConnectionException;
  37 import org.opendaylight.openflowplugin.api.OFConstants;
  38 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
  39 import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
  40 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  41 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
  42 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
  43 import org.opendaylight.openflowplugin.api.openflow.device.MessageTranslator;
  44 import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
  45 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
  46 import org.opendaylight.openflowplugin.api.openflow.device.Xid;
  47 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceInitializationPhaseHandler;
  48 import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
  49 import org.opendaylight.openflowplugin.api.openflow.md.core.TranslatorKey;
  50 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageIntelligenceAgency;
  51 import org.opendaylight.openflowplugin.impl.common.MultipartRequestInputFactory;
  52 import org.opendaylight.openflowplugin.impl.common.NodeStaticReplyTranslatorUtil;
  53 import org.opendaylight.openflowplugin.impl.connection.OutboundQueueProviderImpl;
  54 import org.opendaylight.openflowplugin.impl.device.listener.OpenflowProtocolListenerFullImpl;
  55 import org.opendaylight.openflowplugin.impl.rpc.AbstractRequestContext;
  56 import org.opendaylight.openflowplugin.impl.util.DeviceStateUtil;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeBuilder;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableBuilder;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.FlowTableStatisticsData;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.FlowTableStatisticsDataBuilder;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupFeatures;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodesBuilder;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorBuilder;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeBuilder;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterFeatures;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.Capabilities;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.CapabilitiesV10;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.MultipartType;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortGrouping;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.MultipartReplyBody;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.MultipartReplyDescCase;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.MultipartReplyGroupFeaturesCase;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.MultipartReplyMeterFeaturesCase;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.MultipartReplyPortDescCase;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.MultipartReplyTableFeaturesCase;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.multipart.reply.desc._case.MultipartReplyDesc;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.multipart.reply.group.features._case.MultipartReplyGroupFeatures;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.multipart.reply.meter.features._case.MultipartReplyMeterFeatures;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.multipart.reply.port.desc._case.MultipartReplyPortDesc;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.multipart.reply.table.features._case.MultipartReplyTableFeatures;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.table.features.TableFeatures;
  94 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  95 import org.opendaylight.yangtools.yang.common.RpcError;
  96 import org.opendaylight.yangtools.yang.common.RpcResult;
  97 import org.slf4j.Logger;
  98 import org.slf4j.LoggerFactory;
  99
 100 /**
 101  *
 102  */
 103 public class DeviceManagerImpl implements DeviceManager, AutoCloseable {
 104
 105     private static final Logger LOG = LoggerFactory.getLogger(DeviceManagerImpl.class);
 106
 107     private static final long TICK_DURATION = 10; // 0.5 sec.
 108     private ScheduledThreadPoolExecutor spyPool;
 109     private final int spyRate = 10;
 110
 111     private final DataBroker dataBroker;
 112     private final HashedWheelTimer hashedWheelTimer;
 113     private TranslatorLibrary translatorLibrary;
 114     private DeviceInitializationPhaseHandler deviceInitPhaseHandler;
 115     private NotificationService notificationService;
 116     private NotificationPublishService notificationPublishService;
 117
 118     private final List<DeviceContext> deviceContexts = new ArrayList<DeviceContext>();
 119     private final MessageIntelligenceAgency messageIntelligenceAgency;
 120
 121     private final long barrierNanos = TimeUnit.MILLISECONDS.toNanos(500);
 122     private final int maxQueueDepth = 25600;
 123
 124     public DeviceManagerImpl(@Nonnull final DataBroker dataBroker,
 125                              @Nonnull final MessageIntelligenceAgency messageIntelligenceAgency) {
 126         this.dataBroker = Preconditions.checkNotNull(dataBroker);
 127         hashedWheelTimer = new HashedWheelTimer(TICK_DURATION, TimeUnit.MILLISECONDS, 500);
 128         /* merge empty nodes to oper DS to predict any problems with missing parent for Node */
 129         final WriteTransaction tx = dataBroker.newWriteOnlyTransaction();
 130
 131         final NodesBuilder nodesBuilder = new NodesBuilder();
 132         nodesBuilder.setNode(Collections.<Node>emptyList());
 133         tx.merge(LogicalDatastoreType.OPERATIONAL, InstanceIdentifier.create(Nodes.class), nodesBuilder.build());
 134         try {
 135             tx.submit().get();
 136         } catch (ExecutionException | InterruptedException e) {
 137             LOG.error("Creation of node failed.", e);
 138             throw new IllegalStateException(e);
 139         }
 140
 141         this.messageIntelligenceAgency = messageIntelligenceAgency;
 142     }
 143
 144     @Override
 145     public void setDeviceInitializationPhaseHandler(final DeviceInitializationPhaseHandler handler) {
 146         deviceInitPhaseHandler = handler;
 147     }
 148
 149     @Override
 150     public void onDeviceContextLevelUp(final DeviceContext deviceContext) {
 151         // final phase - we have to add new Device to MD-SAL DataStore
 152         Preconditions.checkNotNull(deviceContext);
 153         ((DeviceContextImpl) deviceContext).initialSubmitTransaction();
 154
 155         deviceContext.onPublished();
 156     }
 157
 158     @Override
 159     public void deviceConnected(@CheckForNull final ConnectionContext connectionContext) {
 160         Preconditions.checkArgument(connectionContext != null);
 161
 162         // Cache this for clarity
 163         final ConnectionAdapter connectionAdapter = connectionContext.getConnectionAdapter();
 164
 165         //FIXME: as soon as auxiliary connection are fully supported then this is needed only before device context published
 166         connectionAdapter.setPacketInFiltering(true);
 167
 168         final Short version = connectionContext.getFeatures().getVersion();
 169         final OutboundQueueProvider outboundQueueProvider = new OutboundQueueProviderImpl(version);
 170
 171         connectionContext.setOutboundQueueProvider(outboundQueueProvider);
 172         final OutboundQueueHandlerRegistration<OutboundQueueProvider> outboundQueueHandlerRegistration =
 173                 connectionAdapter.registerOutboundQueueHandler(outboundQueueProvider, maxQueueDepth, barrierNanos);
 174         connectionContext.setOutboundQueueHandleRegistration(outboundQueueHandlerRegistration);
 175
 176         final DeviceState deviceState = new DeviceStateImpl(connectionContext.getFeatures(), connectionContext.getNodeId());
 177
 178         final DeviceContext deviceContext = new DeviceContextImpl(connectionContext, deviceState, dataBroker,
 179                 hashedWheelTimer, messageIntelligenceAgency, outboundQueueProvider);
 180         deviceContext.setNotificationService(notificationService);
 181         deviceContext.setNotificationPublishService(notificationPublishService);
 182         final NodeBuilder nodeBuilder = new NodeBuilder().setId(deviceState.getNodeId()).setNodeConnector(Collections.<NodeConnector>emptyList());
 183         deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, deviceState.getNodeInstanceIdentifier(), nodeBuilder.build());
 184
 185         connectionContext.setDeviceDisconnectedHandler(deviceContext);
 186         deviceContext.setTranslatorLibrary(translatorLibrary);
 187         deviceContext.addDeviceContextClosedHandler(this);
 188
 189         final OpenflowProtocolListenerFullImpl messageListener = new OpenflowProtocolListenerFullImpl(
 190                 connectionAdapter, deviceContext);
 191         connectionAdapter.setMessageListener(messageListener);
 192
 193         final ListenableFuture<List<RpcResult<List<MultipartReply>>>> deviceFeaturesFuture;
 194         if (OFConstants.OFP_VERSION_1_0 == version) {
 195             final CapabilitiesV10 capabilitiesV10 = connectionContext.getFeatures().getCapabilitiesV10();
 196
 197             DeviceStateUtil.setDeviceStateBasedOnV10Capabilities(deviceState, capabilitiesV10);
 198
 199             deviceFeaturesFuture = createDeviceFeaturesForOF10(deviceContext, deviceState);
 200             // create empty tables after device description is processed
 201             chainTableTrunkWriteOF10(deviceContext, deviceFeaturesFuture);
 202
 203             final short ofVersion = deviceContext.getDeviceState().getVersion();
 204             final TranslatorKey translatorKey = new TranslatorKey(ofVersion, PortGrouping.class.getName());
 205             final MessageTranslator<PortGrouping, FlowCapableNodeConnector> translator = deviceContext.oook().lookupTranslator(translatorKey);
 206             final BigInteger dataPathId = deviceContext.getPrimaryConnectionContext().getFeatures().getDatapathId();
 207
 208             for (final PortGrouping port : connectionContext.getFeatures().getPhyPort()) {
 209                 final FlowCapableNodeConnector fcNodeConnector = translator.translate(port, deviceContext, null);
 210
 211                 final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), port.getPortNo(), ofVersion);
 212                 final NodeConnectorBuilder ncBuilder = new NodeConnectorBuilder().setId(nodeConnectorId);
 213                 ncBuilder.addAugmentation(FlowCapableNodeConnector.class, fcNodeConnector);
 214                 ncBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build());
 215                 final NodeConnector connector = ncBuilder.build();
 216                 final InstanceIdentifier<NodeConnector> connectorII = deviceState.getNodeInstanceIdentifier().child(NodeConnector.class, connector.getKey());
 217                 deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, connectorII, connector);
 218             }
 219         } else if (OFConstants.OFP_VERSION_1_3 == version) {
 220             final Capabilities capabilities = connectionContext.getFeatures().getCapabilities();
 221             LOG.debug("Setting capabilities for device {}", deviceContext.getDeviceState().getNodeId());
 222             DeviceStateUtil.setDeviceStateBasedOnV13Capabilities(deviceState, capabilities);
 223             deviceFeaturesFuture = createDeviceFeaturesForOF13(deviceContext, deviceState);
 224         } else {
 225             deviceFeaturesFuture = Futures.immediateFailedFuture(new ConnectionException("Unsupported version " + version));
 226         }
 227
 228         Futures.addCallback(deviceFeaturesFuture, new FutureCallback<List<RpcResult<List<MultipartReply>>>>() {
 229             @Override
 230             public void onSuccess(final List<RpcResult<List<MultipartReply>>> result) {
 231                 deviceContext.getDeviceState().setValid(true);
 232                 deviceInitPhaseHandler.onDeviceContextLevelUp(deviceContext);
 233                 LOG.trace("Device context level up called.");
 234             }
 235
 236             @Override
 237             public void onFailure(final Throwable t) {
 238                 // FIXME : remove session
 239                 LOG.trace("Device capabilities gathering future failed.");
 240                 LOG.trace("more info in exploration failure..", t);
 241             }
 242         });
 243     }
 244
 245     private static void chainTableTrunkWriteOF10(final DeviceContext deviceContext, final ListenableFuture<List<RpcResult<List<MultipartReply>>>> deviceFeaturesFuture) {
 246         Futures.addCallback(deviceFeaturesFuture, new FutureCallback<List<RpcResult<List<MultipartReply>>>>() {
 247             @Override
 248             public void onSuccess(final List<RpcResult<List<MultipartReply>>> results) {
 249                 boolean allSucceeded = true;
 250                 for (final RpcResult<List<MultipartReply>> rpcResult : results) {
 251                     allSucceeded &= rpcResult.isSuccessful();
 252                 }
 253                 if (allSucceeded) {
 254                     createEmptyFlowCapableNodeInDs(deviceContext);
 255                     makeEmptyTables(deviceContext, deviceContext.getDeviceState().getNodeInstanceIdentifier(),
 256                             deviceContext.getDeviceState().getFeatures().getTables());
 257                 }
 258             }
 259
 260             @Override
 261             public void onFailure(final Throwable t) {
 262                 //NOOP
 263             }
 264         });
 265     }
 266
 267
 268     private static ListenableFuture<List<RpcResult<List<MultipartReply>>>> createDeviceFeaturesForOF10(final DeviceContext deviceContext,
 269                                                                                                        final DeviceState deviceState) {
 270         final ListenableFuture<RpcResult<List<MultipartReply>>> replyDesc = getNodeStaticInfo(MultipartType.OFPMPDESC,
 271                 deviceContext,
 272                 deviceState.getNodeInstanceIdentifier(),
 273                 deviceState.getVersion());
 274
 275         return Futures.allAsList(Arrays.asList(replyDesc));
 276     }
 277
 278     private static ListenableFuture<List<RpcResult<List<MultipartReply>>>> createDeviceFeaturesForOF13(final DeviceContext deviceContext,
 279                                                                                                        final DeviceState deviceState) {
 280
 281         final ListenableFuture<RpcResult<List<MultipartReply>>> replyDesc = getNodeStaticInfo(MultipartType.OFPMPDESC,
 282                 deviceContext,
 283                 deviceState.getNodeInstanceIdentifier(),
 284                 deviceState.getVersion());
 285
 286         //first process description reply, write data to DS and write consequent data if successful
 287         return Futures.transform(replyDesc, new AsyncFunction<RpcResult<List<MultipartReply>>, List<RpcResult<List<MultipartReply>>>>() {
 288             @Override
 289             public ListenableFuture<List<RpcResult<List<MultipartReply>>>> apply(final RpcResult<List<MultipartReply>> rpcResult) throws Exception {
 290
 291                 translateAndWriteReply(MultipartType.OFPMPDESC, deviceContext, deviceState.getNodeInstanceIdentifier(), rpcResult.getResult());
 292
 293                 final ListenableFuture<RpcResult<List<MultipartReply>>> replyMeterFeature = getNodeStaticInfo(MultipartType.OFPMPMETERFEATURES,
 294                         deviceContext,
 295                         deviceState.getNodeInstanceIdentifier(),
 296                         deviceState.getVersion());
 297
 298                 createSuccessProcessingCallback(MultipartType.OFPMPMETERFEATURES,
 299                         deviceContext,
 300                         deviceState.getNodeInstanceIdentifier(),
 301                         replyMeterFeature);
 302
 303                 final ListenableFuture<RpcResult<List<MultipartReply>>> replyGroupFeatures = getNodeStaticInfo(MultipartType.OFPMPGROUPFEATURES,
 304                         deviceContext,
 305                         deviceState.getNodeInstanceIdentifier(),
 306                         deviceState.getVersion());
 307                 createSuccessProcessingCallback(MultipartType.OFPMPGROUPFEATURES,
 308                         deviceContext,
 309                         deviceState.getNodeInstanceIdentifier(),
 310                         replyGroupFeatures);
 311
 312                 final ListenableFuture<RpcResult<List<MultipartReply>>> replyTableFeatures = getNodeStaticInfo(MultipartType.OFPMPTABLEFEATURES,
 313                         deviceContext,
 314                         deviceState.getNodeInstanceIdentifier(),
 315                         deviceState.getVersion());
 316                 createSuccessProcessingCallback(MultipartType.OFPMPTABLEFEATURES,
 317                         deviceContext,
 318                         deviceState.getNodeInstanceIdentifier(),
 319                         replyTableFeatures);
 320
 321                 final ListenableFuture<RpcResult<List<MultipartReply>>> replyPortDescription = getNodeStaticInfo(MultipartType.OFPMPPORTDESC,
 322                         deviceContext,
 323                         deviceState.getNodeInstanceIdentifier(),
 324                         deviceState.getVersion());
 325                 createSuccessProcessingCallback(MultipartType.OFPMPPORTDESC,
 326                         deviceContext,
 327                         deviceState.getNodeInstanceIdentifier(),
 328                         replyPortDescription);
 329
 330
 331                 return Futures.allAsList(Arrays.asList(
 332                         replyMeterFeature,
 333                         replyGroupFeatures,
 334 //                        replyTableFeatures,
 335                         replyPortDescription));
 336             }
 337         });
 338
 339     }
 340
 341     @Override
 342     public TranslatorLibrary oook() {
 343         return translatorLibrary;
 344     }
 345
 346     @Override
 347     public void setTranslatorLibrary(final TranslatorLibrary translatorLibrary) {
 348         this.translatorLibrary = translatorLibrary;
 349     }
 350
 351     private static ListenableFuture<RpcResult<List<MultipartReply>>> getNodeStaticInfo(final MultipartType type, final DeviceContext deviceContext,
 352                                                                                        final InstanceIdentifier<Node> nodeII, final short version) {
 353
 354         final OutboundQueue queue = deviceContext.getPrimaryConnectionContext().getOutboundQueueProvider();
 355
 356         final Long reserved = deviceContext.getReservedXid();
 357         final RequestContext<List<MultipartReply>> requestContext = new AbstractRequestContext<List<MultipartReply>>(reserved) {
 358             @Override
 359             public void close() {
 360                 //NOOP
 361             }
 362         };
 363
 364         final Xid xid = requestContext.getXid();
 365
 366         LOG.trace("Hooking xid {} to device context - precaution.", reserved);
 367
 368         final MultiMsgCollector multiMsgCollector = deviceContext.getMultiMsgCollector(requestContext);
 369         queue.commitEntry(xid.getValue(), MultipartRequestInputFactory.makeMultipartRequestInput(xid.getValue(), version, type), new FutureCallback<OfHeader>() {
 370             @Override
 371             public void onSuccess(final OfHeader ofHeader) {
 372                 if (ofHeader instanceof MultipartReply) {
 373                     final MultipartReply multipartReply = (MultipartReply) ofHeader;
 374                     multiMsgCollector.addMultipartMsg(multipartReply);
 375                 } else if (null != ofHeader) {
 376                     LOG.info("Unexpected response type received {}.", ofHeader.getClass());
 377                 } else {
 378                     multiMsgCollector.endCollecting();
 379                     LOG.info("Response received is null.");
 380                 }
 381             }
 382
 383             @Override
 384             public void onFailure(final Throwable t) {
 385                 LOG.info("Fail response from OutboundQueue for multipart type {}.", type, t);
 386                 requestContext.close();
 387                 if (MultipartType.OFPMPTABLEFEATURES.equals(type)) {
 388                     makeEmptyTables(deviceContext, nodeII, deviceContext.getPrimaryConnectionContext().getFeatures().getTables());
 389                 }
 390             }
 391         });
 392
 393         return requestContext.getFuture();
 394     }
 395
 396     private static void createSuccessProcessingCallback(final MultipartType type, final DeviceContext deviceContext, final InstanceIdentifier<Node> nodeII, final ListenableFuture<RpcResult<List<MultipartReply>>> requestContextFuture) {
 397         Futures.addCallback(requestContextFuture, new FutureCallback<RpcResult<List<MultipartReply>>>() {
 398             @Override
 399             public void onSuccess(final RpcResult<List<MultipartReply>> rpcResult) {
 400                 final List<MultipartReply> result = rpcResult.getResult();
 401                 if (result != null) {
 402                     LOG.info("Static node {} info: {} collected", nodeII.toString(), type);
 403                     translateAndWriteReply(type, deviceContext, nodeII, result);
 404                 } else {
 405                     final Iterator<RpcError> rpcErrorIterator = rpcResult.getErrors().iterator();
 406                     while (rpcErrorIterator.hasNext()) {
 407                         final RpcError rpcError = rpcErrorIterator.next();
 408                         LOG.info("Failed to retrieve static node {} info: {}", type, rpcError.getMessage());
 409                         if (null != rpcError.getCause()) {
 410                             LOG.trace("Detailed error:", rpcError.getCause());
 411                         }
 412                     }
 413                     if (MultipartType.OFPMPTABLEFEATURES.equals(type)) {
 414                         makeEmptyTables(deviceContext, nodeII, deviceContext.getPrimaryConnectionContext().getFeatures().getTables());
 415                     }
 416                 }
 417             }
 418
 419             @Override
 420             public void onFailure(final Throwable throwable) {
 421                 LOG.info("Request of type {} for static info of node {} failed.", type, nodeII);
 422             }
 423         });
 424     }
 425
 426     // FIXME : remove after ovs tableFeatures fix
 427     private static void makeEmptyTables(final DeviceContext dContext, final InstanceIdentifier<Node> nodeII, final Short nrOfTables) {
 428         LOG.debug("About to create {} empty tables.", nrOfTables);
 429         for (int i = 0; i < nrOfTables; i++) {
 430             final short tId = (short) i;
 431             final InstanceIdentifier<Table> tableII = nodeII.augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tId));
 432             final TableBuilder tableBuilder = new TableBuilder().setId(tId).addAugmentation(FlowTableStatisticsData.class, new FlowTableStatisticsDataBuilder().build());
 433             dContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, tableII, tableBuilder.build());
 434         }
 435     }
 436
 437     private static void translateAndWriteReply(final MultipartType type, final DeviceContext dContext,
 438                                                final InstanceIdentifier<Node> nodeII, final Collection<MultipartReply> result) {
 439         for (final MultipartReply reply : result) {
 440             final MultipartReplyBody body = reply.getMultipartReplyBody();
 441             switch (type) {
 442                 case OFPMPDESC:
 443                     Preconditions.checkArgument(body instanceof MultipartReplyDescCase);
 444                     final MultipartReplyDesc replyDesc = ((MultipartReplyDescCase) body).getMultipartReplyDesc();
 445                     final FlowCapableNode fcNode = NodeStaticReplyTranslatorUtil.nodeDescTranslator(replyDesc);
 446                     final InstanceIdentifier<FlowCapableNode> fNodeII = nodeII.augmentation(FlowCapableNode.class);
 447                     dContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, fNodeII, fcNode);
 448                     break;
 449
 450                 case OFPMPTABLEFEATURES:
 451                     Preconditions.checkArgument(body instanceof MultipartReplyTableFeaturesCase);
 452                     final MultipartReplyTableFeatures tableFeatures = ((MultipartReplyTableFeaturesCase) body).getMultipartReplyTableFeatures();
 453                     final List<TableFeatures> tables = NodeStaticReplyTranslatorUtil.nodeTableFeatureTranslator(tableFeatures);
 454                     for (final TableFeatures table : tables) {
 455                         final Short tableId = table.getTableId();
 456                         final InstanceIdentifier<Table> tableII = nodeII.augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tableId));
 457                         final TableBuilder tableBuilder = new TableBuilder().setId(tableId).setTableFeatures(Collections.singletonList(table));
 458                         tableBuilder.addAugmentation(FlowTableStatisticsData.class, new FlowTableStatisticsDataBuilder().build());
 459                         dContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, tableII, tableBuilder.build());
 460                     }
 461                     break;
 462
 463                 case OFPMPMETERFEATURES:
 464                     Preconditions.checkArgument(body instanceof MultipartReplyMeterFeaturesCase);
 465                     final MultipartReplyMeterFeatures meterFeatures = ((MultipartReplyMeterFeaturesCase) body).getMultipartReplyMeterFeatures();
 466                     final NodeMeterFeatures mFeature = NodeStaticReplyTranslatorUtil.nodeMeterFeatureTranslator(meterFeatures);
 467                     final InstanceIdentifier<NodeMeterFeatures> mFeatureII = nodeII.augmentation(NodeMeterFeatures.class);
 468                     dContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, mFeatureII, mFeature);
 469                     if (0L < mFeature.getMeterFeatures().getMaxMeter().getValue()) {
 470                         dContext.getDeviceState().setMeterAvailable(true);
 471                     }
 472                     break;
 473
 474                 case OFPMPGROUPFEATURES:
 475                     Preconditions.checkArgument(body instanceof MultipartReplyGroupFeaturesCase);
 476                     final MultipartReplyGroupFeatures groupFeatures = ((MultipartReplyGroupFeaturesCase) body).getMultipartReplyGroupFeatures();
 477                     final NodeGroupFeatures gFeature = NodeStaticReplyTranslatorUtil.nodeGroupFeatureTranslator(groupFeatures);
 478                     final InstanceIdentifier<NodeGroupFeatures> gFeatureII = nodeII.augmentation(NodeGroupFeatures.class);
 479                     dContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, gFeatureII, gFeature);
 480                     break;
 481
 482                 case OFPMPPORTDESC:
 483                     Preconditions.checkArgument(body instanceof MultipartReplyPortDescCase);
 484                     final MultipartReplyPortDesc portDesc = ((MultipartReplyPortDescCase) body).getMultipartReplyPortDesc();
 485                     for (final PortGrouping port : portDesc.getPorts()) {
 486                         final short ofVersion = dContext.getDeviceState().getVersion();
 487                         final TranslatorKey translatorKey = new TranslatorKey(ofVersion, PortGrouping.class.getName());
 488                         final MessageTranslator<PortGrouping, FlowCapableNodeConnector> translator = dContext.oook().lookupTranslator(translatorKey);
 489                         final FlowCapableNodeConnector fcNodeConnector = translator.translate(port, dContext, null);
 490
 491                         final BigInteger dataPathId = dContext.getPrimaryConnectionContext().getFeatures().getDatapathId();
 492                         final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), port.getPortNo(), ofVersion);
 493                         final NodeConnectorBuilder ncBuilder = new NodeConnectorBuilder().setId(nodeConnectorId);
 494                         ncBuilder.addAugmentation(FlowCapableNodeConnector.class, fcNodeConnector);
 495
 496                         ncBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build());
 497                         final NodeConnector connector = ncBuilder.build();
 498
 499                         final InstanceIdentifier<NodeConnector> connectorII = nodeII.child(NodeConnector.class, connector.getKey());
 500                         dContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, connectorII, connector);
 501                     }
 502
 503                     break;
 504
 505                 default:
 506                     throw new IllegalArgumentException("Unnexpected MultipartType " + type);
 507             }
 508         }
 509     }
 510
 511     @Override
 512     public void setNotificationService(final NotificationService notificationServiceParam) {
 513         notificationService = notificationServiceParam;
 514     }
 515
 516     @Override
 517     public void setNotificationPublishService(final NotificationPublishService notificationService) {
 518         notificationPublishService = notificationService;
 519     }
 520
 521     @Override
 522     public void close() throws Exception {
 523         for (final DeviceContext deviceContext : deviceContexts) {
 524             deviceContext.close();
 525         }
 526     }
 527
 528     private static void createEmptyFlowCapableNodeInDs(final DeviceContext deviceContext) {
 529         final FlowCapableNodeBuilder flowCapableNodeBuilder = new FlowCapableNodeBuilder();
 530         final InstanceIdentifier<FlowCapableNode> fNodeII = deviceContext.getDeviceState().getNodeInstanceIdentifier().augmentation(FlowCapableNode.class);
 531         deviceContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, fNodeII, flowCapableNodeBuilder.build());
 532     }
 533
 534     @Override
 535     public void onDeviceContextClosed(final DeviceContext deviceContext) {
 536         deviceContexts.remove(deviceContext);
 537     }
 538
 539     @Override
 540     public void initialize() {
 541         spyPool = new ScheduledThreadPoolExecutor(1);
 542         spyPool.scheduleAtFixedRate(messageIntelligenceAgency, spyRate, spyRate, TimeUnit.SECONDS);
 543     }
 544 }