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

import edu.ucsb.nceas.metacat.service.ServiceException;

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

  /** Database connection */

  private static String metacatDBDriver;

  private static String metacatDBURL;

  private static String metacatDBUser;

  private static String metacatDBPassword;

  private static String metacatURL;

  /* Instance fields */

  private SimpleDateFormat dateFormatter = new SimpleDateFormat();

  protected String homeDir;

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

  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 docid, doctype, date_updated " +

  "FROM xml_documents " +

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

  "  (docid IN " +

  "     (SELECT docid " +

  "      FROM xml_access " +

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

  "             perm_type = 'allow' AND " +

  "             permission > 3" +

  "           )" +

  "     )" +

  "   AND " +

  "   docid NOT IN " +

  "     (SELECT docid " +

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

    dateFormatter.applyPattern("yyyy-MM-dd'T'HH:mm:ss'Z'");

    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;

  }

  private String date2OAIDatestamp(Date date) {

    return dateFormatter.format(date);

  }

  /**

   * 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 = 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;

  }

  /**

   * 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

   * @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 listIdentifiersMap = new HashMap();

    ArrayList headers = new ArrayList();

    ArrayList identifiers = new ArrayList();

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

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

    int count = 0;

    RecordFactory recordFactory = getRecordFactory();

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

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

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

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

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

        HashMap nativeHeader = getNativeHeader(key);

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

        headers.add(header[0]);

        identifiers.add(header[1]);

        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;

  }

  /**

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

      if (conn != null) {

        stmt = conn.createStatement();                          

        ResultSet rs = stmt.executeQuery(QUERY);

        int documentCount = 0;

        while (rs.next()) {

          documentCount++;

          String docid = rs.getString("docid");

          String doctype = rs.getString("doctype");

          String dateUpdated = rs.getDate("date_updated").toString();

          docTypeMap.put(docid, doctype);

          dateMap.put(docid, dateUpdated);

        }

        logger.info("Number of documents in catalog: " + documentCount);

        stmt.close();   

        conn.close();

      }

    }

    catch(SQLException e) {

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

    }

  }

  /**

   * 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

   *          the OAI metadataPrefix

   * @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 listRecordsMap = new HashMap();

    ArrayList records = new ArrayList();

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

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

    int count = 0;

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

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

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

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

        try {

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

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

  }

  /**

   * Purge tokens that are older than the time-to-live.

   */

  private void purge() {

    ArrayList old = new ArrayList();

    Date then, now = new Date();

    Iterator keySet = resumptionResults.keySet().iterator();

    String key;

    while (keySet.hasNext()) {

      key = (String) keySet.next();

      then = new Date(Long.parseLong(key) + getMillisecondsToLive());

      if (now.after(then)) {

        old.add(key);

      }

    }

    Iterator iterator = old.iterator();

    while (iterator.hasNext()) {

      key = (String) iterator.next();

      resumptionResults.remove(key);

    }

  }

}