Metacat

/**

 * Copyright 2006 OCLC Online Computer Library Center Licensed under the Apache

 * License, Version 2.0 (the "License"); you may not use this file except in

 * compliance with the License. You may obtain a copy of the License at

 * http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or

 * agreed to in writing, software distributed under the License is distributed on

 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express

 * or implied. See the License for the specific language governing permissions and

 * limitations under the License.

 */

package edu.ucsb.nceas.metacat.oaipmh.provider.server.catalog;

import java.io.File;

import java.io.IOException;

import java.io.InputStreamReader;

import java.io.Reader;

import java.sql.Connection;

import java.sql.DriverManager;

import java.sql.ResultSet;

import java.sql.SQLException;

import java.sql.SQLWarning;

import java.sql.Statement;

import java.text.SimpleDateFormat;

import java.util.ArrayList;

import java.util.Date;

import java.util.HashMap;

import java.util.Iterator;

import java.util.Map;

import java.util.NoSuchElementException;

import java.util.Properties;

import java.util.StringTokenizer;

import java.util.Vector;

import org.apache.log4j.Logger;

import edu.ucsb.nceas.metacat.client.DocumentNotFoundException;

import edu.ucsb.nceas.metacat.client.InsufficientKarmaException;

import edu.ucsb.nceas.metacat.client.Metacat;

import edu.ucsb.nceas.metacat.client.MetacatException;

import edu.ucsb.nceas.metacat.client.MetacatFactory;

import edu.ucsb.nceas.metacat.client.MetacatInaccessibleException;

import edu.ucsb.nceas.metacat.oaipmh.provider.server.OAIHandler;

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

import edu.ucsb.nceas.utilities.IOUtil;

import edu.ucsb.nceas.utilities.PropertyNotFoundException;

import ORG.oclc.oai.server.catalog.AbstractCatalog;

import ORG.oclc.oai.server.catalog.RecordFactory;

import ORG.oclc.oai.server.verb.BadResumptionTokenException;

import ORG.oclc.oai.server.verb.CannotDisseminateFormatException;

import ORG.oclc.oai.server.verb.IdDoesNotExistException;

import ORG.oclc.oai.server.verb.NoItemsMatchException;

import ORG.oclc.oai.server.verb.NoMetadataFormatsException;

import ORG.oclc.oai.server.verb.NoSetHierarchyException;

import ORG.oclc.oai.server.verb.OAIInternalServerError;

/**

 * MetacatCatalog is an implementation of AbstractCatalog interface.

 * 

 * @author Ralph LeVan, OCLC Online Computer Library Center

 */

public class MetacatCatalog extends AbstractCatalog {

  /* Class fields */

  private static final Logger logger = Logger.getLogger(MetacatCatalog.class);

  private static String refreshDate = null;

  /** Database connection */

  private static String metacatDBDriver;

  private static String metacatDBURL;

  private static String metacatDBUser;

  private static String metacatDBPassword;

  private static String metacatURL;

  /* Instance fields */

  protected String homeDir;

  private HashMap<String, String> dateMap = new HashMap<String, String>();

  private HashMap<String, String> filteredDateMap = null;

  private HashMap<String, String> docTypeMap = new HashMap<String, String>();

  private HashMap resumptionResults = new HashMap();

  private int maxListSize;

  /*

   * QUERY string to find all eml-2.x.y documents in the Metacat database

   * that are publicly accessible

   */

  private final String QUERY =

  "SELECT xd.docid, xd.doctype, xd.date_updated " +

  "FROM xml_documents xd, identifier id " +

  "WHERE xd.doctype like 'eml://ecoinformatics.org/eml-2%' " +

  " AND xd.docid = id.docid " +

  " AND xd.rev = id.rev " +

  // ALLOW rule

  " AND id.guid IN " +

  "     (SELECT guid " +

  "     FROM xml_access " +

  "		AND lower(principal_name) = 'public' " +

  " 	AND perm_type = 'allow' " +

