import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import io.netty.util.TimerTask;
-import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashSet;
import java.util.Iterator;
-import java.util.List;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import org.opendaylight.openflowplugin.impl.connection.OutboundQueueProviderImpl;
import org.opendaylight.openflowplugin.impl.device.listener.OpenflowProtocolListenerFullImpl;
import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtils;
+import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
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;
private final long globalNotificationQuota;
private final boolean switchFeaturesMandatory;
+ private boolean isNotificationFlowRemovedOff;
- private final int spyRate = 10;
+ private static final int SPY_RATE = 10;
private final DataBroker dataBroker;
+ private final ConvertorExecutor convertorExecutor;
private TranslatorLibrary translatorLibrary;
private DeviceInitializationPhaseHandler deviceInitPhaseHandler;
private DeviceTerminationPhaseHandler deviceTerminPhaseHandler;
private final int barrierCountLimit;
private ExtensionConverterProvider extensionConverterProvider;
private ScheduledThreadPoolExecutor spyPool;
- private List<DeviceSynchronizeListener> deviceSynchronizedListeners;
- private List<DeviceValidListener> deviceValidListeners;
+ private Set<DeviceSynchronizeListener> deviceSynchronizedListeners;
+ private Set<DeviceValidListener> deviceValidListeners;
private final LifecycleConductor conductor;
public DeviceManagerImpl(@Nonnull final DataBroker dataBroker,
- final long globalNotificationQuota, final boolean switchFeaturesMandatory,
- final long barrierInterval, final int barrierCountLimit,
- final LifecycleConductor lifecycleConductor) {
+ final long globalNotificationQuota,
+ final boolean switchFeaturesMandatory,
+ final long barrierInterval,
+ final int barrierCountLimit,
+ final LifecycleConductor lifecycleConductor,
+ boolean isNotificationFlowRemovedOff,
+ final ConvertorExecutor convertorExecutor) {
this.switchFeaturesMandatory = switchFeaturesMandatory;
this.globalNotificationQuota = globalNotificationQuota;
+ this.isNotificationFlowRemovedOff = isNotificationFlowRemovedOff;
this.dataBroker = Preconditions.checkNotNull(dataBroker);
+ this.convertorExecutor = convertorExecutor;
/* merge empty nodes to oper DS to predict any problems with missing parent for Node */
final WriteTransaction tx = dataBroker.newWriteOnlyTransaction();
this.conductor = lifecycleConductor;
spyPool = new ScheduledThreadPoolExecutor(1);
- this.deviceSynchronizedListeners = new ArrayList<>();
- this.deviceValidListeners = new ArrayList<>();
+ this.deviceSynchronizedListeners = new HashSet<>();
+ this.deviceValidListeners = new HashSet<>();
}
// final phase - we have to add new Device to MD-SAL DataStore
LOG.debug("Final phase of DeviceContextLevelUp for Node: {} ", deviceInfo.getNodeId());
DeviceContext deviceContext = Preconditions.checkNotNull(deviceContexts.get(deviceInfo));
- ((DeviceContextImpl) deviceContext).initialSubmitTransaction();
deviceContext.onPublished();
}
Preconditions.checkArgument(connectionContext != null);
DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
- /**
+ /*
* This part prevent destroy another device context. Throwing here an exception result to propagate close connection
* in {@link org.opendaylight.openflowplugin.impl.connection.org.opendaylight.openflowplugin.impl.connection.HandshakeContextImpl}
* If context already exist we are in state closing process (connection flapping) and we should not propagate connection close
connectionContext.setOutboundQueueHandleRegistration(outboundQueueHandlerRegistration);
final DeviceState deviceState = new DeviceStateImpl(deviceInfo);
- this.registerDeviceSynchronizeListeners(deviceState);
- this.registerDeviceValidListeners(deviceState);
+ this.addDeviceSynchronizeListener(deviceState);
+ this.addDeviceValidListener(deviceState);
final DeviceContext deviceContext = new DeviceContextImpl(connectionContext,
deviceState,
dataBroker,
conductor,
outboundQueueProvider,
- translatorLibrary);
+ translatorLibrary,
+ this,
+ connectionContext.getDeviceInfo(),
+ convertorExecutor);
Verify.verify(deviceContexts.putIfAbsent(deviceInfo, deviceContext) == null, "DeviceCtx still not closed.");
+ deviceContext.setSwitchFeaturesMandatory(switchFeaturesMandatory);
+
((ExtensionConverterProviderKeeper) deviceContext).setExtensionConverterProvider(extensionConverterProvider);
deviceContext.setNotificationPublishService(conductor.getNotificationPublishService());
@Override
public void initialize() {
- spyPool.scheduleAtFixedRate(conductor.getMessageIntelligenceAgency(), spyRate, spyRate, TimeUnit.SECONDS);
+ spyPool.scheduleAtFixedRate(conductor.getMessageIntelligenceAgency(), SPY_RATE, SPY_RATE, TimeUnit.SECONDS);
}
@Override
@Override
public ListenableFuture<Void> onClusterRoleChange(final DeviceInfo deviceInfo, final OfpRole role) {
- DeviceContext deviceContext = conductor.getDeviceContext(deviceInfo);
+ DeviceContext deviceContext = deviceContexts.get(deviceInfo);
LOG.trace("onClusterRoleChange {} for node:", role, deviceInfo.getNodeId());
if (OfpRole.BECOMEMASTER.equals(role)) {
return onDeviceTakeClusterLeadership(deviceInfo);
}
@Override
- public void registerDeviceSynchronizeListeners(final DeviceSynchronizeListener deviceSynchronizeListener) {
+ public void addDeviceSynchronizeListener(final DeviceSynchronizeListener deviceSynchronizeListener) {
this.deviceSynchronizedListeners.add(deviceSynchronizeListener);
}
}
@Override
- public void registerDeviceValidListeners(final DeviceValidListener deviceValidListener) {
+ public void addDeviceValidListener(final DeviceValidListener deviceValidListener) {
this.deviceValidListeners.add(deviceValidListener);
}
}
}
+ @Override
+ public void setIsNotificationFlowRemovedOff(boolean isNotificationFlowRemovedOff) {
+ this.isNotificationFlowRemovedOff = isNotificationFlowRemovedOff;
+ }
+
+ @Override
+ public boolean getIsNotificationFlowRemovedOff() {
+ return this.isNotificationFlowRemovedOff;
+ }
+
private ListenableFuture<Void> onDeviceTakeClusterLeadership(final DeviceInfo deviceInfo) {
LOG.trace("onDeviceTakeClusterLeadership for node: {}", deviceInfo.getNodeId());
/* validation */
LOG.warn(errMsg);
return Futures.immediateFailedFuture(new IllegalStateException(errMsg));
}
- DeviceContext deviceContext = conductor.getDeviceContext(deviceInfo);
+ DeviceContext deviceContext = deviceContexts.get(deviceInfo);
/* Prepare init info collecting */
notifyDeviceSynchronizeListeners(deviceInfo, false);
((DeviceContextImpl)deviceContext).getTransactionChainManager().activateTransactionManager();
+ ((DeviceContextImpl)deviceContext).getTransactionChainManager().initialSubmitWriteTransaction();
/* Init Collecting NodeInfo */
final ListenableFuture<Void> initCollectingDeviceInfo = DeviceInitializationUtils.initializeNodeInformation(
- deviceContext, switchFeaturesMandatory);
+ deviceContext, switchFeaturesMandatory, convertorExecutor);
/* Init Collecting StatInfo */
final ListenableFuture<Boolean> statPollFuture = Futures.transform(initCollectingDeviceInfo,
new AsyncFunction<Void, Boolean>() {
}
});
- return Futures.transform(statPollFuture, new Function<Boolean, Void>() {
+ return Futures.transform(statPollFuture, getInitialDeviceInformation(deviceContext));
+ }
- @Override
- public Void apply(final Boolean input) {
- if (ConnectionContext.CONNECTION_STATE.RIP.equals(conductor.gainConnectionStateSafely(deviceInfo))) {
- final String errMsg = String.format("We lost connection for Device %s, context has to be closed.",
- deviceInfo.getNodeId());
- LOG.warn(errMsg);
- throw new IllegalStateException(errMsg);
- }
- if (!input) {
- final String errMsg = String.format("Get Initial Device %s information fails",
- deviceInfo.getNodeId());
- LOG.warn(errMsg);
- throw new IllegalStateException(errMsg);
- }
- LOG.debug("Get Initial Device {} information is successful", deviceInfo.getNodeId());
- notifyDeviceSynchronizeListeners(deviceInfo, true);
- ((DeviceContextImpl)deviceContext).getTransactionChainManager().initialSubmitWriteTransaction();
- deviceContext.getDeviceState().setStatisticsPollingEnabledProp(true);
- return null;
+ private Function<Boolean, Void> getInitialDeviceInformation(final DeviceContext deviceContext) {
+ return input -> {
+ if (ConnectionContext.CONNECTION_STATE.RIP.equals(
+ conductor.gainConnectionStateSafely(deviceContext.getDeviceInfo())
+ )) {
+ final String errMsg =
+ String.format("We lost connection for Device %s, context has to be closed.",
+ deviceContext.getDeviceInfo().getNodeId());
+ LOG.warn(errMsg);
+ throw new IllegalStateException(errMsg);
+ }
+
+ if (input == null || !input) {
+ final String errMsg =
+ String.format("Get Initial Device %s information fails",
+ deviceContext.getDeviceInfo().getNodeId());
+ LOG.warn(errMsg);
+ throw new IllegalStateException(errMsg);
}
- });
+ LOG.debug("Get Initial Device {} information is successful",
+ deviceContext.getDeviceInfo().getNodeId());
+ notifyDeviceSynchronizeListeners(deviceContext.getDeviceInfo(), true);
+ deviceContext.getDeviceState().setStatisticsPollingEnabledProp(true);
+ return null;
+ };
}
}
Code Review / openflowplugin.git / blobdiff