import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.StringReader;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Iterator;

import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;

import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.xml.sax.InputSource;
import org.xml.sax.SAXException;

import com.adeptia.indigo.logging.Logger;
import com.adeptia.indigo.security.AuthUtil;
import com.adeptia.indigo.services.ServiceException;
import com.adeptia.indigo.services.transform.ScriptedService;
import com.adeptia.indigo.services.transport.connector.DatabaseConnectionInfo;
import com.adeptia.indigo.storage.Criteria;
import com.adeptia.indigo.storage.EntityManager;
import com.adeptia.indigo.storage.EntityManagerFactory;
import com.adeptia.indigo.utils.JdbcUtils;


		Connection connection = null;
		Statement stmt = null;
		PreparedStatement psStmt = null;
		Statement checkStmt = null;
		ResultSet checkResultSet = null;
		String databaseInfoName = "operations";
		NodeList childnodes = null;
		NodeList parrentNode = null;
		Node firstParrent = null;
		Document doc = null;
		StringBuffer tempXml = null;
		String xmlData = null;
		String infoTableName = "";
		String tableName = "";
		String sqlTableScript = "CREATE TABLE ";
		String dataType = " varchar(200),";
		String insertQuery = "";
		String checkQuery = "";
		String databasetype = "";
		String quote = "\"";
		boolean isTableExist = false;
		int columnCount = 0;
		int totalRecordCount = 0;
		Logger log = service.getLogger();
		InputStream inputStream = null;

		try {
			inputStream = service.getSourceStream();
			BufferedReader br = new BufferedReader(new InputStreamReader(
					inputStream));
			tempXml = new StringBuffer();
			String lineData = null;
			while ((lineData = br.readLine()) != null) {
				tempXml.append(lineData);
			}
			xmlData = tempXml.toString();
			if (inputStream != null)
				inputStream.close();
		} catch (IOException e) {
			log.error("Error in reading xml stream :: " + e.getMessage(), e);
		} finally {
			try {
				inputStream.close();
			} catch (IOException e) {
				// ignore error
			}
		}

		/*
		 * 
		 * 
		 * This code is used to get the connection from the database using the
		 * 
		 * 
		 * database info name which is passed by the user.
		 */
		try {
			DatabaseConnectionInfo databaseConnectionInfo = null;
			EntityManager entityManager = EntityManagerFactory
					.getEntityManager(DatabaseConnectionInfo.class,
							AuthUtil.getAdminSubject());
			Criteria criteria = new Criteria("p.entityName=\""
					+ databaseInfoName + "\"");
			Iterator it = entityManager.retrieve(criteria);
			if (it != null) {
				if (it.hasNext()) {
					databaseConnectionInfo = (DatabaseConnectionInfo) it.next();
				} else {
					throw new ServiceException(
							"Connection information not available");
				}
			}
			connection = JdbcUtils.getConnection(databaseConnectionInfo,
					AuthUtil.getAdminSubject());
		} catch (Exception e) {
			log.error(
					"Error in getting database connection :: " + e.getMessage(),
					e);
		}

		/*
		 * 
		 * 
		 * Parsing of xml data to generate the table script
		 */

		try {
			DatabaseMetaData db = connection.getMetaData();
			databasetype = db.getDatabaseProductName();
		} catch (SQLException e1) {
			e1.printStackTrace();
		}
		if (databasetype.equalsIgnoreCase("Oracle")) {
			infoTableName = "USER_TABLES";
		} else {
			infoTableName = "INFORMATION_SCHEMA.TABLES";
		}
		if (databasetype.equalsIgnoreCase("MySQL")) {
			quote = "`";
		}
		checkQuery = "SELECT 1 FROM " + infoTableName + " WHERE TABLE_NAME='";
		try {
			DocumentBuilderFactory documentBuilderFactory = DocumentBuilderFactory
					.newInstance();
			DocumentBuilder documentBuilder = documentBuilderFactory
					.newDocumentBuilder();
			InputSource is = new InputSource();
			is.setCharacterStream(new StringReader(xmlData));
			doc = documentBuilder.parse(is);
			totalRecordCount = doc.getDocumentElement().getChildNodes()
					.getLength();
		} catch (ParserConfigurationException e) {
			log.error("Error in xml parsing:: " + e.getMessage(), e);
		} catch (SAXException e) {
			log.error("Error in xml parsing:: " + e.getMessage(), e);
		} catch (IOException e) {
			log.error("Error in xml parsing:: " + e.getMessage(), e);
		}
		if (totalRecordCount > 0) {
			try {
				firstParrent = doc.getDocumentElement().getFirstChild();
				//tableName = firstParrent.getNodeName();
				tableName = context.get("tablename");
				childnodes = firstParrent.getChildNodes();
				checkQuery = checkQuery + tableName + "'";
				insertQuery = "INSERT INTO " + tableName + "(";
				sqlTableScript = sqlTableScript + tableName + "(";
				for (int i = 0; i < childnodes.getLength(); i++) {
					Node node = childnodes.item(i);
					if (node instanceof Element) {
						Element childElement = (Element) node;
						sqlTableScript = sqlTableScript + quote
								+ childElement.getNodeName() + quote + dataType;
						insertQuery = insertQuery + quote
								+ childElement.getNodeName() + quote + ",";
						columnCount++;
					}
				}
				insertQuery = insertQuery.substring(0,
						insertQuery.lastIndexOf(","))
						+ ")";
				insertQuery = insertQuery + " VALUES(";
				for (int i = 0; i < columnCount; i++) {
					insertQuery = insertQuery + "?,";
				}
				insertQuery = insertQuery.substring(0,
						insertQuery.lastIndexOf(","))
						+ ")";
				sqlTableScript = sqlTableScript.substring(0,
						sqlTableScript.lastIndexOf(","));
				sqlTableScript = sqlTableScript + ")";
				checkStmt = connection.createStatement();
				checkResultSet = checkStmt.executeQuery(checkQuery);
				isTableExist = checkResultSet.next();
				if (!isTableExist) {
					stmt = connection.createStatement();
					int isTableCreated = stmt.executeUpdate(sqlTableScript);
					log.debug("Table: " + tableName
							+ " has been created successfully.");
				} else{
					log.debug("Table: " + tableName + " is already exist.");
				}
			} catch (SQLException e) {
				log.error(
						"Error while executing the query :: " + e.getMessage(),
						e);
			} finally {
				try {
					if (stmt != null)
						stmt.close();
					if (checkStmt != null)
						checkStmt.close();
					if (checkResultSet != null)
						checkResultSet.close();
				} catch (Exception e) {
					log.error(
							"Error in closing the database object:: "
									+ e.getMessage(), e);
				}
			}

			/*
			 * 
			 * 
			 * This block of code is used to insert the data into the database
			 * 
			 * 
			 * table from the xml stream
			 */
			try {
				psStmt = connection.prepareStatement(insertQuery);
				int count = 0;
				int totalNumOfInsertedRecord = 0;
				parrentNode = doc.getDocumentElement().getChildNodes();
				for (int i = 0; i < parrentNode.getLength(); i++) {
					NodeList nodeList = parrentNode.item(i).getChildNodes();
					for (int j = 0; j < nodeList.getLength(); j++) {
						Node childnode = nodeList.item(j);
						if (childnode instanceof Element) {
							count++;
							Element childElement = (Element) childnode;
							psStmt.setString(count,
									childElement.getTextContent());
						}
					}
					int status = psStmt.executeUpdate();
					if(status > 0){
					  totalNumOfInsertedRecord++ ;
					}
					count = 0;
				}
				log.debug("Total no. of rows inserted in the table " + tableName
						+ " is: " + totalNumOfInsertedRecord);
				psStmt.executeBatch();
			} catch (SQLException e) {
				log.error(
						"Error while inserting the data into the database table :: "
								+ e.getMessage(), e);
			} finally {
				try {
					if (psStmt != null)
						psStmt.close();
					if (connection != null)
						connection.close();
				} catch (SQLException e) {
					log.error(
							"Error in closing the database object :: "
									+ e.getMessage(), e);
				}
			}
		} else {
			try {
				if (connection != null)
					connection.close();
			} catch (SQLException e) {
				log.error(
						"Error in closing the database object :: "
								+ e.getMessage(), e);
			}
		}