Merge "BUG 1853 : Clustered Data Store causes Out of Memory"
[controller.git] / opendaylight / netconf / netconf-client / src / main / java / org / opendaylight / controller / netconf / client / NetconfClientSessionNegotiator.java
index f8f73fc8e5296b9534e0759bc43f24bbf493521e..cbbee1f65508353b117bbd21db22b523b4244694 100644 (file)
@@ -8,6 +8,9 @@
 
 package org.opendaylight.controller.netconf.client;
 
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
@@ -15,15 +18,22 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.util.Timer;
 import io.netty.util.concurrent.Promise;
+
+import java.util.Collection;
+
+import javax.xml.xpath.XPathConstants;
+import javax.xml.xpath.XPathExpression;
+
 import org.opendaylight.controller.netconf.api.NetconfClientSessionPreferences;
 import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
 import org.opendaylight.controller.netconf.api.NetconfMessage;
-import org.opendaylight.controller.netconf.util.AbstractChannelInitializer;
-import org.opendaylight.controller.netconf.util.AbstractNetconfSessionNegotiator;
+import org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants;
+import org.opendaylight.controller.netconf.nettyutil.AbstractChannelInitializer;
+import org.opendaylight.controller.netconf.nettyutil.AbstractNetconfSessionNegotiator;
+import org.opendaylight.controller.netconf.nettyutil.handler.exi.NetconfStartExiMessage;
 import org.opendaylight.controller.netconf.util.messages.NetconfHelloMessage;
 import org.opendaylight.controller.netconf.util.messages.NetconfMessageUtil;
 import org.opendaylight.controller.netconf.util.xml.XMLNetconfUtil;
-import org.opendaylight.controller.netconf.util.xml.XmlNetconfConstants;
 import org.opendaylight.controller.netconf.util.xml.XmlUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -31,9 +41,6 @@ import org.w3c.dom.Document;
 import org.w3c.dom.Node;
 import org.w3c.dom.NodeList;
 
