Metacat

/**

 *  '$RCSfile$'

 *    Purpose: A Class that handles the SAX XML events as they

 *             are generated from XML documents

 *  Copyright: 2000 Regents of the University of California and the

 *             National Center for Ecological Analysis and Synthesis

 *    Authors: Matt Jones, Jivka Bojilova

 *

 *   '$Author: daigle $'

 *     '$Date: 2009-10-16 11:10:37 -0700 (Fri, 16 Oct 2009) $'

 * '$Revision: 5090 $'

 *

 * This program is free software; you can redistribute it and/or modify

 * it under the terms of the GNU General Public License as published by

 * the Free Software Foundation; either version 2 of the License, or

 * (at your option) any later version.

 *

 * This program is distributed in the hope that it will be useful,

 * but WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

 * GNU General Public License for more details.

 *

 * You should have received a copy of the GNU General Public License

 * along with this program; if not, write to the Free Software

 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

 */

package edu.ucsb.nceas.metacat;

import java.io.BufferedReader;

import java.io.File;

import java.io.FileReader;

import java.io.FileWriter;

import java.io.IOException;

import java.io.Reader;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.SQLException;

import java.sql.Statement;

import java.util.EmptyStackException;

import java.util.Enumeration;

import java.util.Hashtable;

import java.util.Stack;

import java.util.Vector;

import org.apache.log4j.Logger;

import org.xml.sax.Attributes;

import org.xml.sax.SAXException;

import edu.ucsb.nceas.metacat.accesscontrol.AccessControlInterface;

import edu.ucsb.nceas.metacat.accesscontrol.AccessRule;

import edu.ucsb.nceas.metacat.accesscontrol.AccessSection;

import edu.ucsb.nceas.metacat.database.DBConnection;

import edu.ucsb.nceas.metacat.database.DBConnectionPool;

import edu.ucsb.nceas.metacat.properties.PropertyService;

import edu.ucsb.nceas.metacat.util.DocumentUtil;

import edu.ucsb.nceas.metacat.util.MetacatUtil;

import edu.ucsb.nceas.utilities.PropertyNotFoundException;

/**

 * A database aware Class implementing callback methods for the SAX parser to

 * call when processing the XML stream and generating events

 */

public class Eml210SAXHandler extends DBSAXHandler implements AccessControlInterface {

	private boolean processTopLevelAccess = false;

	private boolean processAdditionalAccess = false;

	private boolean processOtherAccess = false;

	private AccessSection accessObject = null;

	private AccessRule accessRule = null;

	private Vector<AccessSection> accessObjectList = new Vector<AccessSection>(); // store

																					// every

																					// access

																					// rule

	private Hashtable<String, AccessSection> topLevelAccessControlMap = new Hashtable<String, AccessSection>();

	private Hashtable<String, AccessSection> additionalAccessControlMap = new Hashtable<String, AccessSection>();// store

	// subtree access for single additionalmetacat

	private Vector<Hashtable<String, AccessSection>> additionalAccessMapList = new Vector<Hashtable<String, AccessSection>>();// store

																																// maps

																																// for

	// every additionalmetadata

	private Vector<String> describesId = new Vector<String>(); // store the ids

																// in

	// additionalmetadata/describes

	private Stack<SubTree> subTreeInfoStack = new Stack<SubTree>();

	private Vector<SubTree> subTreeList = new Vector<SubTree>();// store the

																// final subtree

	// MCD - Removed the following lines. They are used for node level access

	// control

	// which is not yet implemented in EML 2.1.0

	// private Hashtable<String,SubTree> unChangeableSubTreeHash = new

	// Hashtable<String,SubTree>();

	// private Stack<NodeRecord> currentUnChangeableSubtreeNodeStack = new

	// Stack<NodeRecord>();

	// private boolean startCriticalSubTree = false;

	// private boolean firstElementForCriticalSubTree = false;

	// private String firstElementNameForCriticalSubTree;

	private boolean needToCheckAccessModule = false;

	private Vector<AccessSection> unChangeableAccessSubTreeVector = new Vector<AccessSection>();

