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: 2008-10-21 15:29:13 -0700 (Tue, 21 Oct 2008) $'

 * '$Revision: 4472 $'

 *

 * 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.service.PropertyService;

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;

    // 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;

	// 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 (action.equals("UPDATE")

                    && !control.hasPermission(user, groups,

                            AccessControlInterface.ALLSTRING)) {

                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 = MetaCatUtil.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);

            }

            // 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.setInlineDataFileName(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++;    

              // 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());

            }

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

        	// control which is not yet implemented in EML 2.1.0