  " 	AND permission > 3" +

  "		) " +

  // DENY rules?

  " AND id.guid NOT IN " +

  "     (SELECT guid " +

  "     FROM xml_access " +

  "     WHERE lower(principal_name) = 'public' " +

  "		AND perm_type = 'deny' " +

  "		AND perm_order ='allowFirst' " +

  "		AND permission > 3 " +

  "     ) ";

/* Constructors */

  public MetacatCatalog(Properties properties) {

    String errorStr;

    String temp;

    temp = properties.getProperty("oaipmh.maxListSize");

    if (temp == null) {

      errorStr = "oaipmh.maxListSize is missing from the properties file";

      throw new IllegalArgumentException(errorStr);

    }

    maxListSize = Integer.parseInt(temp);

    metacatDBDriver = properties.getProperty("database.driver");

    metacatDBURL = properties.getProperty("database.connectionURI");

    metacatDBUser = properties.getProperty("database.user");

    metacatDBPassword = properties.getProperty("database.password");

    try {

      if (OAIHandler.isIntegratedWithMetacat()) {

        metacatURL = SystemUtil.getServletURL();

      }

      else {

        metacatURL = properties.getProperty("test.metacatUrl");

      }

      logger.warn("metacatURL: " + metacatURL);

    }

    catch (PropertyNotFoundException e) {

      logger.error("PropertyNotFoundException: " + 

             "unable to determine metacat URL from SystemUtil.getServletURL()");

    }

    loadCatalog();

  }

  /* Class methods */

  /**

   * Use the current date as the basis for the resumptiontoken

   * 

   * @return a long integer version of the current time

   */

  private synchronized static String getRSName() {

    Date now = new Date();

    return Long.toString(now.getTime());

  }

  /* Instance methods */

  /**

   * close the repository

   */

  public void close() {

  }

  /**

   * Utility method to construct a Record object for a specified metadataFormat

   * from a native record

   * 

   * @param nativeItem

   *          native item from the dataase

   * @param metadataPrefix

   *          the desired metadataPrefix for performing the crosswalk

   * @return the <record/> String

   * @exception CannotDisseminateFormatException

   *              the record is not available for the specified metadataPrefix.

   */

  private String constructRecord(HashMap nativeItem, String metadataPrefix)

      throws CannotDisseminateFormatException {

    String schemaURL = null;

    Iterator setSpecs = getSetSpecs(nativeItem);

    Iterator abouts = getAbouts(nativeItem);

    if (metadataPrefix != null) {

      if ((schemaURL = getCrosswalks().getSchemaURL(metadataPrefix)) == null)

        throw new CannotDisseminateFormatException(metadataPrefix);

    }

    RecordFactory recordFactory = getRecordFactory();

    String recordString = recordFactory.create(nativeItem, schemaURL, 

                                              metadataPrefix, setSpecs, abouts);

    return recordString;

  }

  /**

   * Using the original dateMap catalog, produce a filtered dateMap catalog

   * consisting of only those entries that match the 'from', 'until', and

   * 'metadataPrefix' criteria.

   * 

   * @param from                 the from date, e.g. "2008-06-01"

   * @param until                the until date, e.g. "2009-01-01"

   * @param metadataPrefix       the metadataPrefix value, e.g. "oai_dc"

   * 

   * @return   aDateMap, a HashMap containing only the matched entries.

   */

