Bug 4957 RoleContext updated with initialization
[openflowplugin.git] / openflowplugin-impl / src / main / java / org / opendaylight / openflowplugin / impl / device / DeviceManagerImpl.java
index 53d950738b724ca6f83f77b5b28c3a9c22da616a..b5a99013b7ea988a08800e7a08cb60afb88b0f47 100644 (file)
@@ -9,231 +9,261 @@ package org.opendaylight.openflowplugin.impl.device;
 
 import javax.annotation.CheckForNull;
 import javax.annotation.Nonnull;
-import java.util.*;
-import java.util.concurrent.Future;
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
 import com.google.common.base.Preconditions;
-import com.google.common.util.concurrent.*;
+import com.google.common.collect.Sets;
 import io.netty.util.HashedWheelTimer;
 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
+import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
+import org.opendaylight.controller.md.sal.binding.api.NotificationService;
+import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
+import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandlerRegistration;
 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
-import org.opendaylight.openflowplugin.api.openflow.device.*;
-import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceContextReadyHandler;
-import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
-import org.opendaylight.openflowplugin.api.openflow.rpc.RpcManager;
-import org.opendaylight.openflowplugin.impl.common.MultipartRequestInputFactory;
-import org.opendaylight.openflowplugin.impl.common.NodeStaticReplyTranslatorUtil;
+import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
+import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
+import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
+import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
+import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
+import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceInitializationPhaseHandler;
+import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageIntelligenceAgency;
+import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
+import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
+import org.opendaylight.openflowplugin.impl.connection.OutboundQueueProviderImpl;
 import org.opendaylight.openflowplugin.impl.device.listener.OpenflowProtocolListenerFullImpl;
-import org.opendaylight.openflowplugin.impl.rpc.RequestContextImpl;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupFeatures;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodesBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterFeatures;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.MultipartType;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.MultipartReplyBody;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.MultipartReplyDescCase;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.MultipartReplyGroupFeaturesCase;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.MultipartReplyMeterFeaturesCase;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.MultipartReplyTableFeaturesCase;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.multipart.reply.desc._case.MultipartReplyDesc;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.multipart.reply.group.features._case.MultipartReplyGroupFeatures;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.multipart.reply.meter.features._case.MultipartReplyMeterFeatures;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.multipart.reply.table.features._case.MultipartReplyTableFeatures;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.table.features.TableFeatures;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole;
 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-import org.opendaylight.yangtools.yang.common.RpcError;
-import org.opendaylight.yangtools.yang.common.RpcResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
  *
  */
