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: Jing Tao

 *

 *   '$Author: daigle $'

 *     '$Date: 2010-02-03 13:29:30 -0800 (Wed, 03 Feb 2010) $'

 * '$Revision: 5208 $'

 *

 * 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.AuthUtil;

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 bethods for the SAX parser to

 * call when processing the XML stream and generating events

 * Here is the rules for user which has write permission in top level access

 * rules(he can update metadata but can't update access module) try to update

 * a document:

 * 1. Checking access part (both in top level and additional level, node by node)

 *    If something was modified, reject the document. Note: for additional part

 *    The access subtree startnode starts at "<describ>" rather than <access>.

 *    This is because make sure ids wouldn't be mess up by user.

 * 2. Checking ids in additional access part, if they points a distribution

 *    module with online or inline. If some ids don't, reject the documents.

 * 3. For inline distribution:

 *    If user can't read the inline data, the empty string in inline element

 *    will be send to user if he read this eml document(user has read access

 *    at top level - metadata, but user couldn't read inline data).

 *    Here is some intrested senario: If user does have read and write

 *    permission in top level access rules(for metadata)

 *    but 1) user doesn't have read and write permission in inline data block,

 *    so if user update the document with some inline data rather than

 *    empty string in same inline data block(same distribution inline id),

 *    this means user updated the inline data. So the document should be

 *    rejected.

 *    2) user doesn't have read permssion, but has write premission in

 *    inline data block. If user send back inline data block with empty

 *    string, we will think user doesn't update inline data and we will

 *    copy old inline data back to the new one. If user send something

 *    else, we will overwrite the old inline data by new one.

 * 4. For online distribution:

 *    It is easy than inline distribution. When the user try to change a external

 *    document id, we will checking if the user have "all" permission to it.

 *    If don't have, reject the document. If have, delete the old rules and apply

 *    The new rules.

 *

 *

 * Here is some info about additional access rules ( data object rules):

 *  The data access rules format looks like:

 *  <additionalMetadata>

 *    <describes>100</describes>

 *    <describes>300</describes>

 *    <access>rulesone</access>

 *  </additionalMetadata>

 *  <additionalMetadata>

 *    <describes>200</describes>

 *    <access>rulesthree</access>

 *  </additionalMetadata>

 *  Because eml additionalMetadata is (describes+, any) and any ocurrence is 1.

 *  So following xml will be rejected by xerces.

 *  <additionalMetadata>

 *    <describes>100</describes>

 *    <describes>300</describes>

 *    <other>....</other>

 *    <access>rulesone</access>

 *  </additionalMetadata>

 */

public class Eml200SAXHandler extends DBSAXHandler implements

        AccessControlInterface