  private HashMap<String, String> filterDateMap(String from, String until,

      String metadataPrefix) {

    if (shouldRefreshCatalog()) {

      loadCatalog();

    }

    HashMap<String, String> aDateMap = new HashMap<String, String>();

    Iterator iterator = dateMap.entrySet().iterator();

    while (iterator.hasNext()) {

      Map.Entry entryDateMap = (Map.Entry) iterator.next();

      String dateUpdated = (String) entryDateMap.getValue();

      /*

       * First filter catalog entries based on whether their date updated falls

       * within the 'from' and 'until' parameters.

       */

      if (dateUpdated.compareTo(from) >= 0 && dateUpdated.compareTo(until) <= 0) 

      {

        String docid = (String) entryDateMap.getKey();

        HashMap<String, String> nativeHeader = getNativeHeader(docid);

        String doctype = nativeHeader.get("doctype");

        /*

         * Next filter catalog entries based on Metacat doctype as compared to

         * OAI-PMH metadataPrefix.

         */

        if (isIncludedDoctype(doctype, metadataPrefix)) {

          aDateMap.put(docid, dateUpdated);

        }

      }

    }

    return aDateMap;

  }

  /**

   * get an Iterator containing the abouts for the nativeItem

   * 

   * @param rs

   *          ResultSet containing the nativeItem

   * @return an Iterator containing the list of about values for this nativeItem

   */

  private Iterator getAbouts(HashMap nativeItem) {

    return null;

  }

  /**

   * Returns a connection to the database. Opens the connection if a connection

   * has not already been made previously.

   * 

   * @return  conn  the database Connection object

   */

  public Connection getConnection() {

    Connection conn = null;

    try {

      Class.forName(metacatDBDriver);

    }

    catch (ClassNotFoundException e) {

      logger.error("Can't load driver " + e);

      return conn;

    } 

    // Make the database connection

    try {

      conn = DriverManager.getConnection(metacatDBURL, metacatDBUser, 

                                           metacatDBPassword);

      // If a SQLWarning object is available, print its warning(s).

      // There may be multiple warnings chained.

      SQLWarning warn = conn.getWarnings();

      if (warn != null) {

        while (warn != null) {

          logger.warn("SQLState: " + warn.getSQLState());

          logger.warn("Message:  " + warn.getMessage());

          logger.warn("Vendor: " + warn.getErrorCode());

          warn = warn.getNextWarning();

        }

      }

    }

    catch (SQLException e) {

      logger.error("Database access failed " + e);

    }

    return conn;

  }

  /**

   * Get the most recent date that the xml_documents table was updated

   * @return

   */

  public String getMaxDateUpdated() {

    String maxDateUpdated = null;

    String query = 

              "SELECT MAX(date_updated) AS max_date_updated FROM xml_documents";

    Statement stmt;

    try {

      Connection conn = getConnection();    

      if (conn != null) {

        stmt = conn.createStatement();                          

        ResultSet rs = stmt.executeQuery(query);

        while (rs.next()) {

          maxDateUpdated = rs.getDate("max_date_updated").toString();

        }

        stmt.close();   

        conn.close();

      }

    }

    catch(SQLException e) {

      logger.error("SQLException: " + e.getMessage());

    }

    return maxDateUpdated;

  }

  /**

   * Get a document from Metacat.

   * 

   * @param docid  the docid of the document to read

   * 

   * @return recordMap       a HashMap holding the document contents

   * 

   * @throws IOException

   */

  private HashMap<String, String> getMetacatDocument(String docid) 

      throws IOException {

    HashMap<String, String> recordMap = getNativeHeader(docid);

    if (recordMap == null) {

      return null;

    } 

    else {

      try {

        /* Perform a Metacat read operation on this docid */

        Metacat metacat = MetacatFactory.createMetacatConnection(metacatURL);

        Reader reader = new InputStreamReader(metacat.read(docid));

        StringBuffer stringBuffer = IOUtil.getAsStringBuffer(reader, true);

        String emlString = stringBuffer.toString();

        recordMap.put("recordBytes", emlString);

      }

      catch (MetacatInaccessibleException e) {

        logger.error("MetacatInaccessibleException:\n" + e.getMessage());

      }

      catch (MetacatException e) {

        logger.error("MetacatException:\n" + e.getMessage());

      }

      catch (DocumentNotFoundException e) {

        logger.error("DocumentNotFoundException:\n" + e.getMessage());

      }

      catch (InsufficientKarmaException e) {

        logger.error("InsufficientKarmaException:\n" + e.getMessage());

      }

      catch (IOException e) {

        logger.error("Error reading EML document from metacat:\n" + 

                     e.getMessage()

                    );

      }

    }

    return recordMap;

  }

