/**
- * Copyright (c) 2013-2014 Cisco Systems, Inc. and others. All rights reserved.
+ * Copyright (c) 2013, 2015 Cisco Systems, Inc. and others. All rights reserved.
*
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
package org.opendaylight.openflowplugin.openflow.md.core;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.Futures;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-
import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
import org.opendaylight.openflowjava.protocol.api.connection.ConnectionReadyListener;
import org.opendaylight.openflowplugin.api.OFConstants;
+import org.opendaylight.openflowplugin.api.openflow.connection.HandshakeContext;
import org.opendaylight.openflowplugin.api.openflow.md.core.ConnectionConductor;
import org.opendaylight.openflowplugin.api.openflow.md.core.ErrorHandler;
import org.opendaylight.openflowplugin.api.openflow.md.core.HandshakeListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.util.concurrent.Futures;
-
/**
* @author mirehak
*/
private int conductorId;
private int ingressMaxQueueSize;
+ private HandshakeContext handshakeContext;
/**
* @param connectionAdapter
/**
* @param connectionAdapter
- * @param ingressMaxQueueSize
- * ingress queue limit (blocking)
+ * @param ingressMaxQueueSize ingress queue limit (blocking)
*/
public ConnectionConductorImpl(ConnectionAdapter connectionAdapter,
- int ingressMaxQueueSize) {
+ int ingressMaxQueueSize) {
this.connectionAdapter = connectionAdapter;
this.ingressMaxQueueSize = ingressMaxQueueSize;
conductorState = CONDUCTOR_STATE.HANDSHAKING;
hsPool = new ThreadPoolLoggingExecutor(handshakeThreadLimit,
handshakeThreadLimit, 0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(), "OFHandshake-"
- + conductorId);
+ + conductorId);
connectionAdapter.setMessageListener(this);
connectionAdapter.setSystemListener(this);
}
/**
- * @param errorHandler
- * the errorHandler to set
+ * @param errorHandler the errorHandler to set
*/
@Override
public void setErrorHandler(ErrorHandler errorHandler) {
/**
* @param message
- * @param queueType
- * enqueue type
+ * @param queueType enqueue type
*/
private void enqueueMessage(OfHeader message, QueueType queueType) {
queue.push(message, this, queueType);
@Override
public void onPortStatusMessage(PortStatusMessage message) {
- processPortStatusMsg(message);
- enqueueMessage(message);
+ try {
+ processPortStatusMsg(message);
+ } finally {
+ enqueueMessage(message);
+ }
}
protected void processPortStatusMsg(PortStatus msg) {
"can't get bandwidth info from port: {}, aborting port update",
msg.toString());
} else {
- this.getSessionContext().getPhysicalPorts().put(portNumber, msg);
- this.getSessionContext().getPortsBandwidth()
- .put(portNumber, portBandwidth);
+ if (null != this.sessionContext) {
+ //FIXME these two properties are never used in code
+ this.getSessionContext().getPhysicalPorts().put(portNumber, msg);
+ this.getSessionContext().getPortsBandwidth()
+ .put(portNumber, portBandwidth);
+ } else {
+ LOG.warn("Trying to process update port message before session context was created.");
+ }
}
}
}
/**
- * @param conductorState
- * the connectionState to set
+ * @param conductorState the connectionState to set
*/
@Override
public void setConductorState(CONDUCTOR_STATE conductorState) {
*/
protected void checkState(CONDUCTOR_STATE expectedState) {
if (!conductorState.equals(expectedState)) {
+ LOG.warn("State of connection to switch {} is not correct, "
+ + "terminating the connection", connectionAdapter.getRemoteAddress());
throw new IllegalStateException("Expected state: " + expectedState
+ ", actual state:" + conductorState);
}
public void onConnectionReady() {
LOG.debug("connection is ready-to-use");
if (!firstHelloProcessed) {
+ checkState(CONDUCTOR_STATE.HANDSHAKING);
HandshakeStepWrapper handshakeStepWrapper = new HandshakeStepWrapper(
null, handshakeManager, connectionAdapter);
hsPool.execute(handshakeStepWrapper);
@Override
public void onHandshakeSuccessfull(GetFeaturesOutput featureOutput,
- Short negotiatedVersion) {
+ Short negotiatedVersion) {
postHandshakeBasic(featureOutput, negotiatedVersion);
// post-handshake actions
if (version == OFConstants.OFP_VERSION_1_3) {
requestPorts();
- requestGroupFeatures();
- requestMeterFeatures();
}
requestDesc();
/**
* used by tests
- *
+ *
* @param featureOutput
* @param negotiatedVersion
*/
protected void postHandshakeBasic(GetFeaturesOutput featureOutput,
- Short negotiatedVersion) {
+ Short negotiatedVersion) {
version = negotiatedVersion;
if (version == OFConstants.OFP_VERSION_1_0) {
// Because the GetFeaturesOutput contains information about the port
}
/**
- * @param isBitmapNegotiationEnable
- * the isBitmapNegotiationEnable to set
+ * @param isBitmapNegotiationEnable the isBitmapNegotiationEnable to set
*/
public void setBitmapNegotiationEnable(boolean isBitmapNegotiationEnable) {
this.isBitmapNegotiationEnable = isBitmapNegotiationEnable;
}
- protected void shutdownPool() {
- hsPool.shutdownNow();
- LOG.debug("pool is terminated: {}", hsPool.isTerminated());
+ @Override
+ public void setId(int conductorId) {
+ this.conductorId = conductorId;
+ }
+
+ @Override
+ public void close() {
+ conductorState = CONDUCTOR_STATE.RIP;
+ if (handshakeContext != null) {
+ try {
+ handshakeContext.close();
+ } catch (Exception e) {
+ LOG.warn("Closing handshake context failed: {}", e.getMessage());
+ LOG.debug("Detail in hanshake context close:", e);
+ }
+ } else {
+ //This condition will occure when Old Helium openflowplugin implementation will be used.
+ shutdownPoolPolitely();
+ }
}
- protected void shutdownPoolPolitely() {
+ private void shutdownPoolPolitely() {
+ LOG.debug("Terminating handshake pool for node {}", connectionAdapter.getRemoteAddress());
hsPool.shutdown();
try {
hsPool.awaitTermination(1, TimeUnit.SECONDS);
} catch (InterruptedException e) {
- LOG.info("Error while awaiting termination on pool. Will use shutdownNow method.");
- shutdownPool();
+ LOG.debug("Error while awaiting termination of pool. Will force shutdown now.");
+ } finally {
+ hsPool.purge();
+ if (!hsPool.isTerminated()) {
+ hsPool.shutdownNow();
+ }
+ LOG.debug("is handshake pool for node {} is terminated : {}",
+ connectionAdapter.getRemoteAddress(), hsPool.isTerminated());
}
- hsPool.purge();
- LOG.debug("pool is terminated: {}", hsPool.isTerminated());
}
@Override
- public void setId(int conductorId) {
- this.conductorId = conductorId;
+ public void setHandshakeContext(HandshakeContext handshakeContext) {
+ this.handshakeContext = handshakeContext;
}
- @Override
- public void close() {
- shutdownPoolPolitely();
- conductorState = CONDUCTOR_STATE.RIP;
+ @VisibleForTesting
+ ThreadPoolExecutor getHsPool() {
+ return hsPool;
}
}
Code Review / openflowplugin.git / blobdiff