	private Stack<NodeRecord> currentUnchangeableAccessModuleNodeStack = new Stack<NodeRecord>();

	private AccessSection topAccessSection;

	// we need an another stack to store the access node which we pull out just

	// from xml. If a reference happened, we can use the stack the compare nodes

	private Stack<NodeRecord> storedAccessNodeStack = new Stack<NodeRecord>();

	// vector stored the data file id which will be write into relation table

	private Vector<String> onlineDataFileIdInRelationVector = new Vector<String>();

	// vector stored the data file id which will be write top access rules to

	// access table

	private Vector<String> onlineDataFileIdInTopAccessVector = new Vector<String>();

	// Indicator of inline data

	private boolean handleInlineData = false;

	private Hashtable<String, String> inlineDataNameSpace = null;

	private FileWriter inlineDataFileWriter = null;

	private String inlineDataFileName = null;

	DistributionSection currentDistributionSection = null;

	Vector<DistributionSection> allDistributionSections = new Vector<DistributionSection>();

	// This variable keeps a counter of each distribution element. This index

	// will be used to name the inline data file that gets written to disk, and to

	// strip inline data from the metadata document on file if the user does not have

	// read access.

	private int distributionIndex = 0;

	// private String distributionOnlineFileName = null;

	// This is used to delete inline files if the xml does not parse correctly.

	private Vector<String> inlineFileIdList = new Vector<String>();

	// Constant

	private static final String EML = "eml";

	private static final String DISTRIBUTION = "distribution";

	private static final String ORDER = "order";

	private static final String ID = "id";

	private static final String REFERENCES = "references";

	public static final String INLINE = "inline";

	private static final String ONLINE = "online";

	private static final String URL = "url";

	// private static final String PERMISSIONERROR = "User tried to update a

	// subtree "

	// + "when they don't have write permission!";

	private static final String UPDATEACCESSERROR = "User tried to update an "

			+ "access module when they don't have \"ALL\" permission!";

	private static final String TOPLEVEL = "top";

	private static final String SUBTREELEVEL = "subtree";

	private static final String RELATION = "Provides info for";

	private Logger logMetacat = Logger.getLogger(Eml210SAXHandler.class);

	/**

	 * Construct an instance of the handler class In this constructor, user can

	 * specify the version need to update

	 * 

	 * @param conn

	 *            the JDBC connection to which information is written

	 * @param action -

	 *            "INSERT" or "UPDATE"

	 * @param docid

	 *            to be inserted or updated into JDBC connection

	 * @param revision,

	 *            the user specified the revision need to be update

	 * @param user

	 *            the user connected to MetaCat servlet and owns the document

	 * @param groups

	 *            the groups to which user belongs

	 * @param pub

	 *            flag for public "read" access on document

	 * @param serverCode

	 *            the serverid from xml_replication on which this document

	 *            resides.

	 * 

	 */

	public Eml210SAXHandler(DBConnection conn, String action, String docid,

			String revision, String user, String[] groups, String pub, int serverCode,

			String createDate, String updateDate) throws SAXException {

		super(conn, action, docid, revision, user, groups, pub, serverCode, createDate,

				updateDate);

		// Get the unchangeable subtrees (user doesn't have write permission)

		try {

			PermissionController control = new PermissionController(docid

					+ PropertyService.getProperty("document.accNumSeparator") + revision);

			// unChangableSubTreeHash = getUnchangableSubTree(control, user,

			// groups);

			// If the action is update and user doesn't have "ALL" permission

			// we need to check if user can update access subtree			

			if (!control.hasPermission(user, groups, AccessControlInterface.ALLSTRING)

					&& action != null && action.equals("UPDATE")) {

				needToCheckAccessModule = true;

				unChangeableAccessSubTreeVector = getAccessSubTreeListFromDB();

			}

		} catch (Exception e) {

			throw new SAXException(e.getMessage());

		}

	}

	// MCD - Removed the following method. It is used for node level access

	// control

	// which is not yet implemented in EML 2.1.0

	// /* Pass a permission control and get the list of unchangable subtree */

	// private Hashtable getUnchangableSubTree(PermissionController controller,

	// String user, String[] groups) throws Exception

	// {