  private HashMap<String, String> getNativeHeader(String localIdentifier) {

    HashMap<String, String> recordMap = null;

    if (dateMap.containsKey(localIdentifier)) {

      recordMap = new HashMap<String, String>();

      recordMap.put("localIdentifier", localIdentifier);

      recordMap.put("lastModified", dateMap.get(localIdentifier));

      recordMap.put("doctype", docTypeMap.get(localIdentifier));

      return recordMap;

    }

    return recordMap;

  }

  /**

   * Retrieve the specified metadata for the specified oaiIdentifier

   * 

   * @param oaiIdentifier

   *          the OAI identifier

   * @param metadataPrefix

   *          the OAI metadataPrefix

   * @return the Record object containing the result.

   * @exception CannotDisseminateFormatException

   *              signals an http status code 400 problem

   * @exception IdDoesNotExistException

   *              signals an http status code 404 problem

   * @exception OAIInternalServerError

   *              signals an http status code 500 problem

   */

  public String getRecord(String oaiIdentifier, String metadataPrefix)

      throws IdDoesNotExistException, 

             CannotDisseminateFormatException,

             OAIInternalServerError 

  {

    HashMap<String, String> nativeItem = null;

    try {

      RecordFactory recordFactory = getRecordFactory();

      String localIdentifier = recordFactory.fromOAIIdentifier(oaiIdentifier);

      nativeItem = getMetacatDocument(localIdentifier);

      if (nativeItem == null) throw new IdDoesNotExistException(oaiIdentifier);

      return constructRecord(nativeItem, metadataPrefix);

    } 

    catch (IOException e) {

      e.printStackTrace();

      throw new OAIInternalServerError("Database Failure");

    }

  }

  /**

   * Retrieve a list of schemaLocation values associated with the specified

   * oaiIdentifier.

   * 

   * We get passed the ID for a record and are supposed to return a list of the

   * formats that we can deliver the record in. Since we are assuming that all

   * the records in the directory have the same format, the response to this is

   * static;

   * 

   * @param oaiIdentifier       the OAI identifier

   * 

   * @return a Vector containing schemaLocation Strings

   * 

   * @exception OAIBadRequestException

   *              signals an http status code 400 problem

   * @exception OAINotFoundException

   *              signals an http status code 404 problem

   * @exception OAIInternalServerError

   *              signals an http status code 500 problem

   */

  public Vector getSchemaLocations(String oaiIdentifier)

      throws IdDoesNotExistException, OAIInternalServerError,

      NoMetadataFormatsException {

    HashMap<String, String> nativeItem = null;

    try {

      String localIdentifier = getRecordFactory().fromOAIIdentifier(

          oaiIdentifier);

      nativeItem = getMetacatDocument(localIdentifier);

    } 

    catch (IOException e) {

      e.printStackTrace();

      throw new OAIInternalServerError("Database Failure");

    }

    if (nativeItem != null) {

      RecordFactory recordFactory = getRecordFactory();

      return recordFactory.getSchemaLocations(nativeItem);

    } 

    else {

      throw new IdDoesNotExistException(oaiIdentifier);

    }

  }

  /**

   * get an Iterator containing the setSpecs for the nativeItem

   * 

   * @param rs

   *          ResultSet containing the nativeItem

   * @return an Iterator containing the list of setSpec values for this

   *         nativeItem

   */

  private Iterator getSetSpecs(HashMap nativeItem) {

    return null;

  }

  /**

   * Should a document with the specified Metacat doctype be included in the

   * list of identifiers/records for the specified OAI-PMH metadataPrefix?

   * 

   * @param doctype              e.g. "eml://ecoinformatics.org/eml-2.1.0"

   * @param metadataPrefix       e.g. "oai_dc", "eml-2.0.1", "eml-2.1.0"

   * @return

   */