-public class DeviceManagerImpl implements DeviceManager {
+public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProviderKeeper, AutoCloseable {
 
     private static final Logger LOG = LoggerFactory.getLogger(DeviceManagerImpl.class);
 
-    private static final long TICK_DURATION = 500; // 0.5 sec.
+    private static final long TICK_DURATION = 10; // 0.5 sec.
+    private final long globalNotificationQuota;
+    private ScheduledThreadPoolExecutor spyPool;
+    private final int spyRate = 10;
 
-    private final RpcManager rpcManager;
     private final DataBroker dataBroker;
     private final HashedWheelTimer hashedWheelTimer;
-    private RequestContextStack dummyRequestContextStack;
+    private TranslatorLibrary translatorLibrary;
+    private DeviceInitializationPhaseHandler deviceInitPhaseHandler;
+    private NotificationService notificationService;
+    private NotificationPublishService notificationPublishService;
 
+    private final Set<DeviceContext> deviceContexts = Sets.newConcurrentHashSet();
+    private final MessageIntelligenceAgency messageIntelligenceAgency;
 
-    public DeviceManagerImpl (@Nonnull final RpcManager rpcManager, @Nonnull final DataBroker dataBroker) {
-        this.rpcManager = Preconditions.checkNotNull(rpcManager);
+    private final long barrierNanos = TimeUnit.MILLISECONDS.toNanos(500);
+    private final int maxQueueDepth = 25600;
+    private final boolean switchFeaturesMandatory;
+    private final DeviceTransactionChainManagerProvider deviceTransactionChainManagerProvider;
+    private ExtensionConverterProvider extensionConverterProvider;
+
+    public DeviceManagerImpl(@Nonnull final DataBroker dataBroker,
+                             @Nonnull final MessageIntelligenceAgency messageIntelligenceAgency,
+                             final boolean switchFeaturesMandatory,
+                             final long globalNotificationQuota) {
+        this.globalNotificationQuota = globalNotificationQuota;
         this.dataBroker = Preconditions.checkNotNull(dataBroker);
-        hashedWheelTimer = new HashedWheelTimer(TICK_DURATION, TimeUnit.MILLISECONDS, 10);
+        hashedWheelTimer = new HashedWheelTimer(TICK_DURATION, TimeUnit.MILLISECONDS, 500);
+        /* merge empty nodes to oper DS to predict any problems with missing parent for Node */
+        final WriteTransaction tx = dataBroker.newWriteOnlyTransaction();
+
+        final NodesBuilder nodesBuilder = new NodesBuilder();
+        nodesBuilder.setNode(Collections.<Node>emptyList());
+        tx.merge(LogicalDatastoreType.OPERATIONAL, InstanceIdentifier.create(Nodes.class), nodesBuilder.build());
+        try {
+            tx.submit().get();
+        } catch (ExecutionException | InterruptedException e) {
+            LOG.error("Creation of node failed.", e);
+            throw new IllegalStateException(e);
+        }
+
+        this.messageIntelligenceAgency = messageIntelligenceAgency;
+        this.switchFeaturesMandatory = switchFeaturesMandatory;
+        deviceTransactionChainManagerProvider = new DeviceTransactionChainManagerProvider(dataBroker);
+    }
+
+
+    @Override
+    public void setDeviceInitializationPhaseHandler(final DeviceInitializationPhaseHandler handler) {
+        deviceInitPhaseHandler = handler;
+    }
+
+    @Override
+    public void onDeviceContextLevelUp(final DeviceContext deviceContext) {
+        // final phase - we have to add new Device to MD-SAL DataStore
+        Preconditions.checkNotNull(deviceContext);
+        try {
+
+            if (deviceContext.getDeviceState().getRole() != OfpRole.BECOMESLAVE) {
+                ((DeviceContextImpl) deviceContext).initialSubmitTransaction();
+                deviceContext.onPublished();
+
+            } else {
+                //if role = slave
+                try {
+                    ((DeviceContextImpl) deviceContext).cancelTransaction();
+                } catch (final Exception e) {
+                    //TODO: how can we avoid it. pingpong does not have cancel
+                    LOG.debug("Expected Exception: Cancel Txn exception thrown for slaves", e);
+                }
 
-        dummyRequestContextStack = new RequestContextStack() {
-            @Override
-            public <T> void forgetRequestContext(RequestContext<T> requestContext) {
-                //NOOP
-            }
-            @Override
-            public <T> SettableFuture<RpcResult<T>> storeOrFail(RequestContext<T> data) {
-                return data.getFuture();
             }
-            @Override
-            public <T> RequestContext<T> createRequestContext() {
-                return new RequestContextImpl<>(this);
+
+        } catch (final Exception e) {
+            LOG.warn("Node {} can not be add to OPERATIONAL DataStore yet because {} ", deviceContext.getDeviceState().getNodeId(), e.getMessage());
+            LOG.trace("Problem with add node {} to OPERATIONAL DataStore", deviceContext.getDeviceState().getNodeId(), e);
+            try {
+                deviceContext.close();
+            } catch (final Exception e1) {
+                LOG.warn("Device context close FAIL - " + deviceContext.getDeviceState().getNodeId());
             }
-        };
+        }
     }
 
     @Override
     public void deviceConnected(@CheckForNull final ConnectionContext connectionContext) {
         Preconditions.checkArgument(connectionContext != null);
-        final DeviceState deviceState = new DeviceStateImpl(connectionContext.getFeatures(), connectionContext.getNodeId());
-        final DeviceContextImpl deviceContext = new DeviceContextImpl(connectionContext, deviceState, dataBroker, hashedWheelTimer);
+        try {
+            initializeDeviceContext(connectionContext);
+        } catch (Exception e) {
+            LOG.warn("Exception during initialization phase.", e);
+        }
+    }
+
+    private void initializeDeviceContext(final ConnectionContext connectionContext) throws Exception{
+        LOG.info("Initializing New Connection DeviceContext for node:{}",  connectionContext.getNodeId());
+        // Cache this for clarity
+        final ConnectionAdapter connectionAdapter = connectionContext.getConnectionAdapter();
+
+        //FIXME: as soon as auxiliary connection are fully supported then this is needed only before device context published
+        connectionAdapter.setPacketInFiltering(true);
+
+        final Short version = connectionContext.getFeatures().getVersion();
+        final OutboundQueueProvider outboundQueueProvider = new OutboundQueueProviderImpl(version);
+
+        connectionContext.setOutboundQueueProvider(outboundQueueProvider);
+        final OutboundQueueHandlerRegistration<OutboundQueueProvider> outboundQueueHandlerRegistration =
+                connectionAdapter.registerOutboundQueueHandler(outboundQueueProvider, maxQueueDepth, barrierNanos);
+        connectionContext.setOutboundQueueHandleRegistration(outboundQueueHandlerRegistration);
+
+        final NodeId nodeId = connectionContext.getNodeId();
+        final DeviceState deviceState = new DeviceStateImpl(connectionContext.getFeatures(), nodeId);
+
+        final DeviceContext deviceContext = new DeviceContextImpl(connectionContext, deviceState, dataBroker,
+                hashedWheelTimer, messageIntelligenceAgency, outboundQueueProvider, translatorLibrary);
+
+        deviceContext.addDeviceContextClosedHandler(this);
+        // We would like to crete/register TxChainManager after
+        final DeviceTransactionChainManagerProvider.TransactionChainManagerRegistration txChainManagerReg = deviceTransactionChainManagerProvider
+                .provideTransactionChainManager(connectionContext);
+        if (txChainManagerReg.ownedByInvokingConnectionContext()) {
+            //this actually is new registration for currently processed connection context
+            ((DeviceContextImpl) deviceContext).setTransactionChainManager(txChainManagerReg.getTransactionChainManager());
+        } else {
+            LOG.info("In deviceConnected {}, ownedByInvokingConnectionContext is false", connectionContext.getNodeId());
+            deviceContext.close();
+            return;
+        }
+        ((ExtensionConverterProviderKeeper) deviceContext).setExtensionConverterProvider(extensionConverterProvider);
+        deviceContext.setNotificationService(notificationService);
+        deviceContext.setNotificationPublishService(notificationPublishService);
+
+        deviceContexts.add(deviceContext);
+
+        updatePacketInRateLimiters();
+
         final OpenflowProtocolListenerFullImpl messageListener = new OpenflowProtocolListenerFullImpl(
-                connectionContext.getConnectionAdapter(), deviceContext);
-        connectionContext.getConnectionAdapter().setMessageListener(messageListener);
-
-        final Xid nodeDescXid = deviceContext.getNextXid();
-        final ListenableFuture<Collection<MultipartReply>> replyDesc = getNodeStaticInfo(nodeDescXid, messageListener,
-                MultipartType.OFPMPDESC, deviceContext, deviceState.getNodeInstanceIdentifier(), deviceState.getVersion());
-
-        final Xid nodeMeterXid = deviceContext.getNextXid();
-        final ListenableFuture<Collection<MultipartReply>> replyMeterFeature = getNodeStaticInfo(nodeMeterXid, messageListener,
-                MultipartType.OFPMPMETERFEATURES, deviceContext, deviceState.getNodeInstanceIdentifier(), deviceState.getVersion());
-
-        final Xid nodeGroupXid = deviceContext.getNextXid();
-        final ListenableFuture<Collection<MultipartReply>> replyGroupFeatures = getNodeStaticInfo(nodeGroupXid, messageListener,
-                MultipartType.OFPMPGROUPFEATURES, deviceContext, deviceState.getNodeInstanceIdentifier(), deviceState.getVersion());
-
-        final Xid nodeTableXid = deviceContext.getNextXid();
-        final ListenableFuture<Collection<MultipartReply>> replyTableFeatures = getNodeStaticInfo(nodeTableXid, messageListener,
-                MultipartType.OFPMPTABLEFEATURES, deviceContext, deviceState.getNodeInstanceIdentifier(), deviceState.getVersion());
-
-        final ListenableFuture<List<Collection<MultipartReply>>> deviceFeaturesFuture =
-                Futures.allAsList(Arrays.asList(replyDesc, replyMeterFeature, replyGroupFeatures, replyTableFeatures));
-        Futures.addCallback(deviceFeaturesFuture, new FutureCallback<List<Collection<MultipartReply>>>() {
-            @Override
-            public void onSuccess(final List<Collection<MultipartReply>> result) {
-                // FIXME : add statistics
-                rpcManager.deviceConnected(deviceContext);
-                deviceContext.submitTransaction();
-            }
+                connectionAdapter, deviceContext);
+        connectionAdapter.setMessageListener(messageListener);
 
-            @Override
-            public void onFailure(final Throwable t) {
-                // FIXME : remove session
-            }
-        });
+        deviceCtxLevelUp(deviceContext);
     }
 
-    @Override
-    public void addRequestContextReadyHandler(final DeviceContextReadyHandler deviceContextReadyHandler) {
-        // TODO Auto-generated method stub
-    }
-
-    private ListenableFuture<Collection<MultipartReply>> getNodeStaticInfo(final Xid xid,
-            final MultiMsgCollector multiMsgCollector, final MultipartType type, final DeviceContext dContext,
-            final InstanceIdentifier<Node> nodeII, final short version) {
-        final ListenableFuture<Collection<MultipartReply>> future = multiMsgCollector.registerMultipartMsg(xid.getValue());
-
-        RequestContext<List<MultipartReply>> requestContext = dummyRequestContextStack.createRequestContext();
-        dContext.hookRequestCtx(xid, requestContext);
-        Futures.addCallback(requestContext.getFuture(), new FutureCallback<RpcResult<List<MultipartReply>>>() {
-            @Override
-            public void onSuccess(final RpcResult<List<MultipartReply>> rpcResult) {
-                List<MultipartReply> result = rpcResult.getResult();
-                if (result != null) {
-                    translateAndWriteReply(type, dContext, nodeII, result);
-                } else {
-                    Iterator<RpcError> rpcErrorIterator = rpcResult.getErrors().iterator();
-                    while (rpcErrorIterator.hasNext()) {
-                        Throwable t = rpcErrorIterator.next().getCause();
-                        LOG.info("Failed to retrieve static node {} info: {}", type, t.getMessage());
-                        LOG.trace("Detailed error:", t);
-                    }
-                    if (MultipartType.OFPMPTABLE.equals(type)) {
-                        makeEmptyTables(dContext, nodeII, dContext.getPrimaryConnectionContext().getFeatures().getTables());
-                    }
+    private void updatePacketInRateLimiters() {
+        synchronized (deviceContexts) {
+            final int deviceContextsSize = deviceContexts.size();
+            if (deviceContextsSize > 0) {
+                long freshNotificationLimit = globalNotificationQuota / deviceContextsSize;
+                if (freshNotificationLimit < 100) {
+                    freshNotificationLimit = 100;
+                }
+                LOG.debug("fresh notification limit = {}", freshNotificationLimit);
+                for (DeviceContext deviceContext : deviceContexts) {
+                    deviceContext.updatePacketInRateLimit(freshNotificationLimit);
                 }
             }
+        }
+    }
 
-            @Override
-            public void onFailure(final Throwable t) {
-                LOG.info("Failed to retrieve static node {} info: {}", type, t.getMessage());
-            }
-        });
-
-        final Future<RpcResult<Void>> rpcFuture = dContext.getPrimaryConnectionContext().getConnectionAdapter()
-                .multipartRequest(MultipartRequestInputFactory.makeMultipartRequestInput(xid.getValue(), version, type));
-        Futures.addCallback(JdkFutureAdapters.listenInPoolThread(rpcFuture), new FutureCallback<RpcResult<Void>>() {
-            @Override
-            public void onSuccess(final RpcResult<Void> result) {
-                // NOOP
-            }
-            @Override
-            public void onFailure(final Throwable t) {
-                future.cancel(true);
-            }
-        });
-        return future;
+    void deviceCtxLevelUp(final DeviceContext deviceContext) {
+        deviceContext.getDeviceState().setValid(true);
+        LOG.trace("Device context level up called.");
+        deviceInitPhaseHandler.onDeviceContextLevelUp(deviceContext);
     }
 
-    // FIXME : remove after ovs tableFeatures fix
-    private static void makeEmptyTables(final DeviceContext dContext, final InstanceIdentifier<Node> nodeII, final Short nrOfTables) {
-        for (int i = 0; i < nrOfTables; i++) {
-            final short tId = (short) i;
-            final InstanceIdentifier<Table> tableII = nodeII.augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tId));
-            dContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, tableII, new TableBuilder().setId(tId).build());
-        }
+    @Override
+    public TranslatorLibrary oook() {
+        return translatorLibrary;
     }
 
-    private static void translateAndWriteReply(final MultipartType type, final DeviceContext dContext,
-            final InstanceIdentifier<Node> nodeII, final Collection<MultipartReply> result) {
-        for (final MultipartReply reply : result) {
-            final MultipartReplyBody body = reply.getMultipartReplyBody();
-            switch (type) {
-            case OFPMPDESC:
-                Preconditions.checkArgument(body instanceof MultipartReplyDescCase);
-                final MultipartReplyDesc replyDesc = ((MultipartReplyDescCase) body).getMultipartReplyDesc();
-                final FlowCapableNode fcNode = NodeStaticReplyTranslatorUtil.nodeDescTranslator(replyDesc);
-                final InstanceIdentifier<FlowCapableNode> fNodeII = nodeII.augmentation(FlowCapableNode.class);
-                dContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, fNodeII, fcNode);
-                break;
-
-            case OFPMPTABLEFEATURES:
-                Preconditions.checkArgument(body instanceof MultipartReplyTableFeaturesCase);
-                final MultipartReplyTableFeatures tableFeatures = ((MultipartReplyTableFeaturesCase) body).getMultipartReplyTableFeatures();
-                final List<TableFeatures> tables = NodeStaticReplyTranslatorUtil.nodeTableFeatureTranslator(tableFeatures);
-                for (final TableFeatures table : tables) {
-                    final Short tableId = table.getTableId();
-                    final InstanceIdentifier<Table> tableII = nodeII.augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tableId));
-                    dContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, tableII, new TableBuilder().setId(tableId).setTableFeatures(Collections.singletonList(table)).build());
-                }
-                break;
-
-            case OFPMPMETERFEATURES:
-                Preconditions.checkArgument(body instanceof MultipartReplyMeterFeaturesCase);
-                final MultipartReplyMeterFeatures meterFeatures = ((MultipartReplyMeterFeaturesCase) body).getMultipartReplyMeterFeatures();
-                final NodeMeterFeatures mFeature = NodeStaticReplyTranslatorUtil.nodeMeterFeatureTranslator(meterFeatures);
-                final InstanceIdentifier<NodeMeterFeatures> mFeatureII = nodeII.augmentation(NodeMeterFeatures.class);
-                dContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, mFeatureII, mFeature);
-                break;
-
-            case OFPMPGROUPFEATURES:
-                Preconditions.checkArgument(body instanceof MultipartReplyGroupFeaturesCase);
-                final MultipartReplyGroupFeatures groupFeatures = ((MultipartReplyGroupFeaturesCase) body).getMultipartReplyGroupFeatures();
-                final NodeGroupFeatures gFeature = NodeStaticReplyTranslatorUtil.nodeGroupFeatureTranslator(groupFeatures);
-                final InstanceIdentifier<NodeGroupFeatures> gFeatureII = nodeII.augmentation(NodeGroupFeatures.class);
-                dContext.writeToTransaction(LogicalDatastoreType.OPERATIONAL, gFeatureII, gFeature);
-                break;
-
-            default:
-                throw new IllegalArgumentException("Unnexpected MultipartType " + type);
-            }
+    @Override
+    public void setTranslatorLibrary(final TranslatorLibrary translatorLibrary) {
+        this.translatorLibrary = translatorLibrary;
+    }
+
+    @Override
+    public void setNotificationService(final NotificationService notificationServiceParam) {
+        notificationService = notificationServiceParam;
+    }
+
+    @Override
+    public void setNotificationPublishService(final NotificationPublishService notificationService) {
+        notificationPublishService = notificationService;
+    }
+
+    @Override
+    public void close() throws Exception {
+        for (final DeviceContext deviceContext : deviceContexts) {
+            deviceContext.close();
         }
     }
+
+    @Override
+    public void onDeviceContextClosed(final DeviceContext deviceContext) {
+        deviceContexts.remove(deviceContext);
+        updatePacketInRateLimiters();
+    }
+
+    @Override
+    public void initialize() {
+        spyPool = new ScheduledThreadPoolExecutor(1);
+        spyPool.scheduleAtFixedRate(messageIntelligenceAgency, spyRate, spyRate, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public void setExtensionConverterProvider(ExtensionConverterProvider extensionConverterProvider) {
+        this.extensionConverterProvider = extensionConverterProvider;
+    }
+
+    @Override
+    public ExtensionConverterProvider getExtensionConverterProvider() {
+        return extensionConverterProvider;
+    }
 }