	// Hashtable list = null;

	// Hashtable result = new Hashtable();

	// // get unwritable sutree from controller

	// list = null;

	// // changed after old code removal

	// //controller.hasUnaccessableSubTree(user, groups,

	// // AccessControlInterface.WRITESTRING);

	// if (list != null) {

	//

	// Enumeration en = list.elements();

	// while (en.hasMoreElements()) {

	// // Get a subtree without node record list

	// SubTree treeWithoutStack = (SubTree) en.nextElement();

	// String subTreeId = treeWithoutStack.getSubTreeId();

	// logMetacat.info(

	// "unchangable subtree id: " + subTreeId);

	// long startNodeId = treeWithoutStack.getStartNodeId();

	// logMetacat.info("unchangable subtree startnodeid: "

	// + startNodeId);

	// long endNodeId = treeWithoutStack.getEndNodeId();

	// logMetacat.info("unchangable subtree endnodeid: "

	// + endNodeId);

	// // Get a subtree has the nodelist

	// SubTree tree = new SubTree(docid, subTreeId, startNodeId,

	// endNodeId);

	// // add this tree to the result

	// result.put(subTreeId, tree);

	//

	// }//while

	//

	// }//if

	//

	// return result;

	// }

	/*

	 * Get the subtree node info from xml_accesssubtree table

	 */

	private Vector<AccessSection> getAccessSubTreeListFromDB() throws Exception {

		Vector<AccessSection> result = new Vector<AccessSection>();

		PreparedStatement pstmt = null;

		ResultSet rs = null;

		String sql = "SELECT controllevel, subtreeid, startnodeid, endnodeid "

				+ "FROM xml_accesssubtree WHERE docid like ? "

				+ "ORDER BY startnodeid ASC";

		try {

			pstmt = connection.prepareStatement(sql);

			// Increase DBConnection usage count

			connection.increaseUsageCount(1);

			// Bind the values to the query

			pstmt.setString(1, docid);

			pstmt.execute();

			// Get result set

			rs = pstmt.getResultSet();

			while (rs.next()) {

				String level = rs.getString(1);

				String sectionId = rs.getString(2);

				long startNodeId = rs.getLong(3);

				long endNodeId = rs.getLong(4);

				// create a new access section

				AccessSection accessObj = new AccessSection();

				accessObj.setControlLevel(level);

				accessObj.setDocId(docid);

				accessObj.setSubTreeId(sectionId);

				accessObj.setStartNodeId(startNodeId);

				accessObj.setEndNodeId(endNodeId);

				Stack<NodeRecord> nodeStack = accessObj.getSubTreeNodeStack();

				accessObj.setSubTreeNodeStack(nodeStack);

				// add this access obj into vector

				result.add(accessObj);

				// Get the top level access subtree control

				if (level != null && level.equals(TOPLEVEL)) {

					topAccessSection = accessObj;

				}

			}

			pstmt.close();

		}// try

		catch (SQLException e) {

			throw new SAXException("EMLSAXHandler.getAccessSubTreeListFromDB(): "

					+ e.getMessage());

		}// catch

		finally {

			try {

				pstmt.close();

			} catch (SQLException ee) {

				throw new SAXException("EMLSAXHandler.getAccessSubTreeListFromDB(): "

						+ ee.getMessage());

			}

		}// finally

		return result;

	}

	/** SAX Handler that is called at the start of each XML element */

	public void startElement(String uri, String localName, String qName, Attributes atts)