  private boolean isIncludedDoctype(String doctype, String metadataPrefix) {

    boolean isIncluded = false;

    /*

     * If the metadataPrefix is "oai_dc", then include all catalog entries

     * in the list of identifiers. Else if the metadataPrefix is an EML

     * document type, then only include those catalog entries whose

     * document type matches that of the metadataPrefix. 

     */

    if (doctype != null && 

        (metadataPrefix.equals("oai_dc") ||

         (doctype.startsWith("eml://ecoinformatics.org/eml-") && 

          doctype.endsWith(metadataPrefix)

         )

        )

       ) {

      isIncluded = true;

    }

    return isIncluded;

  }

  /**

   * Override this method if some files exist in the filesystem that aren't

   * metadata records.

   * 

   * @param child

   *          the File to be investigated

   * @return true if it contains metadata, false otherwise

   */

  protected boolean isMetadataFile(File child) {

    return true;

  }

  /**

   * Retrieve a list of Identifiers that satisfy the criteria parameters

   * 

   * @param from

   *          beginning date in the form of YYYY-MM-DD or null if earliest date

   *          is desired

   * @param until

   *          ending date in the form of YYYY-MM-DD or null if latest date is

   *          desired

   * @param set

   *          set name or null if no set is desired        

   * @param metadataPrefix       

   *          e.g. "oai_dc", "eml-2.0.1", "eml-2.1.0"

   *        

   * @return a Map object containing an optional "resumptionToken" key/value

   *         pair and an "identifiers" Map object. The "identifiers" Map

   *         contains OAI identifier keys with corresponding values of "true" or

   *         null depending on whether the identifier is deleted or not.

   * @exception OAIBadRequestException

   *              signals an http status code 400 problem

   */

  public Map listIdentifiers(String from, String until, String set,

                             String metadataPrefix) 

          throws NoItemsMatchException {

    purge(); // clean out old resumptionTokens

    Map<String, Object> listIdentifiersMap = new HashMap<String, Object>();

    ArrayList<String> headers = new ArrayList<String>();

    ArrayList<String> identifiers = new ArrayList<String>();

    filteredDateMap = filterDateMap(from, until, metadataPrefix);

    Iterator iterator = filteredDateMap.entrySet().iterator();

    int numRows = filteredDateMap.entrySet().size();

    int count = 0;

    RecordFactory recordFactory = getRecordFactory();

    while (count < maxListSize && iterator.hasNext()) {

      Map.Entry entryDateMap = (Map.Entry) iterator.next();

      String dateUpdated = (String) entryDateMap.getValue();

      String key = (String) entryDateMap.getKey();

      HashMap<String, String> nativeHeader = getNativeHeader(key);

      String[] headerArray = recordFactory.createHeader(nativeHeader);

     /* 

      * header, e.g.

      * 

      * <header>

      *   <identifier>urn:lsid:knb.ecoinformatics.org:knb-lter-gce:26</identifier>

      *   <datestamp>2009-03-11</datestamp>

      * </header>

      */

      String header = headerArray[0];

      headers.add(header);

      /*

       * identifier, e.g. urn:lsid:knb.ecoinformatics.org:knb-lter-gce:26

       */

      String identifier = headerArray[1]; 

      identifiers.add(identifier);

      count++;

    }

    if (count == 0) { throw new NoItemsMatchException(); }

    /* decide if you're done */

    if (iterator.hasNext()) {

      String resumptionId = getRSName();

      resumptionResults.put(resumptionId, iterator);

      /*****************************************************************

       * Construct the resumptionToken String however you see fit.

       *****************************************************************/

      StringBuffer resumptionTokenSb = new StringBuffer();

      resumptionTokenSb.append(resumptionId);

      resumptionTokenSb.append(":");

      resumptionTokenSb.append(Integer.toString(count));

      resumptionTokenSb.append(":");

      resumptionTokenSb.append(Integer.toString(numRows));

      resumptionTokenSb.append(":");

      resumptionTokenSb.append(metadataPrefix);

      /*****************************************************************

       * Use the following line if you wish to include the optional

       * resumptionToken attributes in the response. Otherwise, use the line

       * after it that I've commented out.

       *****************************************************************/

      listIdentifiersMap.put("resumptionMap", getResumptionMap(

          resumptionTokenSb.toString(), numRows, 0));

      // listIdentifiersMap.put("resumptionMap",

      // getResumptionMap(resumptionTokenSb.toString()));

    }

    listIdentifiersMap.put("headers", headers.iterator());

    listIdentifiersMap.put("identifiers", identifiers.iterator());

    return listIdentifiersMap;

  }

