/*
 *  Licensed to the Apache Software Foundation (ASF) under one
 *  or more contributor license agreements.  See the NOTICE file
 *  distributed with this work for additional information
 *  regarding copyright ownership.  The ASF licenses this file
 *  to you under the Apache License, Version 2.0 (the
 *  "License"); you may not use this file except in compliance
 *  with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 */
package org.wso2.carbon.cassandra.dataaccess;

import org.apache.axiom.om.OMElement;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.wso2.carbon.utils.CarbonUtils;

import javax.xml.namespace.QName;

/**
 * A factory method to create a <code>ClusterConfiguration</code> instance from the XML configuration
 */
public class ClusterConfigurationFactory {

    private static Log log = LogFactory.getLog(ClusterConfigurationFactory.class);

    /**
     * Create an instance of <code>ClusterConfiguration</code> from the XML configuration
     *
     * @param severElement XML representing the server's configuration
     * @return Not null <code>ClusterConfiguration</code> instance
     */
    public static ClusterConfiguration create(OMElement severElement) {

        ClusterConfiguration clusterConfiguration = new ClusterConfiguration();

        OMElement deployment = severElement.getFirstChildWithName(new QName("Deployment"));
        if (deployment != null) {
            String mode = deployment.getText();
            if (mode != null && !"".equals(mode.trim())) {
                clusterConfiguration.setDeploymentMode(mode.trim());
            }
        }

        OMElement cluster = severElement.getFirstChildWithName(new QName("Cluster"));
        if (cluster != null) {
            OMElement nameElement = cluster.getFirstChildWithName(new QName("Name"));
            if (nameElement != null) {
                String name = nameElement.getText();
                if (name == null || "".equals(name.trim())) {
                    throw new DataAccessComponentException("Name is null or empty", log);
                }
                clusterConfiguration.setClusterName(name.trim());
            }

            OMElement portElement = cluster.getFirstChildWithName(new QName("DefaultPort"));
            if (portElement != null) {
                String port = portElement.getText();
                if (port != null && !"".equals(port.trim())) {
                    try {
                        clusterConfiguration.setDefaultPort(Integer.parseInt(port.trim()));
                    } catch (NumberFormatException e) {
                        throw new DataAccessComponentException("An invalid number for defaultport : " + port, log);
                    }
                }
            }

            OMElement nodesElement = cluster.getFirstChildWithName(new QName("Nodes"));
            if (nodesElement != null) {
                Boolean externalCassandra=Boolean.parseBoolean(nodesElement.getAttributeValue(new QName("externalCassandra")).trim());
                String nodesString = nodesElement.getText();
                if (nodesString != null && !"".endsWith(nodesString.trim())) {
                    if(externalCassandra) {
                        clusterConfiguration.setNodesString(nodesString.trim());
                        String nodes[] = nodesString.split(",");
                        for (String node : nodes) {
                            clusterConfiguration.addNode(node);
                        }
                    } else {
                        String host=nodesString.split(":")[0];
                        int port=Integer.parseInt(nodesString.split(":")[1].trim());
                        port=port+getPortOffset();
                        String newNodeString=host+":"+port;
                        clusterConfiguration.setNodesString(newNodeString.trim());
                        clusterConfiguration.addNode(newNodeString.trim());
                    }
                }
            }

            OMElement autoDiscoveryElement = cluster.getFirstChildWithName(new QName("AutoDiscovery"));
            if (autoDiscoveryElement != null) {
                String disable = autoDiscoveryElement.getAttributeValue(new QName("disable"));
                if (disable != null && !"".equals(disable.trim())) {
                    clusterConfiguration.setAutoDiscovery(!Boolean.parseBoolean(disable.trim()));
                }
                String delay = autoDiscoveryElement.getAttributeValue(new QName("delay"));
                if (delay != null && !"".equals(delay.trim())) {
                    int delayAsInt = -1;
                    try {
                        delayAsInt = Integer.parseInt(delay.trim());
                    } catch (NumberFormatException ignored) {
                    }
                    if (delayAsInt > 0) {
                        clusterConfiguration.setAutoDiscoveryDelay(delayAsInt);
                    }
                }
            }
        }

        return clusterConfiguration;
    }

    /**
     * Read Carbon Server port offset
     * @return offset number
     */
    private static int getPortOffset() {
        String portOffset = System.getProperty("portOffset",
                CarbonUtils.getServerConfiguration().getFirstProperty("Ports.Offset"));
        return Integer.parseInt(portOffset);
    }
}

@@ -1089,16 +1089,44 @@
 .QName;%0D
+%0Aimport java.util.Iterator;%0D
 %0A%0D%0A/**%0D%0A
@@ -3269,32 +3269,206 @@
            %7D%0D%0A%0D%0A
+            /**%0D%0A             * If externalCassandra attribute is not set in the hector configuration, it will assign default value as true%0D%0A             *%0D%0A             */%0D%0A
             OMEl
@@ -3605,17 +3605,19 @@
 Boolean 
-e
+isE
 xternalC
@@ -3629,100 +3629,557 @@
 dra=
-Boolean.parseBoolean(nodesElement.getAttributeValue(new QName(%22externalCassandra%22)).trim());
+false;%0D%0A                Iterator iterator=nodesElement.getAllAttributes();%0D%0A                String externalCassandra=%22%22;%0D%0A                while (iterator.hasNext()) %7B%0D%0A                    externalCassandra=nodesElement.getAttributeValue(new QName(%22externalCassandra%22));%0D%0A                %7D%0D%0A                if(externalCassandra!=null && !%22%22.equals(externalCassandra)) %7B%0D%0A                    isExternalCassandra=Boolean.parseBoolean(externalCassandra.trim());%0D%0A                %7D else %7B%0D%0A                    isExternalCassandra=true;%0D%0A                %7D
 %0D%0A  
@@ -4333,33 +4333,35 @@
              if(
-e
+isE
 xternalCassandra

Make external cassandra property optional as per SS-367