			throws SAXException {

		// for element <eml:eml...> qname is "eml:eml", local name is "eml"

		// for element <acl....> both qname and local name is "eml"

		// uri is namesapce

		logMetacat.debug("Start ELEMENT(qName) " + qName);

		logMetacat.debug("Start ELEMENT(localName) " + localName);

		logMetacat.debug("Start ELEMENT(uri) " + uri);

		DBSAXNode parentNode = null;

		DBSAXNode currentNode = null;

		if (!handleInlineData) {

			// Get a reference to the parent node for the id

			try {

				parentNode = (DBSAXNode) nodeStack.peek();

			} catch (EmptyStackException e) {

				parentNode = null;

			}

			// start handle inline data

			if (localName.equals(INLINE)) {

				handleInlineData = true;

				// initialize namespace hash for in line data

				inlineDataNameSpace = new Hashtable<String, String>();

				// initialize file writer

				String docidWithoutRev = DocumentUtil.getDocIdFromString(docid);

				String seperator = ".";

				try {

					seperator = PropertyService.getProperty("document.accNumSeparator");

				} catch (PropertyNotFoundException pnfe) {

					logMetacat.error("Could not fing property 'accNumSeparator'. "

							+ "Setting separator to '.': " + pnfe.getMessage());

				}

				// the new file name will look like docid.rev.2

				inlineDataFileName = docidWithoutRev + seperator + revision + seperator

						+ distributionIndex;

				inlineDataFileWriter = createInlineDataFileWriter(inlineDataFileName);

				// put the inline file id into a vector. If upload failed,

				// metacat will delete the inline data file

				inlineFileIdList.add(inlineDataFileName);

				currentDistributionSection.setDistributionType(DistributionSection.INLINE_DATA_DISTRIBUTION);

				currentDistributionSection.setDataFileName(inlineDataFileName);

			}

			// If hit a text node, we need write this text for current's parent

			// node This will happen if the element is mixed

			if (hitTextNode && parentNode != null) {

				// MCD - Removed the following method. It is used for node level

				// access control

				// which is not yet implemented in EML 2.1.0

				// //compare text node data for unchangeablesubtree

				// if (startCriticalSubTree) {

				// compareTextNode(currentUnChangeableSubtreeNodeStack,

				// textBuffer, PERMISSIONERROR);

				// }//if

				// compare top level access module

				if (processTopLevelAccess && needToCheckAccessModule) {

					compareTextNode(currentUnchangeableAccessModuleNodeStack, textBuffer,

							UPDATEACCESSERROR);

				}

				if (needToCheckAccessModule

						&& (processAdditionalAccess || processOtherAccess || processTopLevelAccess)) {

					// stored the pull out nodes into storedNode stack

					NodeRecord nodeElement = new NodeRecord(-2, -2, -2, "TEXT", null,

							null, MetacatUtil.normalize(textBuffer.toString()));

					storedAccessNodeStack.push(nodeElement);

				}

				// write the textbuffer into db for parent node.

				endNodeId = writeTextForDBSAXNode(endNodeId, textBuffer, parentNode);

				// rest hitTextNode

				hitTextNode = false;

				// reset textbuffer

				textBuffer = null;

				textBuffer = new StringBuffer();

			}

			// Document representation that points to the root document node

			if (atFirstElement) {

				atFirstElement = false;

				// If no DOCTYPE declaration: docname = root element

				// doctype = root element name or name space

				if (docname == null) {

					docname = localName;

					// if uri isn't null doctype = uri(namespace)

					// othewise root element

					if (uri != null && !(uri.trim()).equals("")) {

						doctype = uri;

					} else {

						doctype = docname;

					}

					logMetacat.debug("DOCNAME-a: " + docname);

					logMetacat.debug("DOCTYPE-a: " + doctype);

				} else if (doctype == null) {

					// because docname is not null and it is declared in dtd

					// so could not be in schema, no namespace

					doctype = docname;

					logMetacat.debug("DOCTYPE-b: " + doctype);

				}

				rootNode.writeNodename(docname);

				try {

					// for validated XML Documents store a reference to XML DB

					// Catalog. Because this is select statement and it needn't

					// roll back if insert document action failed. In order to

					// decrease DBConnection usage count, we get a new

					// dbconnection from pool String catalogid = null;

					DBConnection dbConn = null;

					int serialNumber = -1;

					try {

						// Get dbconnection

						dbConn = DBConnectionPool

								.getDBConnection("DBSAXHandler.startElement");

						serialNumber = dbConn.getCheckOutSerialNumber();

						Statement stmt = dbConn.createStatement();

						ResultSet rs = stmt

								.executeQuery("SELECT catalog_id FROM xml_catalog "

										+ "WHERE entry_type = 'Schema' "

										+ "AND public_id = '" + doctype + "'");

						boolean hasRow = rs.next();

						if (hasRow) {

							catalogid = rs.getString(1);

						}

						stmt.close();

					}// try

					finally {

						// Return dbconnection

						DBConnectionPool.returnDBConnection(dbConn, serialNumber);

					}// finally

					// create documentImpl object by the constructor which can

					// specify the revision

					if (!super.getIsRevisionDoc()) {

						currentDocument = new DocumentImpl(connection, rootNode

								.getNodeID(), docname, doctype, docid, revision, action,

								user, this.pub, catalogid, this.serverCode, createDate,

								updateDate);

					}

				} catch (Exception ane) {

					throw (new SAXException("Error in EMLSaxHandler.startElement "

							+ action, ane));

				}

			}

			// Create the current node representation

			currentNode = new DBSAXNode(connection, qName, localName, parentNode,

					rootNode.getNodeID(), docid, doctype);

			// Use a local variable to store the element node id

			// If this element is a start point of subtree(section), it will be

			// stored otherwise, it will be discarded

			long startNodeId = currentNode.getNodeID();

			// Add all of the namespaces

			String prefix = null;

			String nsuri = null;

			Enumeration<String> prefixes = namespaces.keys();

			while (prefixes.hasMoreElements()) {

				prefix = prefixes.nextElement();

				nsuri = namespaces.get(prefix);

				endNodeId = currentNode.setNamespace(prefix, nsuri, docid);

			}

			// Add all of the attributes

			for (int i = 0; i < atts.getLength(); i++) {

				String attributeName = atts.getQName(i);

				String attributeValue = atts.getValue(i);

				endNodeId = currentNode

						.setAttribute(attributeName, attributeValue, docid);

				// To handle name space and schema location if the attribute

				// name is xsi:schemaLocation. If the name space is in not

				// in catalog table it will be registered.

				if (attributeName != null

						&& attributeName.indexOf(MetaCatServlet.SCHEMALOCATIONKEYWORD) != -1) {

					SchemaLocationResolver resolver = new SchemaLocationResolver(

							attributeValue);

					resolver.resolveNameSpace();

				} else if (attributeName != null && attributeName.equals(ID)) {

					// MCD - Removed the following code. It is used for node

					// level access

					// control which is not yet implemented in EML 2.1.0

					// // check unchangeable subtree hash if contains this

					// // subtree id

					// if (unChangeableSubTreeHash.containsKey(attributeValue))

					// {

					// // this subtree couldn't be changed by the user and

					// // move it from hash

					// SubTree currentUnChangedableSubtree =

					// unChangeableSubTreeHash

					// .remove(attributeValue);

					// currentUnChangeableSubtreeNodeStack =

					// currentUnChangedableSubtree

					// .getSubTreeNodeStack();

					// startCriticalSubTree = true;

					// firstElementForCriticalSubTree = true;

					// }

				}

			}// for

			// handle access stuff

			if (localName.equals(ACCESS)) {

				if (parentNode.getTagName().equals(EML)) {

					processTopLevelAccess = true;

				} else if (parentNode.getTagName() == DISTRIBUTION) {

					processAdditionalAccess = true;

				} else {

					// process other access embedded into resource level

					// module

					processOtherAccess = true;

				}

				// create access object

				accessObject = new AccessSection();

				// set permission order

				String permOrder = currentNode.getAttribute(ORDER);

				accessObject.setPermissionOrder(permOrder);

				// set access id

				String accessId = currentNode.getAttribute(ID);

				accessObject.setSubTreeId(accessId);

				accessObject.setStartNodeId(startNodeId);

				accessObject.setDocId(docid);

				if (processAdditionalAccess) {

					accessObject.setDataFileName(inlineDataFileName);

				}

				// load top level node stack to

				// currentUnchangableAccessModuleNodeStack

				if (processTopLevelAccess && needToCheckAccessModule) {

					// get the node stack for

					currentUnchangeableAccessModuleNodeStack = topAccessSection

							.getSubTreeNodeStack();

				}

			} else if (localName.equals(DISTRIBUTION)) {

				distributionIndex++;

				currentDistributionSection = new DistributionSection(distributionIndex);

				// handle subtree info

				SubTree subTree = new SubTree();

				// set sub tree id

				subTree.setSubTreeId(String.valueOf(distributionIndex));

				// set sub tree start element name

				subTree.setStartElementName(currentNode.getTagName());

				// set start node number

				subTree.setStartNodeId(startNodeId);

				// add to stack, but it didn't get end node id yet

				subTreeInfoStack.push(subTree);

			}

			// Set up a access rule for allow

			else if (parentNode.getTagName() != null

					&& (parentNode.getTagName()).equals(ACCESS)

					&& localName.equals(ALLOW)) {

				accessRule = new AccessRule();

				// set permission type "allow"

				accessRule.setPermissionType(ALLOW);

			}

			// set up an access rule for deny

			else if (parentNode.getTagName() != null

					&& (parentNode.getTagName()).equals(ACCESS) && localName.equals(DENY)) {

				accessRule = new AccessRule();

				// set permission type "allow"

				accessRule.setPermissionType(DENY);

			}

			// Add the node to the stack, so that any text data can be

			// added as it is encountered

			nodeStack.push(currentNode);

			// Add the node to the vector used by thread for writing XML Index

			nodeIndex.addElement(currentNode);

			// MCD - Removed the following code. It is used for node level

			// access

			// control which is not yet implemented in EML 2.1.0

			// // handle critical subtree

			// if (startCriticalSubTree && firstElementForCriticalSubTree) {

			// //store the element name

			// firstElementNameForCriticalSubTree = qName;

			// firstElementForCriticalSubTree = false;

			// }//for first element

			//

			// // handle critical subtree

			// if (startCriticalSubTree) {

			// compareElementNameSpaceAttributes(

			// currentUnChangeableSubtreeNodeStack, namespaces, atts,

			// localName, PERMISSIONERROR);

			// 

			// }

			// compare top access level module

			if (processTopLevelAccess && needToCheckAccessModule) {

				compareElementNameSpaceAttributes(

						currentUnchangeableAccessModuleNodeStack, namespaces, atts,

						localName, UPDATEACCESSERROR);

			}

			// store access module element and attributes into stored stack

			if (needToCheckAccessModule

					&& (processAdditionalAccess || processOtherAccess || processTopLevelAccess)) {

				// stored the pull out nodes into storedNode stack

				NodeRecord nodeElement = new NodeRecord(-2, -2, -2, "ELEMENT", localName,

						prefix, MetacatUtil.normalize(null));

				storedAccessNodeStack.push(nodeElement);

				for (int i = 0; i < atts.getLength(); i++) {

					String attributeName = atts.getQName(i);

					String attributeValue = atts.getValue(i);

					NodeRecord nodeAttribute = new NodeRecord(-2, -2, -2, "ATTRIBUTE",

							attributeName, null, MetacatUtil.normalize(attributeValue));

					storedAccessNodeStack.push(nodeAttribute);

				}

			}

			// reset name space

			namespaces = null;

			namespaces = new Hashtable<String, String>();

		}// not inline data

		else {

			// we don't buffer the inline data in characters() method

			// so start character don't need to hand text node.

			// inline data may be the xml format.

			StringBuffer inlineElements = new StringBuffer();

			inlineElements.append("<").append(qName);

			// append attributes

			for (int i = 0; i < atts.getLength(); i++) {

				String attributeName = atts.getQName(i);

				String attributeValue = atts.getValue(i);

				inlineElements.append(" ");

				inlineElements.append(attributeName);

				inlineElements.append("=\"");

				inlineElements.append(attributeValue);

				inlineElements.append("\"");

			}

			// append namespace

			String prefix = null;

			String nsuri = null;

			Enumeration<String> prefixes = inlineDataNameSpace.keys();

			while (prefixes.hasMoreElements()) {

				prefix = prefixes.nextElement();

				nsuri = namespaces.get(prefix);

				inlineElements.append(" ");

				inlineElements.append("xmlns:");

				inlineElements.append(prefix);

				inlineElements.append("=\"");

				inlineElements.append(nsuri);

				inlineElements.append("\"");

			}

			inlineElements.append(">");

			// reset inline data name space

			inlineDataNameSpace = null;

			inlineDataNameSpace = new Hashtable<String, String>();

			// write inline data into file

			logMetacat.debug("the inline element data is: " + inlineElements.toString());

			writeInlineDataIntoFile(inlineDataFileWriter, inlineElements);

		}// else

	}