  /**

   * Retrieve the next set of Identifiers associated with the resumptionToken

   * 

   * @param resumptionToken

   *          implementation-dependent format taken from the previous

   *          listIdentifiers() Map result.

   * @return a Map object containing an optional "resumptionToken" key/value

   *         pair and an "identifiers" Map object. The "identifiers" Map

   *         contains OAI identifier keys with corresponding values of "true" or

   *         null depending on whether the identifier is deleted or not.

   * @exception OAIBadRequestException

   *              signals an http status code 400 problem

   */

  public Map listIdentifiers(String resumptionToken)

      throws BadResumptionTokenException {

    purge(); // clean out old resumptionTokens

    Map listIdentifiersMap = new HashMap();

    ArrayList headers = new ArrayList();

    ArrayList identifiers = new ArrayList();

    /**********************************************************************

     * parse your resumptionToken and look it up in the resumptionResults, if

     * necessary

     **********************************************************************/

    StringTokenizer tokenizer = new StringTokenizer(resumptionToken, ":");

    String resumptionId;

    int oldCount;

    String metadataPrefix;

    int numRows;

    try {

      resumptionId = tokenizer.nextToken();

      oldCount = Integer.parseInt(tokenizer.nextToken());

      numRows = Integer.parseInt(tokenizer.nextToken());

      metadataPrefix = tokenizer.nextToken();

    } catch (NoSuchElementException e) {

      throw new BadResumptionTokenException();

    }

    /* Get some more records from your database */

    Iterator iterator = (Iterator) resumptionResults.remove(resumptionId);

    if (iterator == null) {

      System.out

          .println("MetacatCatalog.listIdentifiers(): reuse of old resumptionToken?");

      iterator = dateMap.entrySet().iterator();

      for (int i = 0; i < oldCount; ++i)

        iterator.next();

    }

    /* load the headers and identifiers ArrayLists. */

    int count = 0;

    while (count < maxListSize && iterator.hasNext()) {

      Map.Entry entryDateMap = (Map.Entry) iterator.next();

      HashMap nativeHeader = getNativeHeader((String) entryDateMap.getKey());

      String[] header = getRecordFactory().createHeader(nativeHeader);

      headers.add(header[0]);

      identifiers.add(header[1]);

      count++;

    }

    /* decide if you're done. */

    if (iterator.hasNext()) {

      resumptionId = getRSName();

      resumptionResults.put(resumptionId, iterator);

      /*****************************************************************

       * Construct the resumptionToken String however you see fit.

       *****************************************************************/

      StringBuffer resumptionTokenSb = new StringBuffer();

      resumptionTokenSb.append(resumptionId);

      resumptionTokenSb.append(":");

      resumptionTokenSb.append(Integer.toString(oldCount + count));

      resumptionTokenSb.append(":");

      resumptionTokenSb.append(Integer.toString(numRows));

      resumptionTokenSb.append(":");

      resumptionTokenSb.append(metadataPrefix);

      /*****************************************************************

       * Use the following line if you wish to include the optional

       * resumptionToken attributes in the response. Otherwise, use the line

       * after it that I've commented out.

       *****************************************************************/

      listIdentifiersMap.put("resumptionMap", getResumptionMap(

          resumptionTokenSb.toString(), numRows, oldCount));

      // listIdentifiersMap.put("resumptionMap",

      // getResumptionMap(resumptionTokenSb.toString()));

    }

    listIdentifiersMap.put("headers", headers.iterator());

    listIdentifiersMap.put("identifiers", identifiers.iterator());

    return listIdentifiersMap;

  }

