import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
-
+import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
import org.opendaylight.openflowplugin.api.openflow.OFPManager;
import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleConductor;
import org.opendaylight.openflowplugin.api.openflow.lifecycle.RoleChangeListener;
import org.opendaylight.openflowplugin.api.openflow.lifecycle.ServiceChangeListener;
+import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
import org.opendaylight.openflowplugin.api.openflow.rpc.RpcManager;
+import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageIntelligenceAgency;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
+import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
+import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
+import org.opendaylight.openflowplugin.impl.util.MdSalRegistrationUtils;
import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*/
-final class LifecycleConductorImpl implements LifecycleConductor, RoleChangeListener, DeviceContextChangeListener {
+final class LifecycleConductorImpl implements LifecycleConductor, RoleChangeListener, DeviceContextChangeListener, ExtensionConverterProviderKeeper {
private static final Logger LOG = LoggerFactory.getLogger(LifecycleConductorImpl.class);
private static final int TICKS_PER_WHEEL = 500;
private static final long TICK_DURATION = 10; // 0.5 sec.
private final HashedWheelTimer hashedWheelTimer = new HashedWheelTimer(TICK_DURATION, TimeUnit.MILLISECONDS, TICKS_PER_WHEEL);
+ private ExtensionConverterProvider extensionConverterProvider;
private DeviceManager deviceManager;
private StatisticsManager statisticsManager;
private RpcManager rpcManager;
private final MessageIntelligenceAgency messageIntelligenceAgency;
private ConcurrentHashMap<DeviceInfo, ServiceChangeListener> serviceChangeListeners = new ConcurrentHashMap<>();
+ private NotificationPublishService notificationPublishService;
LifecycleConductorImpl(final MessageIntelligenceAgency messageIntelligenceAgency) {
- Preconditions.checkNotNull(messageIntelligenceAgency);
- this.messageIntelligenceAgency = messageIntelligenceAgency;
+ this.messageIntelligenceAgency = Preconditions.checkNotNull(messageIntelligenceAgency);
+ }
+
+ @Override
+ public ExtensionConverterProvider getExtensionConverterProvider() {
+ return extensionConverterProvider;
+ }
+
+ @Override
+ public void setExtensionConverterProvider(ExtensionConverterProvider extensionConverterProvider) {
+ this.extensionConverterProvider = extensionConverterProvider;
}
@Override
public void setSafelyManager(final OFPManager manager){
- if (manager == null) {
- LOG.info("Manager {} is already defined in conductor. ", manager);
- }
if (manager instanceof RpcManager) {
+ if (rpcManager != null) {
+ LOG.info("RPC manager {} is already defined in conductor. ", manager);
+ return;
+ }
this.rpcManager = (RpcManager) manager;
} else {
if (manager instanceof StatisticsManager) {
+ if (statisticsManager != null) {
+ LOG.info("Statistics manager {} is already defined in conductor. ", manager);
+ return;
+ }
this.statisticsManager = (StatisticsManager) manager;
} else {
if (manager instanceof DeviceManager) {
+ if (deviceManager != null) {
+ LOG.info("Device manager {} is already defined in conductor. ", manager);
+ return;
+ }
this.deviceManager = (DeviceManager) manager;
}
}
LOG.debug("Close connection called for node {}", deviceInfo);
final DeviceContext deviceContext = getDeviceContext(deviceInfo);
if (null != deviceContext) {
+ deviceManager.notifyDeviceValidListeners(deviceInfo, false);
deviceContext.shutdownConnection();
}
}
return;
}
- LOG.info("Role change to {} in role context for node {} was successful, starting/stopping services.", newRole, deviceInfo);
+ LOG.info("Role change to {} in role context for node {} was successful.", newRole, deviceInfo);
+
+ final String logText;
if (OfpRole.BECOMEMASTER.equals(newRole)) {
- statisticsManager.startScheduling(deviceContext.getPrimaryConnectionContext().getDeviceInfo());
+ logText = "Start";
+ statisticsManager.startScheduling(deviceInfo);
+ MdSalRegistrationUtils.registerMasterServices(
+ rpcManager.gainContext(deviceInfo),
+ deviceContext,
+ OfpRole.BECOMEMASTER,
+ this.extensionConverterProvider);
+ if (((RpcContext)rpcManager.gainContext(deviceInfo)).isStatisticsRpcEnabled()) {
+ MdSalRegistrationUtils.registerStatCompatibilityServices(
+ rpcManager.gainContext(deviceInfo),
+ deviceManager.gainContext(deviceInfo),
+ notificationPublishService,
+ new AtomicLong());
+ }
+
+ // Fill flow registry with flows found in operational and config datastore
+ deviceContext.getDeviceFlowRegistry().fill(deviceInfo.getNodeInstanceIdentifier());
} else {
- statisticsManager.stopScheduling(deviceContext.getPrimaryConnectionContext().getDeviceInfo());
+ logText = "Stopp";
+ statisticsManager.stopScheduling(deviceInfo);
+ MdSalRegistrationUtils.registerSlaveServices(
+ rpcManager.gainContext(deviceInfo),
+ OfpRole.BECOMESLAVE);
}
- final ListenableFuture<Void> onClusterRoleChange = deviceContext.onClusterRoleChange(null, newRole);
+ final ListenableFuture<Void> onClusterRoleChange = deviceManager.onClusterRoleChange(deviceInfo, newRole);
Futures.addCallback(onClusterRoleChange, new FutureCallback<Void>() {
@Override
public void onSuccess(@Nullable final Void aVoid) {
- LOG.info("Starting/Stopping services for node {} was successful", deviceInfo);
+ LOG.info("{}ing services for node {} was successful", logText, deviceInfo);
if (newRole.equals(OfpRole.BECOMESLAVE)) notifyServiceChangeListeners(deviceInfo, true);
}
@Override
public void onFailure(final Throwable throwable) {
- LOG.warn("Starting/Stopping services for node {} was NOT successful, closing connection", deviceInfo);
+ LOG.warn("{}ing services for node {} was NOT successful, closing connection", logText, deviceInfo);
closeConnection(deviceInfo);
}
});
@Override
public DeviceContext getDeviceContext(DeviceInfo deviceInfo){
- return deviceManager.getDeviceContextFromNodeId(deviceInfo);
+ return deviceManager.gainContext(deviceInfo);
}
@Override
- public Short gainVersionSafely(final DeviceInfo deviceInfo) {
- return (null != getDeviceContext(deviceInfo)) ? getDeviceContext(deviceInfo).getPrimaryConnectionContext().getFeatures().getVersion() : null;
+ public StatisticsContext getStatisticsContext(DeviceInfo deviceInfo){
+ return statisticsManager.gainContext(deviceInfo);
}
public Timeout newTimeout(@Nonnull TimerTask task, long delay, @Nonnull TimeUnit unit) {
return hashedWheelTimer.newTimeout(task, delay, unit);
}
- ConnectionContext.CONNECTION_STATE gainConnectionStateSafely(final DeviceInfo deviceInfo){
+ @Override
+ public ConnectionContext.CONNECTION_STATE gainConnectionStateSafely(final DeviceInfo deviceInfo){
return (null != getDeviceContext(deviceInfo)) ? getDeviceContext(deviceInfo).getPrimaryConnectionContext().getConnectionState() : null;
}
}
@VisibleForTesting
- public boolean isServiceChangeListenersEmpty() {
+ boolean isServiceChangeListenersEmpty() {
return this.serviceChangeListeners.isEmpty();
}
+ @Override
+ public NotificationPublishService getNotificationPublishService() {
+ return notificationPublishService;
+ }
+
+ @Override
+ public void setNotificationPublishService(NotificationPublishService notificationPublishService) {
+ this.notificationPublishService = notificationPublishService;
+ }
}
Code Review / openflowplugin.git / blobdiff