	private void compareElementNameSpaceAttributes(

			Stack<NodeRecord> unchangeableNodeStack,

			Hashtable<String, String> nameSpaces, Attributes attributes,

			String localName, String error) throws SAXException {

		// Get element subtree node stack (element node)

		NodeRecord elementNode = null;

		try {

			elementNode = unchangeableNodeStack.pop();

		} catch (EmptyStackException ee) {

			logMetacat.error("Node stack is empty for element data");

			throw new SAXException(error);

		}

		logMetacat.debug("current node type from xml is ELEMENT");

		logMetacat.debug("node type from stack: " + elementNode.getNodeType());

		logMetacat.debug("node name from xml document: " + localName);

		logMetacat.debug("node name from stack: " + elementNode.getNodeName());

		logMetacat.debug("node data from stack: " + elementNode.getNodeData());

		logMetacat.debug("node id is: " + elementNode.getNodeId());

		// if this node is not element or local name not equal or name space

		// not equals, throw an exception

		if (!elementNode.getNodeType().equals("ELEMENT")

				|| !localName.equals(elementNode.getNodeName()))

		// (uri != null && !uri.equals(elementNode.getNodePrefix())))

		{

			logMetacat.error("Inconsistence happened: ");

			logMetacat.error("current node type from xml is ELEMENT");

			logMetacat.error("node type from stack: " + elementNode.getNodeType());

			logMetacat.error("node name from xml document: " + localName);

			logMetacat.error("node name from stack: " + elementNode.getNodeName());

			logMetacat.error("node data from stack: " + elementNode.getNodeData());

			logMetacat.error("node id is: " + elementNode.getNodeId());

			throw new SAXException(error);

		}

		// compare namespace

		Enumeration<String> nameEn = nameSpaces.keys();

		while (nameEn.hasMoreElements()) {

			// Get namespacke node stack (element node)

			NodeRecord nameNode = null;

			try {

				nameNode = unchangeableNodeStack.pop();

			} catch (EmptyStackException ee) {

				logMetacat.error("Node stack is empty for namespace data");

				throw new SAXException(error);

			}

			String prefixName = nameEn.nextElement();

			String nameSpaceUri = nameSpaces.get(prefixName);

			if (!nameNode.getNodeType().equals("NAMESPACE")

					|| !prefixName.equals(nameNode.getNodeName())

					|| !nameSpaceUri.equals(nameNode.getNodeData())) {

				logMetacat.error("Inconsistence happened: ");

				logMetacat.error("current node type from xml is NAMESPACE");

				logMetacat.error("node type from stack: " + nameNode.getNodeType());

				logMetacat.error("current node name from xml is: " + prefixName);

				logMetacat.error("node name from stack: " + nameNode.getNodeName());

				logMetacat.error("current node data from xml is: " + nameSpaceUri);

				logMetacat.error("node data from stack: " + nameNode.getNodeData());

				logMetacat.error("node id is: " + nameNode.getNodeId());

				throw new SAXException(error);

			}

		}// while

		// compare attributes

		for (int i = 0; i < attributes.getLength(); i++) {

			NodeRecord attriNode = null;

			try {

				attriNode = unchangeableNodeStack.pop();

			} catch (EmptyStackException ee) {

				logMetacat.error("Node stack is empty for attribute data");

				throw new SAXException(error);

			}

			String attributeName = attributes.getQName(i);

			String attributeValue = attributes.getValue(i);

			logMetacat.debug("current node type from xml is ATTRIBUTE ");

			logMetacat.debug("node type from stack: " + attriNode.getNodeType());

			logMetacat.debug("current node name from xml is: " + attributeName);

			logMetacat.debug("node name from stack: " + attriNode.getNodeName());

			logMetacat.debug("current node data from xml is: " + attributeValue);

			logMetacat.debug("node data from stack: " + attriNode.getNodeData());

			logMetacat.debug("node id  is: " + attriNode.getNodeId());

			if (!attriNode.getNodeType().equals("ATTRIBUTE")

					|| !attributeName.equals(attriNode.getNodeName())

					|| !attributeValue.equals(attriNode.getNodeData())) {

				logMetacat.error("Inconsistence happened: ");

				logMetacat.error("current node type from xml is ATTRIBUTE ");

				logMetacat.error("node type from stack: " + attriNode.getNodeType());

				logMetacat.error("current node name from xml is: " + attributeName);

				logMetacat.error("node name from stack: " + attriNode.getNodeName());

				logMetacat.error("current node data from xml is: " + attributeValue);

				logMetacat.error("node data from stack: " + attriNode.getNodeData());

				logMetacat.error("node is: " + attriNode.getNodeId());

				throw new SAXException(error);

			}

		}// for

	}