-import javax.xml.xpath.XPathConstants;
-import javax.xml.xpath.XPathExpression;
-
 public class NetconfClientSessionNegotiator extends
         AbstractNetconfSessionNegotiator<NetconfClientSessionPreferences, NetconfClientSession, NetconfClientSessionListener>
 {
@@ -44,30 +51,58 @@ public class NetconfClientSessionNegotiator extends
 
     private static final String EXI_1_0_CAPABILITY_MARKER = "exi:1.0";
 
-    protected NetconfClientSessionNegotiator(NetconfClientSessionPreferences sessionPreferences,
-                                             Promise<NetconfClientSession> promise,
-                                             Channel channel,
-                                             Timer timer,
-                                             NetconfClientSessionListener sessionListener,
-                                             long connectionTimeoutMillis) {
+    protected NetconfClientSessionNegotiator(final NetconfClientSessionPreferences sessionPreferences,
+                                             final Promise<NetconfClientSession> promise,
+                                             final Channel channel,
+                                             final Timer timer,
+                                             final NetconfClientSessionListener sessionListener,
+                                             final long connectionTimeoutMillis) {
         super(sessionPreferences, promise, channel, timer, sessionListener, connectionTimeoutMillis);
     }
 
     @Override
-    protected void handleMessage(NetconfHelloMessage netconfMessage) throws NetconfDocumentedException {
-        NetconfClientSession session = super.getSessionForHelloMessage(netconfMessage);
-
-        if (shouldUseExi(netconfMessage.getDocument())){
-            logger.debug("Netconf session: {} should use exi.", session);
-            tryToStartExi(session);
+    protected void handleMessage(final NetconfHelloMessage netconfMessage) throws NetconfDocumentedException {
+        final NetconfClientSession session = getSessionForHelloMessage(netconfMessage);
+        replaceHelloMessageInboundHandler(session);
+
+        // If exi should be used, try to initiate exi communication
+        // Call negotiationSuccessFul after exi negotiation is finished successfully or not
+        if (shouldUseExi(netconfMessage)) {
+            logger.debug("Netconf session {} should use exi.", session);
+            NetconfStartExiMessage startExiMessage = (NetconfStartExiMessage) sessionPreferences.getStartExiMessage();
+            tryToInitiateExi(session, startExiMessage);
         } else {
-            logger.debug("Netconf session {} isn't capable using exi.", session);
+            // Exi is not supported, release session immediately
+            logger.debug("Netconf session {} isn't capable of using exi.", session);
             negotiationSuccessful(session);
         }
     }
 
-    private boolean shouldUseExi(Document doc) {
-        return containsExi10Capability(doc)
+    /**
+     * Initiates exi communication by sending start-exi message and waiting for positive/negative response.
+     *
+     * @param startExiMessage
+     */
+    void tryToInitiateExi(final NetconfClientSession session, final NetconfStartExiMessage startExiMessage) {
+        channel.pipeline().addAfter(AbstractChannelInitializer.NETCONF_MESSAGE_DECODER,
+                ExiConfirmationInboundHandler.EXI_CONFIRMED_HANDLER,
+                new ExiConfirmationInboundHandler(session, startExiMessage));
+
+        session.sendMessage(startExiMessage).addListener(new ChannelFutureListener() {
+            @Override
+            public void operationComplete(final ChannelFuture f) {
+                if (!f.isSuccess()) {
+                    logger.warn("Failed to send start-exi message {} on session {}", startExiMessage, this, f.cause());
+                    channel.pipeline().remove(ExiConfirmationInboundHandler.EXI_CONFIRMED_HANDLER);
+                } else {
+                    logger.trace("Start-exi message {} sent to socket on session {}", startExiMessage, this);
+                }
+            }
+        });
+    }
+
+    private boolean shouldUseExi(final NetconfHelloMessage helloMsg) {
+        return containsExi10Capability(helloMsg.getDocument())
                 && containsExi10Capability(sessionPreferences.getHelloMessage().getDocument());
     }
 
@@ -81,25 +116,9 @@ public class NetconfClientSessionNegotiator extends
         return false;
     }
 
-    private void tryToStartExi(final NetconfClientSession session) {
-        final NetconfMessage startExi = sessionPreferences.getStartExiMessage();
-        session.sendMessage(startExi).addListener(new ChannelFutureListener() {
-            @Override
-            public void operationComplete(final ChannelFuture f) {
-                if (!f.isSuccess()) {
-                    logger.warn("Failed to send start-exi message {} on session {}", startExi, session, f.cause());
-                } else {
-                    logger.trace("Start-exi message {} sent to socket on session {}", startExi, session);
-                    NetconfClientSessionNegotiator.this.channel.pipeline().addAfter(
-                            AbstractChannelInitializer.NETCONF_MESSAGE_DECODER, ExiConfirmationInboundHandler.EXI_CONFIRMED_HANDLER,
-                            new ExiConfirmationInboundHandler(session));
-                }
-            }
-        });
-    }
-
-    private long extractSessionId(Document doc) {
+    private long extractSessionId(final Document doc) {
         final Node sessionIdNode = (Node) XmlUtil.evaluateXPath(sessionIdXPath, doc, XPathConstants.NODE);
+        Preconditions.checkState(sessionIdNode != null, "");
         String textContent = sessionIdNode.getTextContent();
         if (textContent == null || textContent.equals("")) {
             throw new IllegalStateException("Session id not received from server");
@@ -109,9 +128,15 @@ public class NetconfClientSessionNegotiator extends
     }
 
     @Override
-    protected NetconfClientSession getSession(NetconfClientSessionListener sessionListener, Channel channel, NetconfHelloMessage message) throws NetconfDocumentedException {
-        return new NetconfClientSession(sessionListener, channel, extractSessionId(message.getDocument()),
-                NetconfMessageUtil.extractCapabilitiesFromHello(message.getDocument()));
+    protected NetconfClientSession getSession(final NetconfClientSessionListener sessionListener, final Channel channel,
+            final NetconfHelloMessage message) throws NetconfDocumentedException {
+        long sessionId = extractSessionId(message.getDocument());
+
+        // Copy here is important: it disconnects the strings from the document
+        Collection<String> capabilities = ImmutableList.copyOf(NetconfMessageUtil.extractCapabilitiesFromHello(message.getDocument()));
+
+        // FIXME: scalability: we could instantiate a cache to share the same collections
+        return new NetconfClientSession(sessionListener, channel, sessionId, capabilities);
     }
 
     /**
@@ -121,13 +146,15 @@ public class NetconfClientSessionNegotiator extends
         private static final String EXI_CONFIRMED_HANDLER = "exiConfirmedHandler";
 
         private final NetconfClientSession session;
+        private final NetconfStartExiMessage startExiMessage;
 
-        ExiConfirmationInboundHandler(NetconfClientSession session) {
+        ExiConfirmationInboundHandler(final NetconfClientSession session, final NetconfStartExiMessage startExiMessage) {
             this.session = session;
+            this.startExiMessage = startExiMessage;
         }
 
         @Override
-        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        public void channelRead(final ChannelHandlerContext ctx, final Object msg) throws Exception {
             ctx.pipeline().remove(ExiConfirmationInboundHandler.EXI_CONFIRMED_HANDLER);
 
             NetconfMessage netconfMessage = (NetconfMessage) msg;
@@ -136,19 +163,19 @@ public class NetconfClientSessionNegotiator extends
             if (NetconfMessageUtil.isOKMessage(netconfMessage)) {
                 logger.trace("Positive response on start-exi call received on session {}", session);
                 try {
-                    session.startExiCommunication(sessionPreferences.getStartExiMessage());
+                    session.startExiCommunication(startExiMessage);
                 } catch (RuntimeException e) {
                     // Unable to add exi, continue without exi
                     logger.warn("Unable to start exi communication, Communication will continue without exi on session {}", session, e);
                 }
 
-            // Error response
+                // Error response
             } else if(NetconfMessageUtil.isErrorMessage(netconfMessage)) {
                 logger.warn(
                         "Error response to start-exi message {}, Communication will continue without exi on session {}",
                         XmlUtil.toString(netconfMessage.getDocument()), session);
 
-            // Unexpected response to start-exi, throwing message away, continue without exi
+                // Unexpected response to start-exi, throwing message away, continue without exi
             } else {
                 logger.warn(
                         "Unexpected response to start-exi message, should be ok, was {}, " +
@@ -159,4 +186,5 @@ public class NetconfClientSessionNegotiator extends
             negotiationSuccessful(session);
         }
     }
+
 }