NetconfSession.java

/*
 Copyright (c) 2013 Juniper Networks, Inc.
 All Rights Reserved

 Use is subject to license terms.

*/

package net.juniper.netconf;

import com.google.common.annotations.VisibleForTesting;
import com.jcraft.jsch.Channel;
import com.jcraft.jsch.JSchException;
import net.juniper.netconf.element.Datastore;
import net.juniper.netconf.element.Hello;
import net.juniper.netconf.element.RpcReply;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.xml.sax.InputSource;
import org.xml.sax.SAXException;

import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.ParserConfigurationException;
import javax.xml.xpath.XPathExpressionException;
import java.io.ByteArrayOutputStream;
import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.StringReader;
import java.net.SocketTimeoutException;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * A {@code NetconfSession} is obtained by first building a
 * {@link Device} and then calling {@link Device#connect()}.
 * <p>
 * Typically, one
 * <ol>
 *   <li>Build a {@link Device} using its fluent {@code builder()}.</li>
 *   <li>Invoke {@link Device#connect()} to establish transport and receive a
 *       {@code NetconfSession}.</li>
 *   <li>Perform RPC operations via the session
 *       (e.g.&nbsp;{@link #executeRPC(String)}, {@link #killSession(String)},
 *       {@link #commitConfirm(long, String)}, {@link #cancelCommit(String)}).</li>
 *   <li>Call {@link #close()} when finished to free resources.</li>
 * </ol>
 * <p>
 * A {@code NetconfSession} should be treated as a single-threaded conversation
 * with the server. The library supports sequential reuse of one session across
 * multiple RPCs, but does not guarantee safety for multiple concurrent
 * in-flight RPCs on the same session. Use separate NETCONF sessions when
 * application-level concurrency is required.
 */
public class NetconfSession {

    private static final org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(NetconfSession.class);

    private final Channel netconfChannel;
    private String serverCapability;
    private Hello serverHello;

    private final InputStream stdInStreamFromDevice;
    private final OutputStream stdOutStreamToDevice;

    private String lastRpcReply;
    private RpcReply lastRpcReplyObject;
    private final DocumentBuilder builder;
    private final int commandTimeout;

    private final Map<String, String> rpcAttrMap = new HashMap<>();
    private String rpcAttributes;

    private int messageId = 0;
    /**
     * Size (in characters) of the temporary read buffer used when collecting
     * RPC replies from the device.  Set larger than the internal buffer in
     * {@link java.io.BufferedReader} so that large replies are less likely to
     * require multiple passes.
     */
    public static final int BUFFER_SIZE = 9 * 1024;

    private static final String CANDIDATE_CONFIG = "candidate";
    private static final String EMPTY_CONFIGURATION_TAG = "<configuration></configuration>";
    private static final String RUNNING_CONFIG = "running";
    private static final String NETCONF_SYNTAX_ERROR_MSG_FROM_DEVICE = "netconf error: syntax error";
    private static final byte[] DEVICE_PROMPT_BYTES = NetconfConstants.DEVICE_PROMPT.getBytes(StandardCharsets.UTF_8);
    private static final Pattern RPC_START_TAG_PATTERN = Pattern.compile("^<rpc\\b([^>]*)>", Pattern.DOTALL);
    private static final Pattern MESSAGE_ID_ATTRIBUTE_PATTERN =
        Pattern.compile("\\bmessage-id\\s*=\\s*(['\"])(.*?)\\1", Pattern.DOTALL);
    private static final Pattern XML_DECLARATION_PATTERN =
        Pattern.compile("^<\\?xml[^>]*\\?>\\s*", Pattern.DOTALL);

    private final List<String> clientCapabilities;
    private boolean useChunkedFraming;
    private NegotiatedCapabilities negotiatedCapabilities;
    private byte[] unreadBytes = new byte[0];

    private record PreparedRpc(String xml, String messageId) { }

    NetconfSession(Channel netconfChannel, int timeout, String hello,
                   DocumentBuilder builder) throws IOException {
        this(netconfChannel, timeout, timeout, defaultClientCapabilities(), hello, builder);
    }

    NetconfSession(Channel netconfChannel, int connectionTimeout, int commandTimeout,
                   String hello,
                   DocumentBuilder builder) throws IOException {
        this(netconfChannel, connectionTimeout, commandTimeout, defaultClientCapabilities(), hello, builder);
    }

    NetconfSession(Channel netconfChannel, int timeout, List<String> clientCapabilities,
                   String hello, DocumentBuilder builder) throws IOException {
        this(netconfChannel, timeout, timeout, clientCapabilities, hello, builder);
    }

    NetconfSession(Channel netconfChannel, int connectionTimeout, int commandTimeout,
                   List<String> clientCapabilities,
                   String hello,
                   DocumentBuilder builder) throws IOException {

        stdInStreamFromDevice = netconfChannel.getInputStream();
        stdOutStreamToDevice = netconfChannel.getOutputStream();
        try {
            netconfChannel.connect(connectionTimeout);
        } catch (JSchException e) {
            throw new NetconfException("Failed to create Netconf session:" +
                    e.getMessage(), e);
        }
        this.netconfChannel = netconfChannel;
        this.commandTimeout = commandTimeout;
        this.builder = builder;
        this.clientCapabilities = clientCapabilities == null ? List.of() : List.copyOf(clientCapabilities);

        sendHello(hello);
    }

    private static List<String> defaultClientCapabilities() {
        return List.of(
            NetconfConstants.URN_IETF_PARAMS_NETCONF_BASE_1_0,
            NetconfConstants.URN_IETF_PARAMS_NETCONF_BASE_1_1
        );
    }

    private XML convertToXML(String xml) throws SAXException, IOException {
        if (xml.contains(NETCONF_SYNTAX_ERROR_MSG_FROM_DEVICE)) {
            throw new NetconfException(String.format("Netconf server detected an error: %s", xml));
        }
        Document doc = builder.parse(new InputSource(new StringReader(xml)));
        Element root = doc.getDocumentElement();
        return new XML(root);
    }

    private void sendHello(String hello) throws IOException {
        setHelloReply(getRpcReply(hello));
    }

    @VisibleForTesting
    String getRpcReply(String rpc) throws IOException {
        PreparedRpc preparedRpc = sendRpcRequest(rpc);
        String reply;
        if (useChunkedFraming) {
            reply = readChunkedRpcReply();
        } else {
            reply = readLegacyRpcReply();
        }
        validateReplyMessageId(preparedRpc.messageId(), reply);
        return reply;
    }

    private String readLegacyRpcReply() throws IOException {
        final byte[] buffer = new byte[BUFFER_SIZE];
        final ByteArrayOutputStream rpcReply = new ByteArrayOutputStream();
        final long startTime = System.nanoTime();
        boolean timeoutNotExceeded = true;
        int promptPosition = -1;
        while (promptPosition < 0 &&
                (timeoutNotExceeded = !commandTimedOut(startTime))) {
            int bytesRead = readIncomingBytes(buffer);
            if (bytesRead < 0) {
                throw new NetconfException("Input Stream has been closed during reading.");
            }
            rpcReply.write(buffer, 0, bytesRead);
            byte[] rpcReplyBytes = rpcReply.toByteArray();
            promptPosition = indexOf(rpcReplyBytes, DEVICE_PROMPT_BYTES);
            if (promptPosition >= 0) {
                int unreadStart = promptPosition + DEVICE_PROMPT_BYTES.length;
                if (unreadStart < rpcReplyBytes.length) {
                    pushBackBytes(Arrays.copyOfRange(rpcReplyBytes, unreadStart, rpcReplyBytes.length));
                }
                String reply = new String(rpcReplyBytes, 0, promptPosition, StandardCharsets.UTF_8);
                log.debug("Received Netconf RPC-Reply\n{}", reply);
                return reply;
            }
        }

        throw new SocketTimeoutException("Command timeout limit was exceeded: " + commandTimeout);
    }

    private String readChunkedRpcReply() throws IOException {
        final long startTime = System.nanoTime();
        final ByteArrayOutputStream rpcReply = new ByteArrayOutputStream();
        while (!commandTimedOut(startTime)) {
            String header = readChunkHeaderLine(startTime);
            if (header.isEmpty()) {
                continue;
            }
            if ("##".equals(header)) {
                String reply = rpcReply.toString(StandardCharsets.UTF_8);
                log.debug("Received Netconf RPC-Reply\n{}", reply);
                return reply;
            }
            if (!header.startsWith("#")) {
                throw new NetconfException("Invalid NETCONF chunked framing header: " + header);
            }
            int chunkSize;
            try {
                chunkSize = Integer.parseInt(header.substring(1));
            } catch (NumberFormatException e) {
                throw new NetconfException("Invalid NETCONF chunk size: " + header, e);
            }
            if (chunkSize < 0) {
                throw new NetconfException("Invalid NETCONF chunk size: " + header);
            }
            rpcReply.write(readExactBytes(chunkSize, startTime));
        }
        throw new SocketTimeoutException("Command timeout limit was exceeded: " + commandTimeout);
    }

    private BufferedReader getRpcReplyRunning(String rpc) throws IOException {
        sendRpcRequest(rpc);
        return new BufferedReader(
                new InputStreamReader(createRpcReplyInputStream(), StandardCharsets.UTF_8));
    }

    private InputStream createRpcReplyInputStream() {
        return useChunkedFraming ? new ChunkedRpcReplyInputStream() : new LegacyRpcReplyInputStream();
    }

    private PreparedRpc sendRpcRequest(String rpc) throws IOException {
        PreparedRpc preparedRpc = prepareRpcRequest(rpc);
        log.debug("Sending Netconf RPC\n{}", preparedRpc.xml());
        stdOutStreamToDevice.write(applyTransportFraming(preparedRpc.xml()));
        stdOutStreamToDevice.flush();
        return preparedRpc;
    }

    private PreparedRpc prepareRpcRequest(String rpc) {
        String normalizedRpc = stripLegacyEndOfMessage(rpc).trim();
        String expectedMessageId = null;
        if (isHelloMessage(normalizedRpc)) {
            return new PreparedRpc(addXmlDeclarationIfMissing(normalizedRpc), null);
        }
        if (isRpcEnvelope(normalizedRpc)) {
            messageId++;
            expectedMessageId = extractRpcMessageId(normalizedRpc);
            if (expectedMessageId == null) {
                expectedMessageId = String.valueOf(messageId);
                normalizedRpc = injectMessageId(normalizedRpc, expectedMessageId);
            }
        }
        normalizedRpc = normalizedRpc.replace("<datastore>", "<datastore xmlns:ds=\"urn:ietf:params:xml:ns:yang:ietf-datastores\">");
        normalizedRpc = normalizedRpc.replace("<get-data>", "<get-data xmlns=\"urn:ietf:params:xml:ns:yang:ietf-netconf-nmda\">");
        return new PreparedRpc(addXmlDeclarationIfMissing(normalizedRpc), expectedMessageId);
    }

    /**
     * Gets the current rpc attribute string. If the rpc attribute string is not yet generated or has been reset then
     * we generate rpc attributes from the RPC Attribute Map.
     *
     * @return The attribute set XML formatted into a string.
     */
    public String getRpcAttributes() {
        if(rpcAttributes == null) {
            StringBuilder attributes = new StringBuilder();
            boolean useDefaultNamespace = true;
            for (Map.Entry<String, String> attribute : rpcAttrMap.entrySet()) {
                attributes.append(String.format(" %1s=\"%2s\"", attribute.getKey(), attribute.getValue()));
                if ("xmlns".equals(attribute.getKey()))
                    useDefaultNamespace = false;
            }
            if (useDefaultNamespace)
                attributes.append(" xmlns=\"" + NetconfConstants.URN_XML_NS_NETCONF_BASE_1_0 + "\"");
            rpcAttributes = attributes.toString();
        }
        return rpcAttributes;
    }


    /**
     * Loads the candidate configuration, Configuration should be in XML format.
     *
     * @param configuration Configuration,in XML format, to be loaded. For example,
     *                      "&lt;configuration&gt;&lt;system&gt;&lt;services&gt;&lt;ftp/&gt;&lt;
     *                      services/&gt;&lt;/system&gt;&lt;/configuration/&gt;"
     *                      will load 'ftp' under the 'systems services' hierarchy.
     * @param loadType      You can choose "merge" or "replace" as the loadType.
     * @throws org.xml.sax.SAXException If there are issues parsing the config file.
     * @throws java.io.IOException      If there are issues reading the config file.
     */
    public void loadXMLConfiguration(String configuration, String loadType) throws IOException, SAXException {
        requireCandidateCapability("loadXMLConfiguration");
        validateLoadType(loadType);
        configuration = configuration.trim();
        if (!configuration.startsWith("<configuration")) {
            configuration = "<configuration>" + configuration
                    + "</configuration>";
        }
        String rpc = "<rpc>" +
                "<edit-config>" +
                "<target>" +
                "<" + CANDIDATE_CONFIG + "/>" +
                "</target>" +
                "<default-operation>" +
                loadType +
                "</default-operation>" +
                "<config>" +
                configuration +
                "</config>" +
                "</edit-config>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        try {
            setLastRpcReply(getRpcReply(rpc));
        } catch (NetconfException e) {
            // Propagate as a LoadException so the caller knows this happened
            throw new LoadException("Load operation returned error.", e);
        }

        if (hasError() || !isOK()) {
            throw new LoadException(
                RpcErrorException.buildMessage("Load operation", lastRpcReplyObject),
                lastRpcReplyObject
            );
        }
    }

    private void setHelloReply(final String reply) throws IOException {
        this.serverCapability = reply;
        this.lastRpcReply = reply;
        try {
            this.serverHello = Hello.from(reply);
            this.negotiatedCapabilities =
                NegotiatedCapabilities.fromCapabilities(clientCapabilities, serverHello.getCapabilities());
            this.useChunkedFraming = negotiatedCapabilities.usesChunkedFraming();
        } catch (final ParserConfigurationException | SAXException | XPathExpressionException e) {
            throw new NetconfException("Invalid <hello> message from server: " + reply, e);
        }
    }

    private void setLastRpcReply(final String reply) throws IOException {
        this.lastRpcReply = reply;
        try {
            this.lastRpcReplyObject = RpcReply.from(reply);
        } catch (final ParserConfigurationException | SAXException | XPathExpressionException e) {
            throw new NetconfException("Invalid <rpc-reply> message from server: " + lastRpcReply, e);
        }
    }

    /**
     * Loads the candidate configuration, Configuration should be in text/tree
     * format.
     *
     * @param configuration Configuration,in text/tree format, to be loaded. For example,
     *                      "system {
     *                          services {
     *                              ftp;
     *                          }
     *                       }"
     *                      will load 'ftp' under the 'systems services' hierarchy.
     * @param loadType      You can choose "merge" or "replace" as the loadType.
     * @throws org.xml.sax.SAXException If there are issues parsing the config file.
     * @throws java.io.IOException      If there are issues reading the config file.
     */
    public void loadTextConfiguration(String configuration, String loadType) throws IOException, SAXException {
        requireCandidateCapability("loadTextConfiguration");
        String rpc = "<rpc>" +
                "<edit-config>" +
                "<target>" +
                "<" + CANDIDATE_CONFIG + "/>" +
                "</target>" +
                "<default-operation>" +
                loadType +
                "</default-operation>" +
                "<config-text>" +
                "<configuration-text>" +
                configuration +
                "</configuration-text>" +
                "</config-text>" +
                "</edit-config>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        try {
            setLastRpcReply(getRpcReply(rpc));
        } catch (NetconfException e) {
            throw new LoadException("Load operation returned error.", e);
        }

        if (hasError() || !isOK()) {
            throw new LoadException(
                RpcErrorException.buildMessage("Load operation", lastRpcReplyObject),
                lastRpcReplyObject
            );
        }
    }

    private String getConfig(String configTree) throws IOException {
        requireCandidateCapability("getCandidateConfig");

        String rpc = "<rpc>" +
                "<get-config>" +
                "<source>" +
                "<" + CANDIDATE_CONFIG + "/>" +
                "</source>" +
                "<filter type=\"subtree\">" +
                configTree +
                "</filter>" +
                "</get-config>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        setLastRpcReply(getRpcReply(rpc));
        return lastRpcReply;
    }

    /**
     * Executes a NETCONF {@code &lt;get&gt;} operation against the device’s running
     * datastore and returns the configuration **and** operational state data.
     *
     * @param xpathFilter optional XPath filter to limit the returned subtree;
     *                    pass {@code null} to retrieve the entire running state
     * @return an {@link XML} object representing the server’s {@code &lt;rpc-reply&gt;}
     * @throws IOException  if communication with the device fails
     * @throws SAXException if the reply cannot be parsed into valid XML
     */
    public XML getRunningConfigAndState(String xpathFilter) throws IOException, SAXException {
        String rpc = "<rpc>" +
                "<get>" +
                (xpathFilter == null ? "" : xpathFilter) +
                "</get>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        setLastRpcReply(getRpcReply(rpc));
        return convertToXML(lastRpcReply);
    }

    /**
     * Executes the YANG NMDA {@code &lt;get-data&gt;} operation against the specified
     * {@link Datastore}.
     *
     * @param xpathFilter optional XPath filter that narrows the returned data;
     *                    may be {@code null} for an unfiltered request
     * @param datastore   the target datastore (e.g., {@code operational}, {@code running});
     *                    must not be {@code null}
     * @return an {@link XML} object containing the server’s reply
     * @throws IllegalArgumentException if {@code datastore} is {@code null}
     * @throws IOException              if communication with the device fails
     * @throws SAXException             if the reply XML is malformed
     */
    public XML getData(String xpathFilter, Datastore datastore)
            throws IOException, SAXException {
        if (datastore == null) {
            throw new IllegalArgumentException("Datastore argument must not be null");
        }
        String rpc = "<rpc>" +
                "<get-data>" +
                "<datastore>ds:" + datastore + "</datastore>" +
                (xpathFilter == null ? "" : xpathFilter) +
                "</get-data>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        lastRpcReply = getRpcReply(rpc);
        return convertToXML(lastRpcReply);
    }

    private String getConfig(String target, String configTree)
            throws IOException {

        String rpc = "<rpc>" +
                "<get-config>" +
                "<source>" +
                "<" + target + "/>" +
                "</source>" +
                (configTree == null ? "" : "<filter type=\"subtree\">" + configTree + "</filter>") +
                "</get-config>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        setLastRpcReply(getRpcReply(rpc));
        return lastRpcReply;
    }

    /**
     * Get capability of the Netconf server.
     *
     * @return server capability
     */
    public String getServerCapability() {
        return serverCapability;
    }

    /**
     * Returns the &lt;hello&gt; message received from the server.
     * See <a href="https://datatracker.ietf.org/doc/html/rfc6241#section-8.1">...</a>
     * @return the &lt;hello&gt; message received from the server.
     */
    public Hello getServerHello() {
        return serverHello;
    }

    /**
     * Returns the negotiated capability view for this session.
     *
     * @return immutable capability snapshot
     */
    public NegotiatedCapabilities getNegotiatedCapabilities() {
        return negotiatedCapabilities;
    }

    /**
     * Sends a raw RPC string, waits for the {@code &lt;rpc-reply&gt;}, and converts the
     * response to an {@link XML} object.
     * <p>
     * If the server includes any {@code &lt;rpc-error&gt;} elements, the call
     * throws a {@link RpcErrorException}. This makes error handling
     * symmetrical with other high‑level helpers (e.g.&nbsp;{@code load*()},
     * {@code commit()}).
     *
     * @param rpcContent the RPC payload (with or without &lt;rpc&gt; wrapper)
     * @return parsed {@link XML} representation of the reply
     * @throws RpcErrorException       if the reply contains one or more
     *                                 {@code &lt;rpc-error&gt;} elements
     * @throws SAXException            if the reply cannot be parsed as XML
     * @throws IOException             on transport errors
     */
    public XML executeRPC(String rpcContent) throws SAXException, IOException {
        String rpcReply = getRpcReply(fixupRpc(rpcContent));
        setLastRpcReply(rpcReply);

        if (hasError()) {
            throw new RpcErrorException(
                RpcErrorException.buildMessage("RPC execution", lastRpcReplyObject),
                lastRpcReplyObject
            );
        }
        return convertToXML(rpcReply);
    }

    /**
     * Send an RPC (as XML object) over the Netconf session and get the response
     * as an XML object.
     *
     * @param rpc RPC to be sent. Use the XMLBuilder to create RPC as an
     *            XML object.
     * @return RPC reply sent by Netconf server
     * @throws org.xml.sax.SAXException If the XML Reply cannot be parsed.
     * @throws java.io.IOException      If there are issues communicating with the netconf server.
     */
    public XML executeRPC(XML rpc) throws SAXException, IOException {
        return executeRPC(rpc.toString());
    }

    /**
     * Send an RPC (as Document object) over the Netconf session and get the
     * response as an XML object.
     *
     * @param rpcDoc RPC content to be sent, as a org.w3c.dom.Document object.
     * @return RPC reply sent by Netconf server
     * @throws org.xml.sax.SAXException If the XML Reply cannot be parsed.
     * @throws java.io.IOException      If there are issues communicating with the netconf server.
     */
    public XML executeRPC(Document rpcDoc) throws SAXException, IOException {
        Element root = rpcDoc.getDocumentElement();
        XML xml = new XML(root);
        return executeRPC(xml);
    }


    /**
     * Given an RPC command, wrap it in RPC tags.
     * <a href="https://tools.ietf.org/html/rfc6241#section-4.1">...</a>
     *
     * @param rpcContent an RPC command that may or may not be wrapped in  &lt; or &gt;
     * @return a string of the RPC command wrapped in &lt;rpc&gt;&lt; &gt;&lt;/rpc&gt;
     * @throws IllegalArgumentException if null is passed in as the rpcContent.
     */
    @VisibleForTesting
    static String fixupRpc(String rpcContent) throws IllegalArgumentException {
        if (rpcContent == null) {
            throw new IllegalArgumentException("Null RPC");
        }
        String trimmedRpcContent = rpcContent.trim();
        if (isRpcEnvelope(trimmedRpcContent)) {
            return rpcContent + NetconfConstants.DEVICE_PROMPT;
        }
        if (trimmedRpcContent.startsWith("<"))
                rpcContent = "<rpc>" + trimmedRpcContent + "</rpc>";
            else
                rpcContent = "<rpc>" + "<" + trimmedRpcContent + "/>" + "</rpc>";
        return rpcContent + NetconfConstants.DEVICE_PROMPT;
    }


    /**
     * Send an RPC (as String object) over the default Netconf session and get
     * the response as a BufferedReader.
     *
     * @param rpcContent RPC content to be sent. For example, to send an rpc
     *                   &lt;rpc&gt;&lt;get-chassis-inventory/&gt;&lt;/rpc&gt;, the
     *                   String to be passed can be
     *                      "&lt;get-chassis-inventory/&gt;" OR
     *                      "get-chassis-inventory" OR
     *                      "&lt;rpc&gt;&lt;get-chassis-inventory/&gt;&lt;/rpc&gt;"
     * @return RPC reply sent by Netconf server as a BufferedReader. This is
     * useful if we want continuous stream of output, rather than wait
     * for whole output till command execution completes.
     * @throws java.io.IOException If there are issues communicating with the netconf server.
     */
    public BufferedReader executeRPCRunning(String rpcContent) throws IOException {
        return getRpcReplyRunning(fixupRpc(rpcContent));
    }

    /**
     * Send an RPC (as XML object) over the Netconf session and get the response
     * as a BufferedReader.
     *
     * @param rpc RPC to be sent. Use the XMLBuilder to create RPC as an
     *            XML object.
     * @return RPC reply sent by Netconf server as a BufferedReader. This is
     * useful if we want continuous stream of output, rather than wait
     * for whole output till command execution completes.
     * @throws java.io.IOException If there are issues communicating with the netconf server.
     */
    public BufferedReader executeRPCRunning(XML rpc) throws IOException {
        return executeRPCRunning(rpc.toString());
    }

    /**
     * Send an RPC (as Document object) over the Netconf session and get the
     * response as a BufferedReader.
     *
     * @param rpcDoc RPC content to be sent, as a org.w3c.dom.Document object.
     * @return RPC reply sent by Netconf server as a BufferedReader. This is
     * useful if we want continuous stream of output, rather than wait
     * for whole output till command execution completes.
     * @throws java.io.IOException If there are issues communicating with the netconf server.
     */
    public BufferedReader executeRPCRunning(Document rpcDoc) throws IOException {
        Element root = rpcDoc.getDocumentElement();
        XML xml = new XML(root);
        return executeRPCRunning(xml);
    }

    /**
     * Get the session ID of the Netconf session.
     *
     * @return Session ID as a string.
     */
    public String getSessionId() {
        return serverHello.getSessionId();
    }

    /**
     * Close the Netconf session. You should always call this once you don't
     * need the session anymore.
     *
     * @throws IOException if there are errors communicating with the Device
     */
    public void close() throws IOException {
        String rpc = "<rpc>" +
                "<close-session/>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        setLastRpcReply(getRpcReply(rpc));
        netconfChannel.disconnect();
    }

    /**
     * Check if the last RPC reply returned from Netconf server has any error.
     *
     * @return true if any errors are found in last RPC reply.
     */
    public boolean hasError() {
        return lastRpcReplyObject.hasErrors();
    }

    /**
     * Check if the last RPC reply returned from Netconf server has any warning.
     *
     * @return true if any errors are found in last RPC reply.
     */
    public boolean hasWarning() {
        return lastRpcReplyObject.hasWarnings();
    }

    /**
     * Check if the last RPC reply returned from Netconf server,
     * contains &lt;ok/&gt; tag.
     *
     * @return true if &lt;ok/&gt; tag is found in last RPC reply.
     */
    public boolean isOK() {
        return lastRpcReplyObject.isOK();
    }

    /**
     * Locks the candidate configuration.
     *
     * @return true if successful.
     * @throws org.xml.sax.SAXException If the XML Reply cannot be parsed.
     * @throws java.io.IOException      If there are issues communicating with the netconf server.
     */
    public boolean lockConfig() throws IOException, SAXException {
        requireCandidateCapability("lockConfig");
        String rpc = "<rpc>" +
                "<lock>" +
                "<target>" +
                "<candidate/>" +
                "</target>" +
                "</lock>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        setLastRpcReply(getRpcReply(rpc));
        return !hasError() && isOK();
    }

    /**
     * Unlocks the candidate configuration.
     *
     * @return true if successful.
     * @throws java.io.IOException      If there are issues communicating with the netconf server.
     */
    public boolean unlockConfig() throws IOException {
        requireCandidateCapability("unlockConfig");
        String rpc = "<rpc>" +
                "<unlock>" +
                "<target>" +
                "<candidate/>" +
                "</target>" +
                "</unlock>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        setLastRpcReply(getRpcReply(rpc));
        return !hasError() && isOK();
    }

    /**
     * Terminates another active NETCONF session on the server
     * (<a href="https://datatracker.ietf.org/doc/html/rfc6241#section-7.9">RFC&nbsp;6241&nbsp;§7.9</a>).
     * <p>
     * The server will abort any operations in progress for that session,
     * release resources and locks, and close the connection.
     *
     * @param sessionId the session identifier to terminate; must not be {@code null} or empty
     * @return {@code true} if the operation succeeded (no &lt;rpc-error&gt; and an &lt;ok/&gt; was returned)
     * @throws IllegalArgumentException if {@code sessionId} is {@code null} or empty
     * @throws IOException              if communication with the device fails
     * @throws SAXException             if the reply cannot be parsed
     */
    public boolean killSession(String sessionId) throws IOException, SAXException {
        if (sessionId == null || sessionId.isEmpty()) {
            throw new IllegalArgumentException("sessionId must not be null or empty");
        }

        String rpc = "<rpc>" +
                "<kill-session>" +
                "<session-id>" + sessionId + "</session-id>" +
                "</kill-session>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;

        setLastRpcReply(getRpcReply(rpc));
        return !hasError() && isOK();
    }

    /**
     * Loads the candidate configuration, Configuration should be in set
     * format.
     * NOTE: This method is applicable only for JUNOS release 11.4 and above.
     *
     * @param configuration Configuration,in set format, to be loaded. For example,
     *                      "set system services ftp"
     *                      will load 'ftp' under the 'systems services' hierarchy.
     *                      To load multiple set statements, separate them by '\n' character.
     * @throws java.io.IOException      If there are issues reading the config file.
     */
    public void loadSetConfiguration(String configuration) throws IOException {
        requireCandidateCapability("loadSetConfiguration");
        String rpc = "<rpc>" +
                "<load-configuration action=\"set\">" +
                "<configuration-set>" +
                configuration +
                "</configuration-set>" +
                "</load-configuration>" +
                "</rpc>";
        setLastRpcReply(getRpcReply(rpc));
        if (hasError() || !isOK()) {
            throw new LoadException(
                RpcErrorException.buildMessage("Load operation", lastRpcReplyObject),
                lastRpcReplyObject
            );
        }
    }

    /**
     * Loads the candidate configuration from file,
     * configuration should be in XML format.
     *
     * @param configFile Path name of file containing configuration,in xml format,
     *                   to be loaded.
     * @param loadType   You can choose "merge" or "replace" as the loadType.
     * @throws org.xml.sax.SAXException If there are issues parsing the config file.
     * @throws java.io.IOException      If there are issues reading the config file.
     */
    public void loadXMLFile(String configFile, String loadType) throws IOException, SAXException {
        requireCandidateCapability("loadXMLFile");
        validateLoadType(loadType);
        loadXMLConfiguration(readConfigFile(configFile), loadType);
    }

    /**
     * Validate that the load type is either merge or replace.
     *
     * @param loadType how to load the config, merge or replace.
     * @throws IllegalArgumentException if the load type is not merge or replace.
     */
    private void validateLoadType(String loadType) throws IllegalArgumentException {
        if (loadType == null || (!loadType.equals("merge") &&
                !loadType.equals("replace")))
            throw new IllegalArgumentException("'loadType' argument must be " +
                    "merge|replace");
    }

    /**
     * Read the config file and return as a string.
     *
     * @param configFile The name of the configuration file
     * @return a string of the config file.
     * @throws java.io.IOException If there are issues reading the config file.
     */
    private String readConfigFile(String configFile) throws IOException {
        try {
            return Files.readString(Paths.get(configFile), Charset.defaultCharset());
        } catch (FileNotFoundException e) {
            throw new FileNotFoundException("The system cannot find the configuration file specified: " + configFile);
        }
    }


    /**
     * Loads the candidate configuration from file,
     * configuration should be in text/tree format.
     *
     * @param configFile Path name of file containing configuration,in xml format,
     *                   to be loaded.
     * @param loadType   You can choose "merge" or "replace" as the loadType.
     * @throws org.xml.sax.SAXException If there are issues parsing the config file.
     * @throws java.io.IOException      If there are issues reading the config file.
     */
    public void loadTextFile(String configFile, String loadType) throws IOException, SAXException {
        requireCandidateCapability("loadTextFile");
        validateLoadType(loadType);
        loadTextConfiguration(readConfigFile(configFile), loadType);
    }

    /**
     * Loads the candidate configuration from file,
     * configuration should be in set format.
     * NOTE: This method is applicable only for JUNOS release 11.4 and above.
     *
     * @param configFile Path name of file containing configuration,in set format,
     *                   to be loaded.
     * @throws java.io.IOException      If there are issues reading the config file.
     */
    public void loadSetFile(String configFile) throws
            IOException {
        requireCandidateCapability("loadSetFile");
        loadSetConfiguration(readConfigFile(configFile));
    }

    /**
     * Loads and commits the candidate configuration, Configuration can be in
     * text/xml/set format.
     *
     * @param configFile Path name of file containing configuration,in text/xml/set format,
     *                   to be loaded. For example,
     *                   "system {
     *                      services {
     *                          ftp;
     *                      }
     *                   }"
     *                   will load 'ftp' under the 'systems services' hierarchy.
     *                   OR
     *                   "&lt;configuration&gt;&lt;system&gt;&lt;services&gt;&lt;ftp/&gt;&lt;
     *                   services/&gt;&lt;/system&gt;&lt;/configuration/&gt;"
     *                   will load 'ftp' under the 'systems services' hierarchy.
     *                   OR
     *                   "set system services ftp"
     *                   will load 'ftp' under the 'systems services' hierarchy.
     * @param loadType   You can choose "merge" or "replace" as the loadType.
     *                   NOTE: This parameter's value is redundant in case the file contains
     *                   configuration in 'set' format.
     * @throws java.io.IOException      if there are errors communication with the netconf server.
     * @throws org.xml.sax.SAXException if there are errors parsing the XML reply.
     */
    public void commitThisConfiguration(String configFile, String loadType) throws IOException, SAXException {
        requireCandidateCapability("commitThisConfiguration");
        String configuration = readConfigFile(configFile);
        configuration = configuration.trim();
        if (!this.lockConfig()) {
            throw new IOException("Unclean lock operation. Cannot proceed " +
                    "further.");
        }
        Throwable pendingFailure = null;
        try {
            if (configuration.startsWith("<")) {
                this.loadXMLConfiguration(configuration, loadType);
            } else if (configuration.startsWith("set")) {
                this.loadSetConfiguration(configuration);
            } else {
                this.loadTextConfiguration(configuration, loadType);
            }
            this.commit();
        } catch (Throwable t) {
            pendingFailure = t;
            throw t;
        } finally {
            try {
                this.unlockConfig();
            } catch (IOException e) {
                if (pendingFailure != null) {
                    pendingFailure.addSuppressed(e);
                } else {
                    throw e;
                }
            }
        }
    }

    /**
     * Commit the candidate configuration.
     *
     * @throws java.io.IOException      If there are errors communicating with the netconf server.
     * @throws org.xml.sax.SAXException If there are errors parsing the XML reply.
     */
    public void commit() throws IOException, SAXException {
        requireCandidateCapability("commit");
        String rpc = "<rpc>" +
                "<commit/>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        setLastRpcReply(getRpcReply(rpc));
        if (hasError() || !isOK()) {
            throw new CommitException(
                RpcErrorException.buildMessage("Commit operation", lastRpcReplyObject),
                lastRpcReplyObject
            );
        }
    }

    /**
     * Sends a *confirmed* &lt;commit&gt; request that follows the original
     * RFC&nbsp;6241 §8.4 semantics (confirmed‑commit&nbsp;**1.0**).
     * <p>
     * The server will roll back the candidate configuration unless a
     * non‑confirmed &lt;commit&gt; is issued from **the same session**
     * within the given timeout period.
     * </p>
     *
     * @param seconds the &lt;confirm-timeout&gt; in seconds; if {@code seconds
     *                &lt;= 0} the device’s default (600 s) is used
     * @throws IOException if communication with the device fails
     *
     * @deprecated Prefer {@link #commitConfirm(long, String)} which adds
     * the <code>&lt;persist&gt;</code> / <code>&lt;persist-id&gt;</code>
     * parameters required by the
     * <em>:confirmed-commit:1.1</em> capability.
     */
    @Deprecated
    public void commitConfirm(long seconds) throws IOException {
        commitConfirm(seconds, null);
    }

    /**
     * Issues a <em>confirmed</em> commit as defined by the
     * <a href="https://datatracker.ietf.org/doc/html/rfc6241#section-8.4">
     * :confirmed-commit:1.1 capability</a>.
     *
     * @param seconds        confirm‑timeout (600 s default).  If {@code seconds &lt;= 0}
     *                       the timeout element is omitted and the server default
     *                       is used.
     * @param persistToken   optional token for cross‑session confirmation; if
     *                       {@code null} the commit can only be confirmed from
     *                       the same session.
     * @throws IOException   if communication with the device fails
     */
    public void commitConfirm(long seconds, String persistToken) throws IOException {
        requireCandidateCapability("commitConfirm");
        requireConfirmedCommitCapability("commitConfirm", persistToken != null);
        StringBuilder rpc = new StringBuilder();
        rpc.append("<rpc><commit><confirmed/>");
        if (seconds > 0) {
            rpc.append("<confirm-timeout>").append(seconds).append("</confirm-timeout>");
        }
        if (persistToken != null) {
            rpc.append("<persist>").append(persistToken).append("</persist>");
        }
        rpc.append("</commit></rpc>").append(NetconfConstants.DEVICE_PROMPT);

        setLastRpcReply(getRpcReply(rpc.toString()));
        if (hasError() || !isOK()) {
            throw new CommitException(
                RpcErrorException.buildMessage("Confirmed-commit operation", lastRpcReplyObject),
                lastRpcReplyObject
            );
        }
    }

    /**
     * Cancels a pending confirmed commit.
     *
     * @param persistId optional token that matches the &lt;persist&gt; used
     *                  in the original confirmed commit; may be {@code null}
     *                  if no token was supplied.
     * @return {@code true} if the server replied &lt;ok/&gt;
     * @throws IOException  if communication with the device fails
     * @throws SAXException if the reply cannot be parsed
     */
    public boolean cancelCommit(String persistId) throws IOException, SAXException {
        requireCandidateCapability("cancelCommit");
        requireConfirmedCommitCapability("cancelCommit", persistId != null);
        StringBuilder rpc = new StringBuilder();
        rpc.append("<rpc><cancel-commit/>");
        if (persistId != null) {
            rpc.insert(rpc.length() - "</cancel-commit/>".length(),
                       "<persist-id>" + persistId + "</persist-id>");
        }
        rpc.append("</rpc>").append(NetconfConstants.DEVICE_PROMPT);

        setLastRpcReply(getRpcReply(rpc.toString()));
        return !hasError() && isOK();
    }

    /**
     * Commit the candidate configuration and rebuild the config database.
     *
     * @throws net.juniper.netconf.CommitException if there is an error committing the config.
     * @throws java.io.IOException                 If there are errors communicating with the netconf server.
     */
    public void commitFull() throws CommitException, IOException {
        requireCandidateCapability("commitFull");
        String rpc = "<rpc>" +
                "<commit-configuration>" +
                "<full/>" +
                "</commit-configuration>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        setLastRpcReply(getRpcReply(rpc));
        if (hasError() || !isOK()) {
            throw new CommitException(
                RpcErrorException.buildMessage("Commit operation", lastRpcReplyObject),
                lastRpcReplyObject
            );
        }
    }


    /**
     * Retrieve the candidate configuration, or part of the configuration.
     *
     * @param configTree configuration hierarchy to be retrieved as the argument.
     *                   For example, to get the whole configuration, argument should be
     *                   &lt;configuration&gt;&lt;/configuration&gt;
     * @return configuration data as XML object.
     * @throws java.io.IOException      If there are errors communicating with the netconf server.
     * @throws org.xml.sax.SAXException If there are errors parsing the XML reply.
     */
    public XML getCandidateConfig(String configTree) throws SAXException,
            IOException {
        return convertToXML(getConfig(configTree));
    }

    /**
     * Retrieve the running configuration, or part of the configuration.
     *
     * @param configTree configuration hierarchy to be retrieved as the argument.
     *                   For example, to get the whole configuration, argument should be
     *                   &lt;configuration&gt;&lt;/configuration&gt;
     * @return configuration data as XML object.
     * @throws java.io.IOException      If there are errors communicating with the netconf server.
     * @throws org.xml.sax.SAXException If there are errors parsing the XML reply.
     */
    public XML getRunningConfig(String configTree) throws SAXException,
            IOException {
        return convertToXML(getConfig(RUNNING_CONFIG, configTree));
    }

    /**
     * Retrieve the whole candidate configuration.
     *
     * @return configuration data as XML object.
     * @throws java.io.IOException      If there are errors communicating with the netconf server.
     * @throws org.xml.sax.SAXException If there are errors parsing the XML reply.
     */
    public XML getCandidateConfig() throws SAXException, IOException {
        return convertToXML(getConfig(EMPTY_CONFIGURATION_TAG));
    }

    /**
     * Retrieve the whole running configuration.
     *
     * @return configuration data as XML object.
     * @throws java.io.IOException      If there are errors communicating with the netconf server.
     * @throws org.xml.sax.SAXException If there are errors parsing the XML reply.
     */
    public XML getRunningConfig() throws SAXException, IOException {
        return convertToXML(getConfig(RUNNING_CONFIG, EMPTY_CONFIGURATION_TAG));
    }

    /**
     * Validate the candidate configuration.
     *
     * @return {@code true} if validation completed with a clean {@code <ok/>}
     *         reply; {@code false} for warning-only or other non-error
     *         non-{@code <ok/>} replies
     * @throws java.io.IOException      If there are errors communicating with the netconf server.
     * @throws ValidateException        if the server returns one or more
     *                                  {@code <rpc-error>} elements
     */
    public boolean validate() throws IOException {
        requireCandidateCapability("validate");
        requireValidateCapability("validate");

        String rpc = "<rpc>" +
                "<validate>" +
                "<source>" +
                "<candidate/>" +
                "</source>" +
                "</validate>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        setLastRpcReply(getRpcReply(rpc));
        if (hasError()) {
            throw new ValidateException(
                RpcErrorException.buildMessage("Validate operation", lastRpcReplyObject),
                lastRpcReplyObject
            );
        }
        return isOK();
    }

    /**
     * Reboot the device corresponding to the Netconf Session.
     *
     * @return RPC reply sent by Netconf server.
     * @throws java.io.IOException If there are errors communicating with the netconf server.
     */
    public String reboot() throws IOException {
        String rpc = "<rpc>" +
                "<request-reboot/>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        return getRpcReply(rpc);
    }

    /**
     * Run a cli command.
     * NOTE: The text output is supported for JUNOS 11.4 and later.
     *
     * @param command the cli command to be executed.
     * @return result of the command, as a String.
     * @throws java.io.IOException      If there are errors communicating with the netconf server.
     * @throws org.xml.sax.SAXException If there are errors parsing the XML reply.
     */
    public String runCliCommand(String command) throws IOException, SAXException {

        String rpc = "<rpc>" +
                "<command format=\"text\">" +
                command +
                "</command>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        String rpcReply = getRpcReply(rpc);
        setLastRpcReply(rpcReply);
        XML xmlReply = convertToXML(rpcReply);
        List<String> tags = new ArrayList<>();
        tags.add("output");
        String output = xmlReply.findValue(tags);
        if (output != null)
            return output;
        else
            return rpcReply;
    }

    /**
     * Run a cli command.
     *
     * @param command the cli command to be executed.
     * @return result of the command, as a BufferedReader. This is
     * useful if we want continuous stream of output, rather than wait
     * for whole output till command execution completes.
     * @throws java.io.IOException If there are errors communicating with the netconf server.
     */
    public BufferedReader runCliCommandRunning(String command) throws
            IOException {

        String rpc = "<command format=\"text\">" +
                command +
                "</command>";
        return executeRPCRunning(rpc);
    }

    /**
     * This method should be called for load operations to happen in 'private'
     * mode.
     *
     * @param mode Mode in which to open the configuration.
     *             Permissible mode(s): "private"
     * @throws java.io.IOException If there are errors communicating with the netconf server.
     */
    public void openConfiguration(String mode) throws IOException {

        StringBuilder rpc = new StringBuilder();
        rpc.append("<rpc>");
        rpc.append("<open-configuration>");
        if (mode.startsWith("<"))
            rpc.append(mode);
        else
            rpc.append("<").append(mode).append("/>");
        rpc.append("</open-configuration>");
        rpc.append("</rpc>");
        rpc.append(NetconfConstants.DEVICE_PROMPT);
        setLastRpcReply(getRpcReply(rpc.toString()));
    }

    /**
     * This method should be called to close a private session, in case its
     * started.
     *
     * @throws java.io.IOException If there are errors communicating with the netconf server.
     */
    public void closeConfiguration() throws IOException {
        String rpc = "<rpc>" +
                "<close-configuration/>" +
                "</rpc>" +
                NetconfConstants.DEVICE_PROMPT;
        setLastRpcReply(getRpcReply(rpc));
    }

    /**
     * Returns the last RPC reply sent by Netconf server.
     *
     * @return Last RPC reply, as a string.
     */
    public String getLastRPCReply() {
        return this.lastRpcReply;
    }

    /**
     * Returns the last RPC reply sent by Netconf server.
     *
     * @return Last RPC reply, as a RpcReply object.
     */
    public RpcReply getLastRpcReplyObject() {
        return lastRpcReplyObject;
    }

    /**
     * Adds an Attribute to the set of RPC attributes used in the RPC XML envelope. Resets the rpcAttributes value
     * to null for generation on the next request.
     *
     * @param name The attribute name for the new attribute.
     * @param value The attribute value for the new attribute.
     */
    public void addRPCAttribute(String name, String value) {
        rpcAttrMap.put(name, value);
        rpcAttributes = null;
    }

    /**
     * Removes an attribute from the set of RPC attributes used in the RPC XML envelope. Resets the rpcAttributes value
     * to null for generation on the next request.
     *
     * @param name The attribute name to be removed.
     *
     * @return The value of the removed attribute.
     */
    public String removeRPCAttribute (String name) {
        rpcAttributes = null;
        return rpcAttrMap.remove(name);
    }

    /**
     * Clears all the RPC attributes from the set of RPC attributes used in the RPC XML envelope. The set will be empty
     * after this call returns. Resets the rpcAttributes value to null for generation on the next request.
     */
    public void removeAllRPCAttributes() {
        rpcAttrMap.clear();
        rpcAttributes = null;
    }

    private static boolean isRpcEnvelope(String rpcContent) {
        String normalizedRpcContent = stripXmlDeclaration(rpcContent.trim());
        if (!normalizedRpcContent.startsWith("<rpc")) {
            return false;
        }
        if (normalizedRpcContent.length() == 4) {
            return false;
        }
        char next = normalizedRpcContent.charAt(4);
        return next == '>' || next == '/' || Character.isWhitespace(next);
    }

    private static String stripLegacyEndOfMessage(String rpc) {
        String trimmed = rpc.trim();
        if (trimmed.endsWith(NetconfConstants.DEVICE_PROMPT)) {
            return trimmed.substring(0, trimmed.length() - NetconfConstants.DEVICE_PROMPT.length()).trim();
        }
        return trimmed;
    }

    private static String stripXmlDeclaration(String xml) {
        return XML_DECLARATION_PATTERN.matcher(xml).replaceFirst("");
    }

    private static String addXmlDeclarationIfMissing(String xml) {
        return xml.startsWith("<?xml") ? xml : NetconfConstants.XML_VERSION + xml;
    }

    private static boolean isHelloMessage(String xml) {
        return stripXmlDeclaration(xml).startsWith("<hello");
    }

    private static boolean isRpcReply(String xml) {
        return stripXmlDeclaration(xml).startsWith("<rpc-reply");
    }

    private String extractRpcMessageId(String rpc) {
        Matcher startTagMatcher = RPC_START_TAG_PATTERN.matcher(stripXmlDeclaration(rpc));
        if (!startTagMatcher.find()) {
            return null;
        }
        Matcher messageIdMatcher = MESSAGE_ID_ATTRIBUTE_PATTERN.matcher(startTagMatcher.group(1));
        if (!messageIdMatcher.find()) {
            return null;
        }
        return messageIdMatcher.group(2);
    }

    private String injectMessageId(String rpc, String generatedMessageId) {
        Matcher startTagMatcher = RPC_START_TAG_PATTERN.matcher(stripXmlDeclaration(rpc));
        if (!startTagMatcher.find()) {
            return rpc;
        }
        String existingAttributes = startTagMatcher.group(1);
        String updatedRpc = "<rpc" + existingAttributes + getMissingRpcAttributes(existingAttributes)
            + " message-id=\"" + generatedMessageId + "\">"
            + stripXmlDeclaration(rpc).substring(startTagMatcher.end());
        return rpc.startsWith("<?xml") ? NetconfConstants.XML_VERSION + updatedRpc : updatedRpc;
    }

    private String getMissingRpcAttributes(String existingAttributes) {
        StringBuilder attributes = new StringBuilder();
        boolean hasDefaultNamespace = hasAttribute(existingAttributes, "xmlns");
        for (Map.Entry<String, String> attribute : rpcAttrMap.entrySet()) {
            if (!hasAttribute(existingAttributes, attribute.getKey())) {
                attributes.append(" ")
                    .append(attribute.getKey())
                    .append("=\"")
                    .append(attribute.getValue())
                    .append("\"");
            }
            if ("xmlns".equals(attribute.getKey())) {
                hasDefaultNamespace = true;
            }
        }
        if (!hasDefaultNamespace) {
            attributes.append(" xmlns=\"").append(NetconfConstants.URN_XML_NS_NETCONF_BASE_1_0).append("\"");
        }
        return attributes.toString();
    }

    private boolean hasAttribute(String existingAttributes, String attributeName) {
        return Pattern.compile("(^|\\s)" + Pattern.quote(attributeName) + "\\s*=")
            .matcher(existingAttributes)
            .find();
    }

    private void validateReplyMessageId(String expectedMessageId, String reply) throws IOException {
        if (expectedMessageId == null || !isRpcReply(reply.trim())) {
            return;
        }
        try {
            RpcReply rpcReply = RpcReply.from(reply);
            String actualMessageId = rpcReply.getMessageId();
            if (actualMessageId != null && !expectedMessageId.equals(actualMessageId)) {
                throw new NetconfException(
                    "Mismatched rpc-reply message-id. expected=" + expectedMessageId + ", actual=" + actualMessageId);
            }
        } catch (ParserConfigurationException | SAXException | XPathExpressionException e) {
            throw new NetconfException("Invalid <rpc-reply> message from server: " + reply, e);
        }
    }

    private byte[] applyTransportFraming(String rpc) {
        if (!useChunkedFraming) {
            return (rpc + NetconfConstants.DEVICE_PROMPT).getBytes(StandardCharsets.UTF_8);
        }
        int payloadLength = rpc.getBytes(StandardCharsets.UTF_8).length;
        String chunkedRpc = "\n#" + payloadLength + "\n" + rpc + "\n##\n";
        return chunkedRpc.getBytes(StandardCharsets.UTF_8);
    }

    private int readIncomingBytes(byte[] buffer) throws IOException {
        if (unreadBytes.length > 0) {
            int bytesToCopy = Math.min(buffer.length, unreadBytes.length);
            System.arraycopy(unreadBytes, 0, buffer, 0, bytesToCopy);
            unreadBytes = Arrays.copyOfRange(unreadBytes, bytesToCopy, unreadBytes.length);
            return bytesToCopy;
        }
        return stdInStreamFromDevice.read(buffer, 0, buffer.length);
    }

    private int readIncomingByte() throws IOException {
        if (unreadBytes.length > 0) {
            int nextByte = unreadBytes[0] & 0xFF;
            unreadBytes = Arrays.copyOfRange(unreadBytes, 1, unreadBytes.length);
            return nextByte;
        }
        return stdInStreamFromDevice.read();
    }

    private void pushBackBytes(byte[] bytes) {
        if (bytes.length == 0) {
            return;
        }
        byte[] combined = new byte[bytes.length + unreadBytes.length];
        System.arraycopy(bytes, 0, combined, 0, bytes.length);
        System.arraycopy(unreadBytes, 0, combined, bytes.length, unreadBytes.length);
        unreadBytes = combined;
    }

    private String readChunkHeaderLine(long startTime) throws IOException {
        ByteArrayOutputStream lineBuffer = new ByteArrayOutputStream();
        while (!commandTimedOut(startTime)) {
            int nextByte = readIncomingByte();
            if (nextByte < 0) {
                throw new NetconfException("Input Stream has been closed during reading.");
            }
            if (nextByte == '\n') {
                return lineBuffer.toString(StandardCharsets.UTF_8);
            }
            if (nextByte != '\r') {
                lineBuffer.write(nextByte);
            }
        }
        throw new SocketTimeoutException("Command timeout limit was exceeded: " + commandTimeout);
    }

    private byte[] readExactBytes(int expectedLength, long startTime) throws IOException {
        byte[] chunkBytes = new byte[expectedLength];
        int offset = 0;
        while (offset < expectedLength && !commandTimedOut(startTime)) {
            int bytesRead;
            if (unreadBytes.length > 0) {
                bytesRead = Math.min(expectedLength - offset, unreadBytes.length);
                System.arraycopy(unreadBytes, 0, chunkBytes, offset, bytesRead);
                unreadBytes = Arrays.copyOfRange(unreadBytes, bytesRead, unreadBytes.length);
            } else {
                bytesRead = stdInStreamFromDevice.read(chunkBytes, offset, expectedLength - offset);
            }
            if (bytesRead < 0) {
                throw new NetconfException("Input Stream has been closed during reading.");
            }
            offset += bytesRead;
        }
        if (offset < expectedLength) {
            throw new SocketTimeoutException("Command timeout limit was exceeded: " + commandTimeout);
        }
        return chunkBytes;
    }

    private boolean commandTimedOut(long startTime) {
        return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime) >= commandTimeout;
    }

    private NegotiatedCapabilities requireNegotiatedCapabilities() {
        if (negotiatedCapabilities == null) {
            throw new IllegalStateException("NETCONF session capabilities have not been negotiated");
        }
        return negotiatedCapabilities;
    }

    private void requireCandidateCapability(String operation) throws UnsupportedCapabilityException {
        requireCapability(requireNegotiatedCapabilities().supportsCandidate(), "candidate", operation);
    }

    private void requireValidateCapability(String operation) throws UnsupportedCapabilityException {
        requireCapability(requireNegotiatedCapabilities().supportsValidate(), "validate", operation);
    }

    private void requireConfirmedCommitCapability(String operation, boolean requiresPersistSupport)
        throws UnsupportedCapabilityException {
        NegotiatedCapabilities capabilities = requireNegotiatedCapabilities();
        requireCapability(
            requiresPersistSupport ? capabilities.supportsConfirmedCommit11() : capabilities.supportsConfirmedCommit(),
            requiresPersistSupport ? "confirmed-commit:1.1" : "confirmed-commit",
            operation
        );
    }

    private void requireCapability(boolean supported, String capability, String operation)
        throws UnsupportedCapabilityException {
        if (!supported) {
            throw new UnsupportedCapabilityException(operation, capability);
        }
    }

    private abstract class RpcReplyInputStream extends InputStream {

        private final long startTime = System.nanoTime();
        private boolean finished;
        private boolean closed;

        @Override
        public int read() throws IOException {
            if (closed) {
                throw new IOException("Stream closed");
            }
            if (finished) {
                return -1;
            }
            int nextByte = readPayloadByte();
            if (nextByte < 0) {
                finished = true;
            }
            return nextByte;
        }

        @Override
        public int read(byte[] buffer, int off, int len) throws IOException {
            if (buffer == null) {
                throw new NullPointerException("buffer");
            }
            if (off < 0 || len < 0 || len > buffer.length - off) {
                throw new IndexOutOfBoundsException();
            }
            if (len == 0) {
                return 0;
            }
            int firstByte = read();
            if (firstByte < 0) {
                return -1;
            }
            buffer[off] = (byte) firstByte;
            int bytesRead = 1;
            while (bytesRead < len) {
                int nextByte = read();
                if (nextByte < 0) {
                    break;
                }
                buffer[off + bytesRead] = (byte) nextByte;
                bytesRead++;
            }
            return bytesRead;
        }

        @Override
        public void close() throws IOException {
            if (closed) {
                return;
            }
            try {
                byte[] discardBuffer = new byte[BUFFER_SIZE];
                while (read(discardBuffer, 0, discardBuffer.length) != -1) {
                    // Drain the current reply so the session stays aligned for the next RPC.
                }
            } finally {
                finished = true;
                closed = true;
            }
        }

        protected abstract int readPayloadByte() throws IOException;

        protected final int readIncomingByteOrThrow() throws IOException {
            if (commandTimedOut(startTime)) {
                throw new SocketTimeoutException("Command timeout limit was exceeded: " + commandTimeout);
            }
            int nextByte = readIncomingByte();
            if (nextByte < 0) {
                throw new NetconfException("Input Stream has been closed during reading.");
            }
            return nextByte;
        }

        protected final String readChunkHeaderLine() throws IOException {
            return NetconfSession.this.readChunkHeaderLine(startTime);
        }
    }

    private final class LegacyRpcReplyInputStream extends RpcReplyInputStream {

        private final byte[] candidatePrompt = new byte[DEVICE_PROMPT_BYTES.length];
        private int candidateLength;

        @Override
        protected int readPayloadByte() throws IOException {
            while (true) {
                if (candidateLength > 0 && !matchesPromptPrefix()) {
                    return shiftCandidateByte();
                }
                if (candidateLength == DEVICE_PROMPT_BYTES.length) {
                    candidateLength = 0;
                    return -1;
                }
                candidatePrompt[candidateLength++] = (byte) readIncomingByteOrThrow();
            }
        }

        private boolean matchesPromptPrefix() {
            for (int i = 0; i < candidateLength; i++) {
                if (candidatePrompt[i] != DEVICE_PROMPT_BYTES[i]) {
                    return false;
                }
            }
            return true;
        }

        private int shiftCandidateByte() {
            int nextByte = candidatePrompt[0] & 0xFF;
            if (candidateLength > 1) {
                System.arraycopy(candidatePrompt, 1, candidatePrompt, 0, candidateLength - 1);
            }
            candidateLength--;
            return nextByte;
        }
    }

    private final class ChunkedRpcReplyInputStream extends RpcReplyInputStream {

        private int remainingChunkBytes;

        @Override
        protected int readPayloadByte() throws IOException {
            while (remainingChunkBytes == 0) {
                String header = readChunkHeaderLine();
                if (header.isEmpty()) {
                    continue;
                }
                if ("##".equals(header)) {
                    return -1;
                }
                if (!header.startsWith("#")) {
                    throw new NetconfException("Invalid NETCONF chunked framing header: " + header);
                }
                try {
                    remainingChunkBytes = Integer.parseInt(header.substring(1));
                } catch (NumberFormatException e) {
                    throw new NetconfException("Invalid NETCONF chunk size: " + header, e);
                }
                if (remainingChunkBytes < 0) {
                    throw new NetconfException("Invalid NETCONF chunk size: " + header);
                }
            }
            remainingChunkBytes--;
            return readIncomingByteOrThrow();
        }
    }

    private static int indexOf(byte[] haystack, byte[] needle) {
        outer:
        for (int i = 0; i <= haystack.length - needle.length; i++) {
            for (int j = 0; j < needle.length; j++) {
                if (haystack[i + j] != needle[j]) {
                    continue outer;
                }
            }
            return i;
        }
        return -1;
    }
}