	/* method to compare current text node and node in db */

	private void compareTextNode(Stack<NodeRecord> nodeStack, StringBuffer text,

			String error) throws SAXException {

		NodeRecord node = null;

		// get node from current stack

		try {

			node = nodeStack.pop();

		} catch (EmptyStackException ee) {

			logMetacat.error("Node stack is empty for text data in startElement");

			throw new SAXException(error);

		}

		logMetacat.debug("current node type from xml is TEXT in start element");

		logMetacat.debug("node type from stack: " + node.getNodeType());

		logMetacat.debug("current node data from xml is: " + text.toString());

		logMetacat.debug("node data from stack: " + node.getNodeData());

		logMetacat.debug("node name from stack: " + node.getNodeName());

		logMetacat.debug("node is: " + node.getNodeId());

		if (!node.getNodeType().equals("TEXT")

				|| !(text.toString()).equals(node.getNodeData())) {

			logMetacat.error("Inconsistence happened: ");

			logMetacat.error("current node type from xml is TEXT in start element");

			logMetacat.error("node type from stack: " + node.getNodeType());

			logMetacat.error("current node data from xml is: " + text.toString());

			logMetacat.error("node data from stack: " + node.getNodeData());

			logMetacat.error("node name from stack: " + node.getNodeName());

			logMetacat.error("node is: " + node.getNodeId());

			throw new SAXException(error);

		}// if

	}

