package org.opendaylight.openflowplugin.openflow.md.core;
-import java.util.Arrays;
-import java.util.Collection;
import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
import org.opendaylight.openflowjava.protocol.api.connection.ConnectionReadyListener;
import org.opendaylight.openflowplugin.openflow.md.core.session.SessionContext;
import org.opendaylight.openflowplugin.openflow.md.core.session.SessionManager;
import org.opendaylight.openflowplugin.openflow.md.queue.QueueKeeper;
-import org.opendaylight.openflowplugin.openflow.md.queue.QueueKeeperLightImpl;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoInputBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoOutput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoReplyInputBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.DisconnectEvent;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.SwitchIdleEvent;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.SystemNotificationsListener;
-import org.opendaylight.yangtools.yang.binding.DataObject;
import org.opendaylight.yangtools.yang.common.RpcError;
import org.opendaylight.yangtools.yang.common.RpcResult;
import org.slf4j.Logger;
* BitMaps from switches.
*/
private static final boolean isBitmapNegotiationEnable = true;
- private LinkedBlockingQueue<Exception> errorQueue = new LinkedBlockingQueue<>();
+ private ErrorHandler errorHandler;
- protected final ConnectionAdapter connectionAdapter;
+ private final ConnectionAdapter connectionAdapter;
private ConnectionConductor.CONDUCTOR_STATE conductorState;
private Short version;
private SessionContext sessionContext;
- private Map<Class<? extends DataObject>, Collection<IMDMessageListener>> listenerMapping;
-
protected boolean isFirstHelloNegotiation = true;
// TODO: use appropriate interface instead of Object
public ConnectionConductorImpl(ConnectionAdapter connectionAdapter) {
this.connectionAdapter = connectionAdapter;
conductorState = CONDUCTOR_STATE.HANDSHAKING;
- new Thread(new ErrorQueueHandler(errorQueue)).start();
}
@Override
this.queueKeeper = queueKeeper;
}
+ /**
+ * @param errorHandler the errorHandler to set
+ */
+ @Override
+ public void setErrorHandler(ErrorHandler errorHandler) {
+ this.errorHandler = errorHandler;
+ }
/**
* send first hello message to switch
if (isBitmapNegotiationEnable) {
helloInput = MessageFactory.createHelloInput(highestVersion, helloXid, ConnectionConductor.versionOrder);
LOG.debug("sending first hello message: vertsion header={} , version bitmap={}",
- highestVersion, helloInput.getElements());
+ highestVersion, MessageFactory.digVersions(helloInput.getElements()));
} else {
helloInput = MessageFactory.createHelloInput(highestVersion, helloXid);
LOG.debug("sending first hello message: version header={} ", highestVersion);
LOG.debug("FIRST HELLO sent.");
} catch (Throwable e) {
LOG.debug("FIRST HELLO sending failed.");
- handleException(e);
+ errorHandler.handleException(e, getSessionContext());
}
}
builder.setXid(echoRequestMessage.getXid());
builder.setData(echoRequestMessage.getData());
- connectionAdapter.echoReply(builder.build());
+ getConnectionAdapter().echoReply(builder.build());
}
}).start();
}
@Override
public void onErrorMessage(ErrorMessage errorMessage) {
queueKeeper.push(ErrorMessage.class, errorMessage, this);
- // notifyListeners(ErrorMessage.class, errorMessage);
}
@Override
public void onExperimenterMessage(ExperimenterMessage experimenterMessage) {
queueKeeper.push(ExperimenterMessage.class, experimenterMessage, this);
-// notifyListeners(ExperimenterMessage.class, experimenterMessage);
}
@Override
public void onFlowRemovedMessage(FlowRemovedMessage message) {
- notifyListeners(FlowRemovedMessage.class, message);
+ queueKeeper.push(FlowRemovedMessage.class, message, this);
}
List<Elements> elements = hello.getElements();
Long xid = hello.getXid();
Short proposedVersion;
- LOG.debug("Hello message version={} and bitmap={}", remoteVersion, elements);
+ LOG.debug("Hello message version={} and bitmap={}", remoteVersion, MessageFactory.digVersions(elements));
try {
// find the version from header version field
proposedVersion = proposeVersion(remoteVersion);
} catch (IllegalArgumentException e) {
- handleException(e);
- connectionAdapter.disconnect();
+ errorHandler.handleException(e, getSessionContext());
+ getConnectionAdapter().disconnect();
return;
}
// version in bitmap
proposedVersion = proposeBitmapVersion(elements);
} catch (IllegalArgumentException ex) {
- handleException(ex);
- connectionAdapter.disconnect();
+ errorHandler.handleException(ex, getSessionContext());
+ getConnectionAdapter().disconnect();
return;
}
LOG.debug("sending helloReply for common bitmap version : {}", proposedVersion);
} else {
// terminate the connection.
LOG.debug("Version negotiation failed. unsupported version : {}", remoteVersion);
- connectionAdapter.disconnect();
+ getConnectionAdapter().disconnect();
}
}
}
result = connectionAdapter.hello(helloMsg).get(getMaxTimeout(), getMaxTimeoutUnit());
smokeRpc(result);
} catch (Throwable e) {
- handleException(e);
+ errorHandler.handleException(e, getSessionContext());
}
}
OFSessionUtil.registerSession(this,
featureOutput, version);
- this.setListenerMapping(OFSessionUtil.getListenersMap());
LOG.info("handshake SETTLED: datapathId={}, auxiliaryId={}", featureOutput.getDatapathId(), featureOutput.getAuxiliaryId());
} catch (Throwable e) {
//handshake failed
LOG.error("issuing disconnect during handshake, reason: "+e.getMessage());
- handleException(e);
+ errorHandler.handleException(e, getSessionContext());
disconnect();
}
}
/**
* @return rpc-response timeout in [ms]
*/
- private long getMaxTimeout() {
+ protected long getMaxTimeout() {
// TODO:: get from configuration
return 2000;
}
/**
* @return milliseconds
*/
- private TimeUnit getMaxTimeoutUnit() {
+ protected TimeUnit getMaxTimeoutUnit() {
// TODO:: get from configuration
return TimeUnit.MILLISECONDS;
}
-
- /**
- * @param e
- */
- protected void handleException(Throwable e) {
- String sessionKeyId = null;
- if (getSessionContext() != null) {
- sessionKeyId = Arrays.toString(getSessionContext().getSessionKey().getId());
- }
-
- Exception causeAndThread = new Exception(
- "IN THREAD: "+Thread.currentThread().getName() +
- "; session:"+sessionKeyId, e);
- try {
- errorQueue.put(causeAndThread);
- } catch (InterruptedException e1) {
- LOG.error(e1.getMessage(), e1);
- }
- }
-
@Override
- public void onMultipartReplyMessage(MultipartReplyMessage arg0) {
- // TODO Auto-generated method stub
+ public void onMultipartReplyMessage(MultipartReplyMessage message) {
+ queueKeeper.push(MultipartReplyMessage.class, message, this);
}
@Override
- public void onMultipartRequestMessage(MultipartRequestMessage arg0) {
- // TODO Auto-generated method stub
+ public void onMultipartRequestMessage(MultipartRequestMessage message) {
+ queueKeeper.push(MultipartRequestMessage.class, message, this);
}
@Override
public void onPacketInMessage(PacketInMessage message) {
- notifyListeners(PacketInMessage.class, message);
+ queueKeeper.push(PacketInMessage.class, message, this);
}
@Override
public void onPortStatusMessage(PortStatusMessage message) {
this.getSessionContext().processPortStatusMsg(message);
- notifyListeners(PortStatusMessage.class, message);
+ queueKeeper.push(PortStatusMessage.class, message, this);
}
@Override
public void onSwitchIdleEvent(SwitchIdleEvent notification) {
- if (!CONDUCTOR_STATE.WORKING.equals(conductorState)) {
- // idle state in any other conductorState than WORKING means real
- // problem and wont be handled by echoReply, but disconnection
- disconnect();
- OFSessionUtil.getSessionManager().invalidateOnDisconnect(this);
- } else {
- LOG.debug("first idle state occured");
- EchoInputBuilder builder = new EchoInputBuilder();
- builder.setVersion(version);
- // TODO: get xid from sessionContext
- builder.setXid(42L);
-
- Future<RpcResult<EchoOutput>> echoReplyFuture = connectionAdapter
- .echo(builder.build());
-
- try {
- // TODO: read timeout from config
- RpcResult<EchoOutput> echoReplyValue = echoReplyFuture.get(getMaxTimeout(),
- getMaxTimeoutUnit());
- if (echoReplyValue.isSuccessful()) {
- conductorState = CONDUCTOR_STATE.WORKING;
+ new Thread(new Runnable() {
+ @Override
+ public void run() {
+ if (!CONDUCTOR_STATE.WORKING.equals(getConductorState())) {
+ // idle state in any other conductorState than WORKING means real
+ // problem and wont be handled by echoReply, but disconnection
+ disconnect();
+ OFSessionUtil.getSessionManager().invalidateOnDisconnect(ConnectionConductorImpl.this);
} else {
- for (RpcError replyError : echoReplyValue.getErrors()) {
- Throwable cause = replyError.getCause();
- LOG.error(
- "while receiving echoReply in TIMEOUTING state: "
- + cause.getMessage(), cause);
+ LOG.debug("first idle state occured");
+ EchoInputBuilder builder = new EchoInputBuilder();
+ builder.setVersion(getVersion());
+ builder.setXid(getSessionContext().getNextXid());
+
+ Future<RpcResult<EchoOutput>> echoReplyFuture = getConnectionAdapter()
+ .echo(builder.build());
+
+ try {
+ RpcResult<EchoOutput> echoReplyValue = echoReplyFuture.get(getMaxTimeout(),
+ getMaxTimeoutUnit());
+ if (echoReplyValue.isSuccessful()) {
+ setConductorState(CONDUCTOR_STATE.WORKING);
+ } else {
+ for (RpcError replyError : echoReplyValue.getErrors()) {
+ Throwable cause = replyError.getCause();
+ LOG.error(
+ "while receiving echoReply in TIMEOUTING state: "
+ + cause.getMessage(), cause);
+ }
+ //switch issue occurred
+ throw new Exception("switch issue occurred");
+ }
+ } catch (Exception e) {
+ LOG.error("while waiting for echoReply in TIMEOUTING state: "
+ + e.getMessage(), e);
+ //switch is not responding
+ disconnect();
+ OFSessionUtil.getSessionManager().invalidateOnDisconnect(ConnectionConductorImpl.this);
}
- //switch issue occurred
- throw new Exception("switch issue occurred");
}
- } catch (Exception e) {
- LOG.error("while waiting for echoReply in TIMEOUTING state: "
- + e.getMessage(), e);
- //switch is not responding
- disconnect();
- OFSessionUtil.getSessionManager().invalidateOnDisconnect(this);
}
- }
+
+ }).start();
}
/**
return sessionContext;
}
- /**
- * @param listenerMapping the listenerMapping to set
- */
- public void setListenerMapping(
- Map<Class<? extends DataObject>, Collection<IMDMessageListener>> listenerMapping) {
- this.listenerMapping = listenerMapping;
- }
-
- /**
- * @param messageType
- * @param message
- * @deprecated use {@link QueueKeeper} strategy
- */
- @Deprecated
- private void notifyListeners(Class<? extends DataObject> messageType, DataObject message) {
- Collection<IMDMessageListener> listeners = listenerMapping.get(messageType);
- if (listeners != null) {
- for (IMDMessageListener listener : listeners) {
- // Pass cookie only for PACKT_IN
- if ( messageType.equals("PacketInMessage.class")){
- listener.receive(this.getAuxiliaryKey(), this.getSessionContext(), message);
- } else {
- listener.receive(null, this.getSessionContext(), message);
- }
- }
- } else {
- LOG.warn("No listeners for this message Type {}", messageType);
- }
- }
-
@Override
public ConnectionAdapter getConnectionAdapter() {
return connectionAdapter;
@Override
public void onConnectionReady() {
LOG.debug("connection is ready-to-use");
- //TODO: fire first helloMessage
new Thread(new Runnable() {
@Override
public void run() {
protected static final Logger LOG = LoggerFactory
.getLogger(ConnectionConductorImplTest.class);
+ /** in [ms] */
+ private final int maxProcessingTimeout = 500;
+
protected ConnectionAdapterStackImpl adapter;
private ConnectionConductorImpl connectionConductor;
private MDController controller;
popListener = new PopListenerCountingImpl<>();
queueKeeper = new QueueKeeperLightImpl<>();
- queueKeeper.setListenerMapping(assembleListenerMapping());
queueKeeper.init();
queueKeeper.addPopListener(popListener);
connectionConductor.init();
controller = new MDController();
controller.init();
+ queueKeeper.setListenerMapping(controller.getMessageListeners());
eventPlan = new Stack<>();
adapter.setEventPlan(eventPlan);
adapter.setProceedTimeout(5000L);
adapter.checkListeners();
+
+ controller.getMessageListeners().putAll(assembleListenerMapping());
}
/**
eventPlan.add(0, EventFactory.createDefaultNotificationEvent(42L,
EventFactory.DEFAULT_VERSION, builder1));
- connectionConductor.setListenerMapping(assembleListenerMapping());
-
executeLater();
Runnable sendExperimenterCmd = new Runnable() {
FlowRemovedMessageBuilder builder1 = new FlowRemovedMessageBuilder();
builder1.setXid(1L);
connectionConductor.onFlowRemovedMessage(builder1.build());
+ synchronized (popListener) {
+ popListener.wait(maxProcessingTimeout);
+ }
Assert.assertEquals(1, flowremovedMessageCounter);
builder1.setXid(2L);
connectionConductor.onFlowRemovedMessage(builder1.build());
+ synchronized (popListener) {
+ popListener.wait(maxProcessingTimeout);
+ }
Assert.assertEquals(2, flowremovedMessageCounter);
}
PacketInMessageBuilder builder1 = new PacketInMessageBuilder();
builder1.setBufferId((long)1);
connectionConductor.onPacketInMessage(builder1.build());
+ synchronized (popListener) {
+ popListener.wait(maxProcessingTimeout);
+ }
Assert.assertEquals(1, packetinMessageCounter);
builder1.setBufferId((long)2);
connectionConductor.onPacketInMessage(builder1.build());
+ synchronized (popListener) {
+ popListener.wait(maxProcessingTimeout);
+ }
Assert.assertEquals(2, packetinMessageCounter);
}
PortFeatures features = new PortFeatures(true,false,false,false,false,false,false,false,false,false,false,false,false,false,false,false);
builder1.setPortNo(90L).setReason(PortReason.OFPPRADD).setCurrentFeatures(features);
connectionConductor.onPortStatusMessage(builder1.build());
+ synchronized (popListener) {
+ popListener.wait(maxProcessingTimeout);
+ }
Assert.assertEquals(1, portstatusAddMessageCounter);
builder1.setPortNo(90L).setReason(PortReason.OFPPRMODIFY).setCurrentFeatures(features);
connectionConductor.onPortStatusMessage(builder1.build());
+ synchronized (popListener) {
+ popListener.wait(maxProcessingTimeout);
+ }
Assert.assertEquals(1, portstatusModifyMessageCounter);
builder1.setPortNo(90L).setReason(PortReason.OFPPRDELETE).setCurrentFeatures(features);
connectionConductor.onPortStatusMessage(builder1.build());
+ synchronized (popListener) {
+ popListener.wait(maxProcessingTimeout);
+ }
Assert.assertEquals(1, portstatusDeleteMessageCounter);
}
*/
@Test
public void testOnExperimenterMessage() throws InterruptedException {
- final int maxProcessingTimeout = 500;
-
- connectionConductor.setListenerMapping(assembleListenerMapping());
ExperimenterMessageBuilder builder1 = new ExperimenterMessageBuilder();
builder1.setExperimenter(84L).setExpType(4L);
connectionConductor.onExperimenterMessage(builder1.build());
*/
@Test
public void testOnErrorMessage() throws InterruptedException {
- final int maxProcessingTimeout = 500;
- connectionConductor.setListenerMapping(assembleListenerMapping());
ErrorMessageBuilder builder1 = new ErrorMessageBuilder();
builder1.setCode(100);
connectionConductor.onErrorMessage(builder1.build());
}
/**
- * @return listener mapping
+ * @return listener mapping for :
+ * <ul>
+ * <li>experimenter</li>
+ * <li>error</li>
+ * </ul>
*/
private Map<Class<? extends DataObject>, Collection<IMDMessageListener>> assembleListenerMapping() {
IMDMessageListener objEms = new ExperimenterMessageService() ;
Code Review / openflowplugin.git / commitdiff