{

    private boolean processTopLeverAccess = false;

    // now additionalAccess will be explained as distribution access control

    // - data file

    private boolean processAdditionalAccess = false;

    private boolean processOtherAccess = false;

    private AccessSection accessObject = null;

    private AccessRule accessRule = null;

    private Vector describesId = new Vector(); // store the ids in

                                               //additionalmetadata/describes

    //store all distribution element id for online url. key is the distribution

    // id and  data  is url

    private Hashtable onlineURLDistributionIdList = new Hashtable();

    // distribution/oneline/url will store this vector if distribution doesn't

    // have a id.

    private Vector onlineURLDistributionListWithoutId = new Vector();

    //store all distribution element id for online other distribution, such as

    // connection or connectiondefination. key is the distribution id

    // and  data  is distribution id

    private Hashtable onlineOtherDistributionIdList = new Hashtable();

    //store all distribution element id for inline data.

    // key is the distribution id, data is the internal inline id

    private Hashtable inlineDistributionIdList = new Hashtable();

    //store all distribution element id for off line data.

    // key is the distribution id, data is the id too.

    private Hashtable offlineDistributionIdList = new Hashtable();

    // a hash to stored all distribution id, both key and value are id itself

    private Hashtable distributionAllIdList = new Hashtable();

    // temporarily store distribution id

    private String distributionId = null;

    // flag to indicate to handle distrubiton

    private boolean proccessDistribution = false;

    // a hash table to stored the distribution which is a reference and this

    // distribution has a id too. The key is itself id of this distribution,

    // the value is the referenced id.

    // So, we only stored the format like this:

    // <distribution id ="100"><reference>300</reference></distribution>

    // the reason is:

    // if not id in distribution, then the distribution couldn't be added

    // to additional access module. The distribution block which was referenced

    // id (300) will be stored in above distribution lists.

    private Hashtable distributionReferenceList = new Hashtable();

    private boolean needCheckingAccessModule = false;

    private AccessSection unChangebleTopAccessSubTree = null;

    private Vector unChangebleAdditionalAccessSubTreeVector = new Vector();

    private Hashtable unChangebleReferencedAccessSubTreeHash = new Hashtable();

    private AccessSection topAccessSection;

    private Vector additionalAccessVector = new Vector();

    // key is subtree id and value is accessSection object

    private Hashtable possibleReferencedAccessHash = new Hashtable();

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

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

    private Stack storedAccessNodeStack = new Stack();

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

    private Vector onlineDataFileIdInRelationVector = new Vector();

    // Indicator of inline data

    private boolean handleInlineData = false;

    private Hashtable inlineDataNameSpace = null;

    private FileWriter inlineDataFileWriter = null;

    private String inlineDataFileName = null;

    private int inLineDataIndex = 0;

    private Vector inlineFileIDList = new Vector();

    private boolean inadditionalMetaData = false;

    //user has unwritable inline data object when it updates a document

    private boolean unWritableInlineDataObject = false;

    //user has unreadable inline data when it updates a dcoument

    private boolean unReadableInlineDataObject = false;

    // the hashtable contains the info from xml_access table which

    // inline data the user can't read when user update a document.

    // The key in hashtable is subtree id and data is the inline data internal

    // file name.

    private Hashtable previousUnreadableInlineDataObjectHash = new Hashtable();

    // the hashtable contains the info from xml_access table which

    // inline data the user can't write when user update a document.

    // The key in hashtable is subtree id and data is the inline data internal

    // file name.

    private Hashtable previousUnwritableInlineDataObjectHash = new Hashtable();

    private Hashtable accessSubTreeAlreadyWriteDBList = new Hashtable();

    //This hashtable will stored the id which already has additional access

    // control. So the top level access control will ignore them.

    private Hashtable onlineURLIdHasadditionalAccess   = new Hashtable();

    // additional access module will be in additionalMetadata part. Its format

    // look like

    //<additionalMetadata>

    //   <describes>100</describes>

    //   <describes>300</describes>

    //   <access>rulesone</access>

    //</additionalMetadata>

    // If user couldn't have all permission to update access rules, he/she could

    // not update access module, but also couldn't update "describes".

    // So the start node id for additional access section is the first describes

    // element

    private boolean firstDescribesInAdditionalMetadata = true;

    private long    firstDescribesNodeId               = -1;

    private int numberOfHitUnWritableInlineData = 0;

    // Constant

    private static final String EML = "eml";

    private static final String DESCRIBES = "describes";

    private static final String ADDITIONALMETADATA = "additionalMetadata";

    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 OFFLINE = "offline";

    private static final String CONNECTION = "connection";

    private static final String CONNECTIONDEFINITION = "connectionDefinition";

    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!";

    public static final String TOPLEVEL = "top";

    public static final String DATAACCESSLEVEL = "dataAccess";

    // this level is for the access module which is not in top or additional

    // place, but it was referenced by top or additional

    private static final String REFERENCEDLEVEL = "referenced";

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

    private static final String DISTRIBUTION = "distribution";

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

    /**

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

     * specify the version need to upadate

     *

     * @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 Eml200SAXHandler(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 unchangable 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 update access subtree

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

                    && !control.hasPermission(user, groups,

                            AccessControlInterface.ALLSTRING) 

                            && !AuthUtil.isAdministrator(user, groups))

            {

                needCheckingAccessModule = true;

                unChangebleTopAccessSubTree = getTopAccessSubTreeFromDB();

                unChangebleAdditionalAccessSubTreeVector =

                                         getAdditionalAccessSubTreeListFromDB();

                unChangebleReferencedAccessSubTreeHash =

                                         getReferencedAccessSubTreeListFromDB();

            }

            //Here is for  data object checking.

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

            {

              //info about inline data object which user doesn't have read

              //permission the info come from xml_access table

              previousUnreadableInlineDataObjectHash = PermissionController.

                            getUnReadableInlineDataIdList(docid, user,

                                                          groups, true);

              //info about data object which user doesn't have write permission

              // the info come from xml_accesss table

              previousUnwritableInlineDataObjectHash = PermissionController.

                            getUnWritableInlineDataIdList(docid, user,

                                                          groups, true);

            }

        }

        catch (Exception e)

        {

            logMetacat.error("error in Eml200SAXHandler is " + e.getMessage());

            throw new SAXException(e.getMessage());

        }

    }

    /*

     * Get the top level access subtree info from xml_accesssubtree table.

     * If no top access subtree found, null will be return.

     */

     private AccessSection getTopAccessSubTreeFromDB()

                                                       throws SAXException

     {

       AccessSection topAccess = null;

       PreparedStatement pstmt = null;

       ResultSet rs = null;

       String sql = "SELECT subtreeid, startnodeid, endnodeid "

                + "FROM xml_accesssubtree WHERE docid like ? "

                + "AND controllevel like ?";

       try

       {

            pstmt = connection.prepareStatement(sql);

            // Increase DBConnection usage count

            connection.increaseUsageCount(1);

            // Bind the values to the query

            pstmt.setString(1, docid);

            pstmt.setString(2, TOPLEVEL);

            logMetacat.debug("Eml200SAXHandler.getTopAccessSubTreeFromDB - executing SQL: " + pstmt.toString());

            pstmt.execute();

            // Get result set

            rs = pstmt.getResultSet();

            if (rs.next())

            {

                String sectionId = rs.getString(1);

                long startNodeId = rs.getLong(2);

                long endNodeId = rs.getLong(3);

                // create a new access section

                topAccess = new AccessSection();

                topAccess.setControlLevel(TOPLEVEL);

                topAccess.setDocId(docid);

                topAccess.setSubTreeId(sectionId);

                topAccess.setStartNodeId(startNodeId);

                topAccess.setEndNodeId(endNodeId);

            }

            pstmt.close();

        }//try

        catch (SQLException e)

        {

            throw new SAXException(

                    "EMLSAXHandler.getTopAccessSubTreeFromDB(): "

                            + e.getMessage());

        }//catch

        finally

        {

            try

            {

                pstmt.close();

            }

            catch (SQLException ee)

            {

                throw new SAXException(

                        "EMLSAXHandler.getTopAccessSubTreeFromDB(): "

                                + ee.getMessage());

            }

        }//finally

        return topAccess;

     }

    /*

     * Get the subtree node info from xml_accesssubtree table

     */

    private Vector getAdditionalAccessSubTreeListFromDB() throws Exception

    {

        Vector result = new Vector();

        PreparedStatement pstmt = null;

        ResultSet rs = null;

        String sql = "SELECT subtreeid, startnodeid, endnodeid "

                + "FROM xml_accesssubtree WHERE docid like ? "

                + "AND controllevel 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.setString(2, DATAACCESSLEVEL);

            pstmt.execute();

            // Get result set

            rs = pstmt.getResultSet();

            while (rs.next())

            {

                String sectionId = rs.getString(1);

                long startNodeId = rs.getLong(2);

                long endNodeId = rs.getLong(3);

                // create a new access section

                AccessSection accessObj = new AccessSection();

                accessObj.setControlLevel(DATAACCESSLEVEL);

                accessObj.setDocId(docid);

                accessObj.setSubTreeId(sectionId);

                accessObj.setStartNodeId(startNodeId);

                accessObj.setEndNodeId(endNodeId);

                // add this access obj into vector

                result.add(accessObj);

            }

            pstmt.close();

        }//try

        catch (SQLException e)

        {

            throw new SAXException(

                    "EMLSAXHandler.getadditionalAccessSubTreeListFromDB(): "

                            + e.getMessage());

        }//catch

        finally

        {

            try

            {

                pstmt.close();

            }

            catch (SQLException ee)

            {

                throw new SAXException(

                        "EMLSAXHandler.getAccessSubTreeListFromDB(): "

                                + ee.getMessage());

            }

        }//finally

        return result;

    }

   /*

    * Get the access subtree for referenced info from xml_accesssubtree table

    */

   private Hashtable getReferencedAccessSubTreeListFromDB() throws Exception

   {

       Hashtable result = new Hashtable();

       PreparedStatement pstmt = null;

       ResultSet rs = null;

       String sql = "SELECT subtreeid, startnodeid, endnodeid "

               + "FROM xml_accesssubtree WHERE docid like ? "

               + "AND controllevel 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.setString(2, REFERENCEDLEVEL);

           pstmt.execute();

           // Get result set

           rs = pstmt.getResultSet();

           while (rs.next())

           {

               String sectionId = rs.getString(1);

               long startNodeId = rs.getLong(2);

               long endNodeId = rs.getLong(3);

               // create a new access section

               AccessSection accessObj = new AccessSection();

               accessObj.setControlLevel(DATAACCESSLEVEL);

               accessObj.setDocId(docid);

               accessObj.setSubTreeId(sectionId);

               accessObj.setStartNodeId(startNodeId);

               accessObj.setEndNodeId(endNodeId);

               // add this access obj into hastable

               if ( sectionId != null && !sectionId.trim().equals(""))

               {

                 result.put(sectionId, accessObj);

               }

           }

           pstmt.close();

       }//try

       catch (SQLException e)

       {

           throw new SAXException(

                   "EMLSAXHandler.getReferencedAccessSubTreeListFromDB(): "

                           + e.getMessage());

       }//catch

       finally

       {

           try

           {

               pstmt.close();

           }

           catch (SQLException ee)

           {

               throw new SAXException(

                       "EMLSAXHandler.getReferencedSubTreeListFromDB(): "

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

        // none inline part

        //+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

        //+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

        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 (qName.equals(INLINE) && !inadditionalMetaData)

            {

                handleInlineData = true;

                inLineDataIndex++;

                //intitialize namespace hash for in line data

                inlineDataNameSpace = new Hashtable();

                //initialize file writer

                String docidWithoutRev = DocumentUtil.getDocIdFromString(docid);

                String seperator = ".";

                try {

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

				} catch (PropertyNotFoundException pnfe) {

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

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

				}

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

                inlineDataFileName = docidWithoutRev + seperator + revision

                        + seperator + inLineDataIndex;

                inlineDataFileWriter = createInlineDataFileWriter(inlineDataFileName);

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

                // metacat will

                // delete the inline data file

                inlineFileIDList.add(inlineDataFileName);

                // put distribution id and inline file id into a  hash

                if (distributionId != null)

                {

                  //check to see if this inline data is readable or writable to

                  // this user

                  if (!previousUnreadableInlineDataObjectHash.isEmpty() &&

                       previousUnreadableInlineDataObjectHash.containsKey(distributionId))

                  {

                      unReadableInlineDataObject = true;

                  }

                  if (!previousUnwritableInlineDataObjectHash.isEmpty() &&

                       previousUnwritableInlineDataObjectHash.containsKey(distributionId))

                  {

                     unWritableInlineDataObject = true;

                     numberOfHitUnWritableInlineData++;

                  }

                  // store the distributid and inlinedata filename into a hash

                  inlineDistributionIdList.put(distributionId, inlineDataFileName);

                }

            }

            //==============================================================

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

            // node

            // This will happend if the element is mixted

            //==============================================================

            if (hitTextNode && parentNode != null)

            {

                if (needCheckingAccessModule

                        && (processAdditionalAccess || processOtherAccess || processTopLeverAccess)) {

                    // 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.info("DOCNAME-a: " + docname);

                    logMetacat.info("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.info("DOCTYPE-b: " + doctype);

                }

                rootNode.writeNodename(docname);

                //System.out.println("here!!!!!!!!!!!!!!!!!!1");

                try {

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

                    // Catalog

                    // Because this is select statement and it needn't to roll

                    // back if

                    // insert document action fialed.

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

                        //System.out.println("here!!!!!!!!!!!!!!!!!!2");

                    }//try

                    finally {

                        // Return dbconnection

                        DBConnectionPool.returnDBConnection(dbConn,

                                serialNumber);

                    }//finally

                    //create documentImpl object by the constructor which can

                    // specify

                    //the revision

                    if (!super.getIsRevisionDoc())

                    {

                       logMetacat.debug("EML200SaxHandler.startElement - creating new DocumentImple for " + docid);

                       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("EML200SaxHandler.startElement - error with action " + 

                    		action + " : " + ane.getMessage()));

                }

            }

            //==================================================================

            // node

            //==================================================================

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

            long startNodeId = currentNode.getNodeID();

            // Add all of the namespaces

            String prefix = null;

            String nsuri = null;

            Enumeration prefixes = namespaces.keys();

            while (prefixes.hasMoreElements())

            {

                prefix = (String) prefixes.nextElement();

                nsuri = (String) namespaces.get(prefix);

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

            }

            //=================================================================

           // attributes

           // 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 regeistered.

              if (attributeName != null

                      && attributeName

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

                  SchemaLocationResolver resolver = new SchemaLocationResolver(

                          attributeValue);

                  resolver.resolveNameSpace();

              }

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

                       currentNode.getTagName().equals(DISTRIBUTION) &&

                       !inadditionalMetaData)

              {

                 // this is a distribution element and the id is distributionID

                 distributionId = attributeValue;

                 distributionAllIdList.put(distributionId, distributionId);

              }

          }//for

           //=================================================================

            // handle access stuff

            //==================================================================

            if (localName.equals(ACCESS))

            {

                //make sure the access is top level

                // this mean current node's parent's parent should be "eml"

                DBSAXNode tmpNode = (DBSAXNode) nodeStack.pop();// pop out

                                                                    // parent

                                                                    // node

                //peek out grandParentNode

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

                // put parent node back

                nodeStack.push(tmpNode);

                String grandParentTag = grandParentNode.getTagName();

                if (grandParentTag.equals(EML) && !inadditionalMetaData)

                {

                  processTopLeverAccess = true;

                }

                else if ( !inadditionalMetaData )

                {

                  // process other access embedded into resource level

                  // module

                  processOtherAccess = true;

                }

                else

                {

                  // this for access in additional data which don't have

                  // described element. If it has a descirbed element,

                  // this code would hurt any thing

                  processAdditionalAccess = true;

                  logMetacat.info("accessing process additional access true when meet access");

                }

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

                // for additional access subtree, the  start of node id should

                // be describe element. We also stored the start access element

                // node id too.

                if (processAdditionalAccess)

                {

                  accessObject.setStartedDescribesNodeId(firstDescribesNodeId);

                  accessObject.setControlLevel(DATAACCESSLEVEL);

                }

                else if (processTopLeverAccess)

                {

                  accessObject.setControlLevel(TOPLEVEL);

                }

                else if (processOtherAccess)

                {

                  accessObject.setControlLevel(REFERENCEDLEVEL);

                }

                accessObject.setStartNodeId(startNodeId);

                accessObject.setDocId(docid);

            }

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

            else if (parentNode.getTagName() != null

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

                    && localName.equals(DENY))

           {

                accessRule = new AccessRule();

                //set permission type "allow"

                accessRule.setPermissionType(DENY);

            }

            //=================================================================

            // some other independ stuff

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

            // store access module element and attributes into stored stack

            if (needCheckingAccessModule

                    && (processAdditionalAccess || processOtherAccess || processTopLeverAccess))

            {

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

                }

            }

            if (currentNode.getTagName().equals(ADDITIONALMETADATA))

            {

              inadditionalMetaData = true;

            }

            else if (currentNode.getTagName().equals(DESCRIBES) &&

                     parentNode.getTagName().equals(ADDITIONALMETADATA) &&

                     firstDescribesInAdditionalMetadata)

            {

              // this is first decirbes element in additional metadata

              firstDescribesNodeId = startNodeId;

              // we started process additional access rules here

              // because access and describe couldn't be seperated

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

                        localName, prefix, MetacatUtil.normalize(null));

              storedAccessNodeStack.push(nodeElement);

              processAdditionalAccess = true;

              logMetacat.info("set processAdditonalAccess true when meet describe");

            }

            else if (inadditionalMetaData && processAdditionalAccess &&

                     parentNode.getTagName().equals(ADDITIONALMETADATA) &&

                     !currentNode.getTagName().equals(DESCRIBES) &&

                     !currentNode.getTagName().equals(ACCESS))

            {

               // we start processadditionalAccess  module when first hit describes

               // in additionalMetadata. So this is possible, there are

               // "describes" but not "access". So here is try to terminate

               // processAddionalAccess. In this situation, there another element

               // rather than "describes" or "access" as a child of additionalMetadata

               // so this is impossible it will have access element now.

               // If additionalMetadata has access element, the flag will be

               // terminated in endElement

               processAdditionalAccess = false;

               logMetacat.warn("set processAddtionAccess false if the there is no access in additional");

            }

            else if (currentNode.getTagName().equals(DISTRIBUTION) &&

                     !inadditionalMetaData)

            {

              proccessDistribution = true;

            }

             //==================================================================

            // reset name space