  /**

   * Retrieve a list of records that satisfy the specified criteria

   * 

   * @param from

   *          beginning date in the form of YYYY-MM-DD or null if earliest date

   *          is desired

   * @param until

   *          ending date in the form of YYYY-MM-DD or null if latest date is

   *          desired

   * @param set

   *          set name or null if no set is desired

   * @param metadataPrefix       

   *          e.g. "oai_dc", "eml-2.0.1", "eml-2.1.0"

   *        

   * @return a Map object containing an optional "resumptionToken" key/value

   *         pair and a "records" Iterator object. The "records" Iterator

   *         contains a set of Records objects.

   * @exception OAIBadRequestException

   *              signals an http status code 400 problem

   * @exception OAIInternalServerError

   *              signals an http status code 500 problem

   */

  public Map listRecords(String from, String until, String set,

                         String metadataPrefix) 

      throws CannotDisseminateFormatException,

             OAIInternalServerError, 

             NoItemsMatchException 

  {

    purge(); // clean out old resumptionTokens

    Map<String, Object> listRecordsMap = new HashMap<String, Object>();

    ArrayList<String> records = new ArrayList<String>();

    filteredDateMap = filterDateMap(from, until, metadataPrefix);

    Iterator iterator = filteredDateMap.entrySet().iterator();

    int numRows = filteredDateMap.entrySet().size();

    int count = 0;

    while (count < maxListSize && iterator.hasNext()) {

      Map.Entry entryDateMap = (Map.Entry) iterator.next();

      try {

        String localIdentifier = (String) entryDateMap.getKey();

        HashMap<String, String> nativeItem =getMetacatDocument(localIdentifier);

        String record = constructRecord(nativeItem, metadataPrefix);

        records.add(record);

        count++;

      } 

      catch (IOException e) {

        e.printStackTrace();

        throw new OAIInternalServerError(e.getMessage());

      }

    }

    if (count == 0) { throw new NoItemsMatchException(); }

    /* decide if you're done */

    if (iterator.hasNext()) {

      String resumptionId = getRSName();

      resumptionResults.put(resumptionId, iterator);

      /*****************************************************************

       * Construct the resumptionToken String however you see fit.

       *****************************************************************/

      StringBuffer resumptionTokenSb = new StringBuffer();

      resumptionTokenSb.append(resumptionId);

      resumptionTokenSb.append(":");

      resumptionTokenSb.append(Integer.toString(count));

      resumptionTokenSb.append(":");

      resumptionTokenSb.append(Integer.toString(numRows));

      resumptionTokenSb.append(":");

      resumptionTokenSb.append(metadataPrefix);

      /*****************************************************************

       * Use the following line if you wish to include the optional

       * resumptionToken attributes in the response. Otherwise, use the line

       * after it that I've commented out.

       *****************************************************************/

      listRecordsMap.put("resumptionMap", 

                         getResumptionMap(resumptionTokenSb.toString(), 

                                          numRows, 0

                                         )

                        );

      // listRecordsMap.put("resumptionMap",

      // getResumptionMap(resumptionTokenSb.toString()));

    }

    listRecordsMap.put("records", records.iterator()); 

    return listRecordsMap;

  }

  /**

   * Retrieve the next set of records associated with the resumptionToken

   * 

   * @param resumptionToken

   *          implementation-dependent format taken from the previous

   *          listRecords() Map result.

   * @return a Map object containing an optional "resumptionToken" key/value

   *         pair and a "records" Iterator object. The "records" Iterator

   *         contains a set of Records objects.

   * @exception OAIBadRequestException

   *              signals an http status code 400 problem

   */