	/** SAX Handler that is called for each XML text node */

	public void characters(char[] cbuf, int start, int len) throws SAXException {

		logMetacat.debug("CHARACTERS");

		if (!handleInlineData) {

			// buffer all text nodes for same element. This is for if text was

			// split into different nodes

			textBuffer.append(new String(cbuf, start, len));

			// set hittextnode true

			hitTextNode = true;

			// if text buffer .size is greater than max, write it to db.

			// so we can save memory

			if (textBuffer.length() > MAXDATACHARS) {

				logMetacat.debug("Write text into DB in charaters"

						+ " when text buffer size is greater than maxmum number");

				DBSAXNode currentNode = (DBSAXNode) nodeStack.peek();

				endNodeId = writeTextForDBSAXNode(endNodeId, textBuffer, currentNode);

				textBuffer = null;

				textBuffer = new StringBuffer();

			}

		} else {

			// this is inline data and write file system directly

			// we don't need to buffer it.

			StringBuffer inlineText = new StringBuffer();

			inlineText.append(new String(cbuf, start, len));

			logMetacat.debug("The inline text data write into file system: "

					+ inlineText.toString());

			writeInlineDataIntoFile(inlineDataFileWriter, inlineText);

		}

	}

	/** SAX Handler that is called at the end of each XML element */

	public void endElement(String uri, String localName, String qName)

			throws SAXException {

		logMetacat.debug("End ELEMENT " + qName);

		if (localName.equals(INLINE) && handleInlineData) {

			// Get the node from the stack

			DBSAXNode currentNode = (DBSAXNode) nodeStack.pop();

			logMetacat.debug("End of inline data");

			// close file writer

			try {

				inlineDataFileWriter.close();

				handleInlineData = false;

			} catch (IOException ioe) {

				throw new SAXException(ioe.getMessage());