*/
package org.opendaylight.openflowplugin.impl;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.SettableFuture;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
import java.lang.management.ManagementFactory;
import java.util.Collection;
import java.util.List;
-import java.util.Objects;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
-import javax.annotation.Nonnull;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.inject.Inject;
import javax.management.ObjectName;
import org.apache.aries.blueprint.annotation.service.Reference;
import org.apache.aries.blueprint.annotation.service.Service;
-import org.opendaylight.controller.md.sal.binding.api.DataBroker;
-import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
-import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
import org.opendaylight.infrautils.diagstatus.ServiceState;
import org.opendaylight.infrautils.ready.SystemReadyListener;
import org.opendaylight.infrautils.ready.SystemReadyMonitor;
+import org.opendaylight.mdsal.binding.api.DataBroker;
+import org.opendaylight.mdsal.binding.api.NotificationPublishService;
+import org.opendaylight.mdsal.binding.api.RpcProviderService;
import org.opendaylight.mdsal.eos.binding.api.EntityOwnershipService;
import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
+import org.opendaylight.openflowjava.protocol.api.connection.OpenflowDiagStatusProvider;
import org.opendaylight.openflowjava.protocol.spi.connection.SwitchConnectionProvider;
import org.opendaylight.openflowjava.protocol.spi.connection.SwitchConnectionProviderList;
import org.opendaylight.openflowplugin.api.openflow.OpenFlowPluginProvider;
private final Collection<SwitchConnectionProvider> switchConnectionProviders;
private final DeviceInitializerProvider deviceInitializerProvider;
private final ConvertorManager convertorManager;
- private final RpcProviderRegistry rpcProviderRegistry;
+ private final RpcProviderService rpcProviderRegistry;
private final ClusterSingletonServiceProvider singletonServicesProvider;
private final OpenflowProviderConfig config;
private final EntityOwnershipService entityOwnershipService;
private ConnectionManager connectionManager;
private ListeningExecutorService executorService;
private ContextChainHolderImpl contextChainHolder;
- private final OpenflowPluginDiagStatusProvider openflowPluginStatusMonitor;
+ private final OpenflowDiagStatusProvider openflowDiagStatusProvider;
+ private final SettableFuture<Void> fullyStarted = SettableFuture.create();
+ private static final String OPENFLOW_SERVICE_NAME = "OPENFLOW";
public static MessageIntelligenceAgency getMessageIntelligenceAgency() {
return MESSAGE_INTELLIGENCE_AGENCY;
public OpenFlowPluginProviderImpl(final ConfigurationService configurationService,
final SwitchConnectionProviderList switchConnectionProviders,
final PingPongDataBroker pingPongDataBroker,
- final @Reference RpcProviderRegistry rpcProviderRegistry,
+ final @Reference RpcProviderService rpcProviderRegistry,
final @Reference NotificationPublishService notificationPublishService,
final @Reference ClusterSingletonServiceProvider singletonServiceProvider,
final @Reference EntityOwnershipService entityOwnershipService,
final MastershipChangeServiceManager mastershipChangeServiceManager,
- final OpenflowPluginDiagStatusProvider openflowPluginStatusMonitor,
+ final @Reference OpenflowDiagStatusProvider openflowDiagStatusProvider,
final @Reference SystemReadyMonitor systemReadyMonitor) {
this.switchConnectionProviders = switchConnectionProviders;
this.dataBroker = pingPongDataBroker;
deviceInitializerProvider = DeviceInitializerProviderFactory.createDefaultProvider();
config = new OpenFlowProviderConfigImpl(configurationService);
this.mastershipChangeServiceManager = mastershipChangeServiceManager;
- this.openflowPluginStatusMonitor = openflowPluginStatusMonitor;
+ this.openflowDiagStatusProvider = openflowDiagStatusProvider;
systemReadyMonitor.registerListener(this);
- LOG.debug("registered onSystemBootReady() listener for deferred startSwitchConnections()");
+ LOG.info("registered onSystemBootReady() listener for deferred startSwitchConnections()");
}
@Override
public void onSystemBootReady() {
- LOG.debug("onSystemBootReady() received, starting the switch connections");
- startSwitchConnections();
- }
-
- private void startSwitchConnections() {
+ LOG.info("onSystemBootReady() received, starting the switch connections");
Futures.addCallback(Futures.allAsList(switchConnectionProviders.stream().map(switchConnectionProvider -> {
// Inject OpenFlowPlugin custom serializers and deserializers into OpenFlowJava
if (config.isUseSingleLayerSerialization()) {
return switchConnectionProvider.startup();
}).collect(Collectors.toSet())), new FutureCallback<List<Boolean>>() {
@Override
- public void onSuccess(@Nonnull final List<Boolean> result) {
+ public void onSuccess(final List<Boolean> result) {
LOG.info("All switchConnectionProviders are up and running ({}).", result.size());
- openflowPluginStatusMonitor.reportStatus(ServiceState.OPERATIONAL);
+ openflowDiagStatusProvider.reportStatus(OPENFLOW_SERVICE_NAME, ServiceState.OPERATIONAL);
+ fullyStarted.set(null);
}
@Override
- public void onFailure(@Nonnull final Throwable throwable) {
+ public void onFailure(final Throwable throwable) {
LOG.warn("Some switchConnectionProviders failed to start.", throwable);
- openflowPluginStatusMonitor.reportStatus(ServiceState.ERROR, throwable);
+ openflowDiagStatusProvider.reportStatus(OPENFLOW_SERVICE_NAME, throwable);
+ fullyStarted.setException(throwable);
}
}, MoreExecutors.directExecutor());
}
+ @VisibleForTesting
+ public Future<Void> getFullyStarted() {
+ return fullyStarted;
+ }
+
private ListenableFuture<List<Boolean>> shutdownSwitchConnections() {
final ListenableFuture<List<Boolean>> listListenableFuture =
Futures.allAsList(switchConnectionProviders.stream().map(switchConnectionProvider -> {
Futures.addCallback(listListenableFuture, new FutureCallback<List<Boolean>>() {
@Override
- public void onSuccess(@Nonnull final List<Boolean> result) {
+ public void onSuccess(final List<Boolean> result) {
LOG.info("All switchConnectionProviders were successfully shut down ({}).", result.size());
}
@Override
- public void onFailure(@Nonnull final Throwable throwable) {
+ public void onFailure(final Throwable throwable) {
LOG.warn("Some switchConnectionProviders failed to shutdown.", throwable);
}
}, MoreExecutors.directExecutor());
// constructed threads when they are available.
// Threads that have not been used for x seconds are terminated and removed from the cache.
executorService = MoreExecutors.listeningDecorator(new ThreadPoolLoggingExecutor(
- config.getThreadPoolMinThreads(),
- config.getThreadPoolMaxThreads().getValue(),
- config.getThreadPoolTimeout(),
+ config.getThreadPoolMinThreads().toJava(),
+ config.getThreadPoolMaxThreads().getValue().toJava(),
+ config.getThreadPoolTimeout().toJava(),
TimeUnit.SECONDS, new SynchronousQueue<>(), POOL_NAME));
deviceManager = new DeviceManagerImpl(
notificationPublishService,
hashedWheelTimer,
convertorManager,
- deviceInitializerProvider);
+ deviceInitializerProvider,
+ executorService);
TranslatorLibraryUtil.injectBasicTranslatorLibrary(deviceManager, convertorManager);
((ExtensionConverterProviderKeeper) deviceManager).setExtensionConverterProvider(extensionConverterManager);
convertorManager,
executorService);
- roleManager = new RoleManagerImpl(hashedWheelTimer, config);
+ roleManager = new RoleManagerImpl(hashedWheelTimer, config, executorService);
contextChainHolder = new ContextChainHolderImpl(
executorService,
gracefulShutdown(executorService);
gracefulShutdown(hashedWheelTimer);
unregisterMXBean(MESSAGE_INTELLIGENCE_AGENCY_MX_BEAN_NAME);
- openflowPluginStatusMonitor.reportStatus(ServiceState.UNREGISTERED);
+ openflowDiagStatusProvider.reportStatus(ServiceState.UNREGISTERED);
}
@SuppressWarnings("checkstyle:IllegalCatch")
private static void gracefulShutdown(final AutoCloseable closeable) {
- if (Objects.isNull(closeable)) {
- return;
- }
-
- try {
- closeable.close();
- } catch (Exception e) {
- LOG.warn("Failed to shutdown {} gracefully.", closeable);
+ if (closeable != null) {
+ try {
+ closeable.close();
+ } catch (Exception e) {
+ LOG.warn("Failed to shutdown {} gracefully.", closeable);
+ }
}
}
private static void gracefulShutdown(final Timer timer) {
- if (Objects.isNull(timer)) {
- return;
- }
-
- try {
- timer.stop();
- } catch (IllegalStateException e) {
- LOG.warn("Failed to shutdown {} gracefully.", timer);
+ if (timer != null) {
+ try {
+ timer.stop();
+ } catch (IllegalStateException e) {
+ LOG.warn("Failed to shutdown {} gracefully.", timer);
+ }
}
}
private static void gracefulShutdown(final ExecutorService executorService) {
- if (Objects.isNull(executorService)) {
- return;
- }
-
- try {
- executorService.shutdownNow();
- } catch (SecurityException e) {
- LOG.warn("Failed to shutdown {} gracefully.", executorService);
+ if (executorService != null) {
+ try {
+ executorService.shutdownNow();
+ } catch (SecurityException e) {
+ LOG.warn("Failed to shutdown {} gracefully.", executorService);
+ }
}
}
| NotCompliantMBeanException
| MBeanRegistrationException
| InstanceAlreadyExistsException e) {
- LOG.warn("Error registering MBean {}", e);
+ LOG.warn("Error registering MBean {}", beanName, e);
}
}
} catch (InstanceNotFoundException
| MBeanRegistrationException
| MalformedObjectNameException e) {
- LOG.warn("Error unregistering MBean {}", e);
+ LOG.warn("Error unregistering MBean {}", beanName, e);
}
}
}
Code Review / openflowplugin.git / blobdiff