  public Map listRecords(String resumptionToken)

      throws BadResumptionTokenException {

    purge(); // clean out old resumptionTokens

    Map listRecordsMap = new HashMap();

    ArrayList records = new ArrayList();

    /**********************************************************************

     * parse your resumptionToken and look it up in the resumptionResults, if

     * necessary

     **********************************************************************/

    StringTokenizer tokenizer = new StringTokenizer(resumptionToken, ":");

    String resumptionId;

    int oldCount;

    String metadataPrefix;

    int numRows;

    try {

      resumptionId = tokenizer.nextToken();

      oldCount = Integer.parseInt(tokenizer.nextToken());

      numRows = Integer.parseInt(tokenizer.nextToken());

      metadataPrefix = tokenizer.nextToken();

    } 

    catch (NoSuchElementException e) {

      throw new BadResumptionTokenException();

    }

    /* Get some more records from your database */

    Iterator iterator = (Iterator) resumptionResults.remove(resumptionId);

    if (iterator == null) {

      System.out

          .println("MetacatCatalog.listRecords(): reuse of old resumptionToken?");

      iterator = dateMap.entrySet().iterator();

      for (int i = 0; i < oldCount; ++i)

        iterator.next();

    }

    /* load the records ArrayLists. */

    int count = 0;

    while (count < maxListSize && iterator.hasNext()) {

      Map.Entry entryDateMap = (Map.Entry) iterator.next();

      try {

        String localIdentifier = (String) entryDateMap.getKey();

        HashMap nativeItem = getMetacatDocument(localIdentifier);

        String record = constructRecord(nativeItem, metadataPrefix);

        records.add(record);

        count++;

      } 

      catch (CannotDisseminateFormatException e) {

        /* the client hacked the resumptionToken beyond repair */

        throw new BadResumptionTokenException();

      } 

      catch (IOException e) {

        /* the file is probably missing */

        throw new BadResumptionTokenException();

      }

    }

    /* decide if you're done. */

    if (iterator.hasNext()) {

      resumptionId = getRSName();

      resumptionResults.put(resumptionId, iterator);

      /*****************************************************************

       * Construct the resumptionToken String however you see fit.

       *****************************************************************/

      StringBuffer resumptionTokenSb = new StringBuffer();

      resumptionTokenSb.append(resumptionId);

      resumptionTokenSb.append(":");

      resumptionTokenSb.append(Integer.toString(oldCount + count));

      resumptionTokenSb.append(":");

      resumptionTokenSb.append(Integer.toString(numRows));

      resumptionTokenSb.append(":");

      resumptionTokenSb.append(metadataPrefix);

      /*****************************************************************

       * Use the following line if you wish to include the optional

       * resumptionToken attributes in the response. Otherwise, use the line

       * after it that I've commented out.

       *****************************************************************/

      listRecordsMap.put("resumptionMap", 

                         getResumptionMap(resumptionTokenSb.toString(), 

                         numRows, 

                         oldCount)

                        );

      // listRecordsMap.put("resumptionMap",

      // getResumptionMap(resumptionTokenSb.toString()));

    }

    listRecordsMap.put("records", records.iterator());

    return listRecordsMap;

  }

  public Map listSets() throws NoSetHierarchyException {

    throw new NoSetHierarchyException();

    // Map listSetsMap = new HashMap();

    // listSetsMap.put("sets", setsList.iterator());

    // return listSetsMap;

  }

  public Map listSets(String resumptionToken)

      throws BadResumptionTokenException {

    throw new BadResumptionTokenException();

  }

  /**

   * Run a query of the Metacat database to load the catalog of EML documents.

   * For each EML document, we store its 'docid', 'doctype', and 'date_updated'

   * values.

   */

  public void loadCatalog() {

    Statement stmt;

    try {

      Connection conn = getConnection();