import org.w3c.dom.Document;
import org.w3c.dom.NodeList;
-public abstract class AbstractNetconfSessionNegotiator<P extends NetconfSessionPreferences, S extends AbstractNetconfSession<S, L>, L extends NetconfSessionListener<S>>
+public abstract class AbstractNetconfSessionNegotiator<P extends NetconfSessionPreferences,
+ S extends AbstractNetconfSession<S, L>, L extends NetconfSessionListener<S>>
extends AbstractSessionNegotiator<NetconfHelloMessage, S> {
private static final Logger LOG = LoggerFactory.getLogger(AbstractNetconfSessionNegotiator.class);
private Timeout timeout;
/**
- * Possible states for Finite State Machine
+ * Possible states for Finite State Machine.
*/
protected enum State {
IDLE, OPEN_WAIT, FAILED, ESTABLISHED
private final Timer timer;
private final long connectionTimeoutMillis;
- protected AbstractNetconfSessionNegotiator(final P sessionPreferences, final Promise<S> promise, final Channel channel, final Timer timer,
- final L sessionListener, final long connectionTimeoutMillis) {
+ protected AbstractNetconfSessionNegotiator(final P sessionPreferences, final Promise<S> promise,
+ final Channel channel, final Timer timer,
+ final L sessionListener, final long connectionTimeoutMillis) {
super(promise, channel);
this.sessionPreferences = sessionPreferences;
this.promise = promise;
@Override
protected final void startNegotiation() {
- final Optional<SslHandler> sslHandler = getSslHandler(channel);
- if (sslHandler.isPresent()) {
- Future<Channel> future = sslHandler.get().handshakeFuture();
- future.addListener(new GenericFutureListener<Future<? super Channel>>() {
- @Override
- public void operationComplete(final Future<? super Channel> future) {
- Preconditions.checkState(future.isSuccess(), "Ssl handshake was not successful");
- LOG.debug("Ssl handshake complete");
- start();
- }
- });
+ if (ifNegotiatedAlready()) {
+ LOG.debug("Negotiation on channel {} already started", channel);
} else {
- start();
+ final Optional<SslHandler> sslHandler = getSslHandler(channel);
+ if (sslHandler.isPresent()) {
+ Future<Channel> future = sslHandler.get().handshakeFuture();
+ future.addListener(new GenericFutureListener<Future<? super Channel>>() {
+ @Override
+ public void operationComplete(final Future<? super Channel> future) {
+ Preconditions.checkState(future.isSuccess(), "Ssl handshake was not successful");
+ LOG.debug("Ssl handshake complete");
+ start();
+ }
+ });
+ } else {
+ start();
+ }
}
}
+ protected final boolean ifNegotiatedAlready() {
+ // Indicates whether negotiation already started
+ return this.state != State.IDLE;
+ }
+
private static Optional<SslHandler> getSslHandler(final Channel channel) {
final SslHandler sslHandler = channel.pipeline().get(SslHandler.class);
- return sslHandler == null ? Optional.<SslHandler> absent() : Optional.of(sslHandler);
+ return sslHandler == null ? Optional.<SslHandler>absent() : Optional.of(sslHandler);
}
public P getSessionPreferences() {
timeout = this.timer.newTimeout(new TimerTask() {
@Override
+ @SuppressWarnings("checkstyle:hiddenField")
public void run(final Timeout timeout) {
synchronized (this) {
if (state != State.ESTABLISHED) {
// Do not fail negotiation if promise is done or canceled
// It would result in setting result of the promise second time and that throws exception
- if (isPromiseFinished() == false) {
+ if (!isPromiseFinished()) {
LOG.warn("Netconf session was not established after {}", connectionTimeoutMillis);
changeState(State.FAILED);
channel.close().addListener(new GenericFutureListener<ChannelFuture>() {
@Override
public void operationComplete(final ChannelFuture future) throws Exception {
- if(future.isSuccess()) {
+ if (future.isSuccess()) {
LOG.debug("Channel {} closed: success", future.channel());
} else {
LOG.warn("Channel {} closed: fail", future.channel());
}
});
}
- } else if(channel.isOpen()) {
+ } else if (channel.isOpen()) {
channel.pipeline().remove(NAME_OF_EXCEPTION_HANDLER);
}
}
}
private void cancelTimeout() {
- if(timeout!=null) {
+ if (timeout != null) {
timeout.cancel();
}
}
- protected final S getSessionForHelloMessage(final NetconfHelloMessage netconfMessage) throws NetconfDocumentedException {
+ protected final S getSessionForHelloMessage(final NetconfHelloMessage netconfMessage)
+ throws NetconfDocumentedException {
Preconditions.checkNotNull(netconfMessage, "netconfMessage");
final Document doc = netconfMessage.getDocument();
}
/**
- * Insert chunk framing handlers into the pipeline
+ * Insert chunk framing handlers into the pipeline.
*/
private void insertChunkFramingToPipeline() {
replaceChannelHandler(channel, AbstractChannelInitializer.NETCONF_MESSAGE_FRAME_ENCODER,
/**
* Remove special inbound handler for hello message. Insert regular netconf xml message (en|de)coders.
*
+ * <p>
* Inbound hello message handler should be kept until negotiation is successful
* It caches any non-hello messages while negotiation is still in progress
*/
protected final void replaceHelloMessageInboundHandler(final S session) {
- ChannelHandler helloMessageHandler = replaceChannelHandler(channel, AbstractChannelInitializer.NETCONF_MESSAGE_DECODER, new NetconfXMLToMessageDecoder());
+ ChannelHandler helloMessageHandler = replaceChannelHandler(channel,
+ AbstractChannelInitializer.NETCONF_MESSAGE_DECODER, new NetconfXMLToMessageDecoder());
Preconditions.checkState(helloMessageHandler instanceof NetconfXMLToHelloMessageDecoder,
"Pipeline handlers misplaced on session: %s, pipeline: %s", session, channel.pipeline());
((NetconfXMLToHelloMessageDecoder) helloMessageHandler).getPostHelloNetconfMessages();
// Process messages received during negotiation
- // The hello message handler does not have to be synchronized, since it is always call from the same thread by netty
+ // The hello message handler does not have to be synchronized,
+ // since it is always call from the same thread by netty.
// It means, we are now using the thread now
for (NetconfMessage message : netconfMessagesFromNegotiation) {
session.handleMessage(message);
* Remove special outbound handler for hello message. Insert regular netconf xml message (en|de)coders.
*/
private void replaceHelloMessageOutboundHandler() {
- replaceChannelHandler(channel, AbstractChannelInitializer.NETCONF_MESSAGE_ENCODER, new NetconfMessageToXMLEncoder());
+ replaceChannelHandler(channel, AbstractChannelInitializer.NETCONF_MESSAGE_ENCODER,
+ new NetconfMessageToXMLEncoder());
}
- private static ChannelHandler replaceChannelHandler(final Channel channel, final String handlerKey, final ChannelHandler decoder) {
+ private static ChannelHandler replaceChannelHandler(final Channel channel, final String handlerKey,
+ final ChannelHandler decoder) {
return channel.pipeline().replace(handlerKey, handlerKey, decoder);
}
- protected abstract S getSession(L sessionListener, Channel channel, NetconfHelloMessage message) throws NetconfDocumentedException;
+ @SuppressWarnings("checkstyle:hiddenField")
+ protected abstract S getSession(L sessionListener, Channel channel, NetconfHelloMessage message)
+ throws NetconfDocumentedException;
private synchronized void changeState(final State newState) {
LOG.debug("Changing state from : {} to : {} for channel: {}", state, newState, channel);
- Preconditions.checkState(isStateChangePermitted(state, newState), "Cannot change state from %s to %s for chanel %s", state,
- newState, channel);
+ Preconditions.checkState(isStateChangePermitted(state, newState),
+ "Cannot change state from %s to %s for chanel %s", state, newState, channel);
this.state = newState;
}
}
/**
- * Handler to catch exceptions in pipeline during negotiation
+ * Handler to catch exceptions in pipeline during negotiation.
*/
private final class ExceptionHandlingInboundChannelHandler extends ChannelInboundHandlerAdapter {
@Override
Code Review / netconf.git / blobdiff