Metacat

/**

 *  '$RCSfile$'

 *  Copyright: 2009 University of New Mexico and the 

 *                  Regents of the University of California

 *

 *   '$Author: costa $'

 *     '$Date: 2009-07-27 17:47:44 -0400 (Mon, 27 Jul 2009) $'

 * '$Revision: 4999 $'

 *

 * 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

 * 

 * Additional 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.harvester;

import java.io.*;

import java.lang.NoSuchFieldException;

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.util.HashMap;

import java.util.StringTokenizer;

import javax.xml.parsers.DocumentBuilder;

import javax.xml.parsers.DocumentBuilderFactory;

import javax.xml.parsers.ParserConfigurationException;

import javax.xml.transform.TransformerException;

import org.apache.log4j.BasicConfigurator;

import org.apache.log4j.Logger;

import org.w3c.dom.Document;

import org.w3c.dom.Element;

import org.w3c.dom.Node;

import org.w3c.dom.NodeList;

import org.w3c.dom.Text;

import org.xml.sax.InputSource;

import org.xml.sax.SAXException;

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

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

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

import edu.ucsb.nceas.utilities.PropertyNotFoundException;

/**

 * Main class for running the OAI-PMH Harvester program

 * 

 * @author dcosta

 *

 */

public class OaipmhHarvester {

  /* Class variables */

  private static final String METACAT_CONFIG_DIR = "../../build/war/WEB-INF";

  private static HashMap<String, String> metacatDatestamps = 

                                                  new HashMap<String, String>();

  private static HashMap<String, Integer> metacatRevisions = 

                                                 new HashMap<String, Integer>();

  private static Metacat metacatClient = null;

  private static String metacatURL = null;

  private static Logger logger = Logger.getLogger(OaipmhHarvester.class);

  static {

    BasicConfigurator.configure();

  }

  /*

   * Query string to determine the 'date_updated' value stored

   * in Metacat's 'xml_documents' table for a given docid value.

   */

  private static final String METACAT_QUERY =

                           "SELECT docid, rev, date_updated FROM xml_documents";

  /* Class methods */

  /**

   * Converts a Dryad identifier to a Metacat docid (scope + identifier)

   * 

   * @param dryadID  The Dryad identifier, e.g.

   *                 "oai:dryad-dev.nescent.org:10255/dryad.12"

   * @return  Metacat docid, e.g. "10255/dryad.12"

   */

  private static String docidFromDryadIdentifier(String dryadID) {

    String docid = null;

    String scopeAndIdentifier = null;

    String scope = null;

    String identifier = null;  

    StringTokenizer stringTokenizer = new StringTokenizer(dryadID, ":");

    String token = null;

    int tokenCount = stringTokenizer.countTokens();

    int i = 1;    

    while (stringTokenizer.hasMoreTokens()) {

      token = stringTokenizer.nextToken();

      if (i == tokenCount) { scopeAndIdentifier = token; }

      i++;

    }

    if (scopeAndIdentifier != null) {

      stringTokenizer = new StringTokenizer(scopeAndIdentifier, ".");

      tokenCount = stringTokenizer.countTokens();

      if (tokenCount == 2) {  

        i = 1;

        while (stringTokenizer.hasMoreTokens()) {

          token = stringTokenizer.nextToken();

          if (i == (tokenCount - 1)) { scope = token; }

          if (i == tokenCount) { identifier = token; }

          i++;

        }

      }

      else {

        logger.error("Error parsing Dryad identifier: " + dryadID);

      }

    }

    if (scope != null && identifier != null) {

      scope = scope.replace('/', '-'); // Metacat doesn't allow '/' in docid

      docid = scope + "." + identifier;

    }

    return docid;

  }

  /**

   * Converts an OAI-PMH identifier to a Metacat docid (scope + identifier)

   * 

   * @param   identifier    the OAI-PMH identifier

   * @return  docid         Metacat docid

   */

  private static String docidFromIdentifier(String identifier) {

    String docid = null;

    /*

     * Call the appropriate method to convert identifier to a Metacat docid.

     */

    if (identifier != null) {

      /*

       * Check for LSID syntax.

       */

      if (identifier.startsWith("urn:lsid:")) {

        docid = docidFromLSID(identifier);

      }

      /* Dryad identifier: http://hdl.handle.net/10255/dryad.66

       * Equivalent Metacat identifier: 10255-dryad.66.1

       */

      else if (identifier.contains("/dryad.")) {

        docid = docidFromDryadIdentifier(identifier);

      }

    }

    return docid;

  }

  /**

   * Converts an LSID identifier to a Metacat docid (scope + identifier)

   * 

   * @param lsidIdentifier  The LSID identifier, e.g.

   *                        "urn:lsid:knb.ecoinformatics.org:knb-lter-sgs:6"

   * @return  Metacat docid, e.g. "knb-lter-sgs.6"

   */

  private static String docidFromLSID(String lsidIdentifier) {

    String docid = null;

    String scope = null;

    String identifier = null;  

    StringTokenizer stringTokenizer = new StringTokenizer(lsidIdentifier, ":");

    int tokenCount = stringTokenizer.countTokens();

    int i = 1;    

    while (stringTokenizer.hasMoreTokens()) {

      String token = stringTokenizer.nextToken();

      if (i == (tokenCount - 1)) { scope = token; }

      if (i == tokenCount) { identifier = token; }

      i++;

    }

    if (scope != null && identifier != null) {

      docid = scope + "." + identifier;

    }

    return docid;

  }

  /**

   * Extracts the metadata content from the XML string returned by the GetRecord

   * verb.

   * 

   * @param getRecordString    The XML string returned by the GetRecord verb

   *                           operation.

   * @return  metadataString   The document string extracted from the GetRecord

   *                           XML string.

   */

  private static String extractMetadata(String getRecordString) {

    String metadataString = null;

    StringBuffer stringBuffer = 

               new StringBuffer("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n");

    /* The document string is everything between the <metadata> and </metadata>

     * tags.

     */

    int metadataStartIndex = getRecordString.indexOf("<metadata>");

    int metadataEndIndex = getRecordString.indexOf("</metadata>");

    if ((metadataStartIndex >= 0) &&

        (metadataEndIndex >= 0) &&

        (metadataStartIndex < metadataEndIndex)

       ) {

      int startPosition = metadataStartIndex + "<metadata>".length();

      int endPosition = metadataEndIndex;

      String docString = getRecordString.substring(startPosition, endPosition);

      stringBuffer.append(docString);

      stringBuffer.append("\n");

      metadataString = stringBuffer.toString();

    }

    return metadataString;

  }

  /**

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

   * has not already been made previously.

   * 

   * @return  conn  the database Connection object

   */

  private static Connection getConnection() {

    Connection conn = null;

    String dbDriver = "";

    String defaultDB = null;

    String password = null;

    String user = null;

    SQLWarning warn;

    if (conn == null) {

        try {

          dbDriver = PropertyService.getProperty("database.driver");

          defaultDB = PropertyService.getProperty("database.connectionURI");

          password = PropertyService.getProperty("database.password");

          user = PropertyService.getProperty("database.user");

        } 

        catch (PropertyNotFoundException pnfe) {

          logger.error("Can't find database connection property " + pnfe);

          System.exit(1);

        }

      // Load the jdbc driver

      try {

        Class.forName(dbDriver);

      }

      catch (ClassNotFoundException e) {

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

        System.exit(1);

      } 

      // Make the database connection

      try {

        conn = DriverManager.getConnection(defaultDB, user, password);

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

        // There may be multiple warnings chained.

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

        System.exit(1);

      }

    }

    return conn;

  }

  /**

   * Parses command line options and packages them into a HashMap.

   *  

   * @param   args     array of command-line strings

   * @return  options  HashMap of option/value pairs

   */

  private static HashMap<String, String> getOptions(String[] args) {

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

    boolean foundDN = false;

    boolean foundPassword = false;

    for (int i=0; i<args.length; ++i) {

      if (args[i].charAt(0) != '-') {

        options.put("baseURL", args[i]);

      } 

      else if (i + 1 < args.length) {

        if (args[i].equals("-dn")) { foundDN = true; }

        if (args[i].equals("-password")) { foundPassword = true; }

        options.put(args[i], args[++i]);

      }

      else {

        throw new IllegalArgumentException();

      }

    }

    // Check for required command-line options "-dn" and "-password"

    if (!foundDN || !foundPassword) { throw new IllegalArgumentException(); }

    return options;

  }

  /**

   * Boolean to determine whether the content returned from the GetRecord verb

   * indicates a deleted document.

   * 

   * @param    getRecordString    the content returned by the GetRecord verb

   * @return   true if this is a deleted record, else false

   */

  private static boolean isDeletedRecord(String getRecordString) {

    boolean isDeleted = false;

    final String DELETED_FLAG_1 = "status=\"deleted\"";

    final String DELETED_FLAG_2 = "status='deleted'";

    if (getRecordString != null) {

      if ((getRecordString.contains(DELETED_FLAG_1) ||

           getRecordString.contains(DELETED_FLAG_2)

          ) &&

          !getRecordString.contains("<metadata>")

         ) {

        isDeleted = true;

      }

    }

    return isDeleted;

  }

  /**

   * Load datestamps for all Metacat documents. This will be used to determine

   * whether the document in the OAI-PMH repository is newer than the copy

   * in Metacat. If it is newer, the document should be harvested.

   */

  private static void loadMetacatCatalog() {

    try {

      Connection conn = getConnection();    

      if (conn != null) {

        Statement stmt = conn.createStatement();                          

        ResultSet rs = stmt.executeQuery(METACAT_QUERY);

        while (rs.next()) {

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

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

          int rev = rs.getInt("rev");

          Integer revInteger = new Integer(rev);

          metacatDatestamps.put(docid, dateUpdated);

          metacatRevisions.put(docid, revInteger);

        }

        stmt.close();   

        conn.close();

      }

    }

    catch(SQLException e) {

      metacatDatestamps = null;

      metacatRevisions = null;

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

    }

  }

  /**

   * Loads OaipmhHarvester properties from a configuration file. These are

   * configuration values that are not specified on the command line, such

   * as the database connection values. They are typically stored in the

   * 'metacat.properties' file.

   * 

   * @param   metacatConfigDir   The metacat configuration directory.

   *                             Typically, the directory in which the

   *                             'metacat.properties' file is found.

   */

  private static void loadProperties(String metacatConfigDir) {   

    try {

        PropertyService.getInstance(metacatConfigDir);

    } 

    catch (ServiceException e) {

      logger.error("Error in loading properties: " + e.getMessage());

    }

  }

  /**

   * The main() method.

   * 

   * @param args    

   * 

   * Command line arguments:

   * 

   *  -dn distinguished_name    -- LDAP user name of the harvester account

   *  -password password        -- LDAP password of the harvester account

   *  <-metacatConfigdir dir>   -- Directory where metacat.properties file is

   *                               found.

   *  <-from date>              -- from date of the harvest documents

   *  <-until date>             -- until date of the harvest documents

   *  <-metadataPrefix prefix>  -- metadata prefix of the harvest documents,

   *                               e.g. 'oai_dc'

   *  <-setSpec setName>        -- set specification of the harvest documents

   *  baseURL                   -- base URL of the OAI-PMH data provider

   *

   *  Command options appearing inside angle brackets (<>) are optional.

   */

  public static void main(String[] args) {

    try {	    

      HashMap<String, String> options = getOptions(args);

      String baseURL = options.get("baseURL");

      String dn = options.get("-dn");                 // LDAP distinguished name

      String password = options.get("-password");     // LDAP password

      String from = (String) options.get("-from");

      String until = (String) options.get("-until");

      String metadataPrefix = (String) options.get("-metadataPrefix");

      String metacatConfigDir = (String) options.get("-metacatConfigDir");

      String setSpec = (String) options.get("-setSpec");

      /* Use default values if the values aren't specified on command line */

      if (metadataPrefix == null) { metadataPrefix = "oai_dc"; }

      if (metacatConfigDir == null) { metacatConfigDir = METACAT_CONFIG_DIR; }

      OaipmhHarvester.loadProperties(metacatConfigDir);

      metacatURL = SystemUtil.getServletURL();

      metacatClient = MetacatFactory.createMetacatConnection(metacatURL);

      OaipmhHarvester.loadMetacatCatalog();

      /* 

       * If the Metacat catalog failed to load then we can't continue on.

       */

      if ((metacatURL != null) && 

          (metacatClient != null) && 

          (metacatDatestamps != null)

         ) {

        run(baseURL, dn, password, from, until, metadataPrefix, setSpec); 

      }

      else {

        logger.error("Unable to load document catalog from Metacat database.");

      }

    }

	catch (IllegalArgumentException e) {

      logger.error("OaipmhHarvester " +

                   "-dn distinguished_name " +

                   "-password password " +

                   "<-from date> " +

                   "<-until date> " +

                   "<-metadataPrefix prefix> " +

                   "<-setSpec setName> " +

                   "baseURL"

                  );

	}

    catch (MetacatInaccessibleException e) {

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

    }

    catch (PropertyNotFoundException e) {

      logger.error("PropertyNotFoundException: " + 

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

    }

    catch (IOException e) {

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

                   e.getMessage()

                  );

    }

	catch (Exception e) {

	  e.printStackTrace();

	  System.exit(-1);

	}

  }

  /**

   * Determines the datestamp for a Metacat document based on the 'date_updated'

   * value stored in the Metacat database for a given 'docid' value.

   * 

   * @param   docid    The metacat docid (scope + revision).

   * @return  String representing the 'date_updated' value stored in the Metacat

   *          database for this document based on its 'docid' value.

   */

  private static String metacatDatestamp(String docid) {

    String metacatDatestamp = metacatDatestamps.get(docid);

    return metacatDatestamp;

  }

  /**

   * Boolean to determine whether Metacat has a document with the specified

   * docid.

   * 

   * @param   docid                   Metacat docid value

   * @return  true if Metacat has this docid, else false

   */

  private static boolean metacatHasDocid(String docid) {

    boolean hadDocid = false;

    String metacatDatestamp = metacatDatestamp(docid);

    if (metacatDatestamp != null) {

      hadDocid = true;                // Metacat has the docid

    }

    return hadDocid;

  }

  /**

   * Login to Metacat using the ldapDN and ldapPwd

   * 

   * @param  ldapDN   the LDAP distinguished name, e.g.

   *                  "uid=dryad,o=LTER,dc=ecoinformatics,dc=org"

   * @param  ldapPwd  the corresponding LDAP password string

   * 

   * @return  loginSuccess, true if login succeeded, else false

   */

  private static boolean metacatLogin(String ldapDN, String ldapPwd) {

    boolean loginSuccess = false;

    try {

      logger.info("Logging in to Metacat: " + ldapDN);

      String response = metacatClient.login(ldapDN, ldapPwd);

      logger.info("Metacat login response: " + response);

      loginSuccess = true;

    } 

    catch (MetacatInaccessibleException e) {

      logger.error("Metacat login failed." + e.getMessage());

    } 

    catch (Exception e) {

      logger.error("Metacat login failed." + e.getMessage());

    }

    return loginSuccess;

  }

  /**

   * Logout from Metacat

   */

  private static void metacatLogout() {

    try {    

      // Log out from the Metacat session

      logger.info("Logging out from Metacat");

      metacatClient.logout();

    }

    catch (MetacatInaccessibleException e) {

      logger.error("Metacat inaccessible: " + e.getMessage());

    }

    catch (MetacatException e) {

      logger.error("Metacat exception: " + e.getMessage());

    }

  }

  /**

   * Determines the revision for a Metacat document based on the 'rev'

   * value stored in the Metacat database for a given 'docid' value.

   * 

   * @param   docid    The metacat docid (scope + revision).

   * @return  Integer representing the 'rev' value stored in the Metacat

   *          database for this document based on its 'docid' value.

   */

  private static Integer metacatRevision(String docid) {

    Integer metacatRevision = metacatRevisions.get(docid);

    return metacatRevision;

  }

  /**

   * Process the output of the ListIdentifiers verb. For each identifier

   * listed, determine whether the document should be harvested (inserted or

   * updated), deleted, or if no action is needed.

   * 

   * @param baseURL          The base URL of the data provider.

   * @param from             Value of 'from' option, a date string or null

   * @param until            Value of 'until' option, a date string or null

   * @param metadataPrefix   Value of 'metadataPrefix' option, may be null

   * @param setSpec          Value of 'setSpec' option, may be null

   * @param xmlString        The XML string from ListIdentifiers

   * @param principal        Distinguished name of the LDAP account for the

   *                         harvester user, 

   *                         e.g. "uid=dryad,o=LTER,dc=ecoinformatics,dc=org"

   */

  private static void processListIdentifiers(String baseURL, 

                                             String from, 

                                             String until,

                                             String metadataPrefix,

                                             String setSpec,

                                             String xmlString,

                                             String principal) {

    DocumentBuilderFactory documentBuilderFactory =

                                           DocumentBuilderFactory.newInstance();

    StringReader stringReader = new StringReader(xmlString);

    try {

      DocumentBuilder documentBuilder = documentBuilderFactory.newDocumentBuilder();

      InputSource inputSource = new InputSource(stringReader);

      Document document = documentBuilder.parse(inputSource);

      Element rootElement = document.getDocumentElement();

      NodeList nodeList = rootElement.getChildNodes();

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

        Node child = nodeList.item(i);

        if (child instanceof Element) {

          Element childElement = (Element) child;

          if (childElement.getTagName().equals("ListIdentifiers")) {

            NodeList listIdentifiersNodeList = childElement.getChildNodes();

            for (int j = 0; j < listIdentifiersNodeList.getLength(); j++) {

              Node listIdentifiersNode = listIdentifiersNodeList.item(j);

              if (listIdentifiersNode instanceof Element) {

                Element listIdentifiersElement = (Element) listIdentifiersNode;

                if (listIdentifiersElement.getTagName().equals("header")) {

                  NodeList headerNodeList = listIdentifiersElement.getChildNodes();

                  String identifier = null;

                  String datestamp = null;

                  for (int k = 0; k < headerNodeList.getLength(); k++) {

                    Node headerNode = headerNodeList.item(k);

                    if (headerNode instanceof Element) {

                      Element headerElement = (Element) headerNode;

                      if (headerElement.getTagName().equals("identifier")) {

                        Text textNode = (Text) headerElement.getFirstChild();

                        identifier = textNode.getData().trim();

                      }

                      else if (headerElement.getTagName().equals("datestamp")) {

                        Text textNode = (Text) headerElement.getFirstChild();

                        datestamp = textNode.getData().trim();

                      }             

                    }

                  }

                  if (identifier != null) {

                    String docid = docidFromIdentifier(identifier);

                    logger.debug("identifier: " + identifier + 

                                 "; docid: " + docid + 

                                 "; datestamp: " + datestamp);

                    if (docid != null) { 

                      if (shouldHarvestDocument(docid, datestamp)) {                    

                        GetRecord getRecord = 

                             new GetRecord(baseURL, identifier, metadataPrefix);

                        getRecord.runVerb();  // Run the GetRecord verb

                        NodeList errors = getRecord.getErrors();

                        if (errors != null && errors.getLength() > 0) {

                          logger.error("Found errors in GetRecord results");

                          int length = errors.getLength();

                          for (int l = 0; l < length; ++l) {

                            Node item = errors.item(l);

                            logger.error(item);

                          }

                          logger.error("Error record: " + getRecord.toString());

                        }

                        else {

                          String getRecordString = getRecord.toString();

                          boolean isDeleted = isDeletedRecord(getRecordString);

                          if (isDeleted) {

                            logger.info("GetRecord indicates deleted record: " + 

                                        docid);

                            if (metacatHasDocid(docid)) {

                              logger.info(

                                        "Deleting " + docid + " from Metacat.");

                              String deleteReturnString = null;

                              deleteReturnString = metacatClient.delete(docid);

                              if (deleteReturnString != null && 

                                  !deleteReturnString.equals("")) {

                                logger.info(deleteReturnString);

                              }

                            }                           

                          }

                          else {

                            String metadataString = 

                                               extractMetadata(getRecordString);

                            uploadToMetacat(docid, datestamp, metadataString, 

                                            principal);

                          }

                        }

                      }

                      else {

                        logger.info(

                          "Not harvesting docid '" + docid + 

                          "' from the OAI-PMH provider. " +

                          "Metacat already has this document at datestamp '" + 

                          datestamp + "' or higher.");

                      }

                    }

                    else {

                      logger.warn("Unrecognized identifier format: " +

                                  identifier);

                    }

                  }

                }             

              }

            }

          }

        }

      }

    }

    catch (Exception e) {

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

      e.printStackTrace();

    }

  }

  /**

   * Runs a OAI-PMH harvest.

   * 

   * @param baseURL          The base URL of the data provider.

   * @param dn               Value of 'dn' option, a LDAP distinguished name,

   *                         e.g. "uid=dryad,o=LTER,dc=ecoinformatics,dc=org"

   * @param password         Value of 'password' option, a string

   * @param from             Value of 'from' option, a date string or null

   * @param until            Value of 'until' option, a date string or null

   * @param metadataPrefix   Value of 'metadataPrefix' option, may be null

   * @param setSpec          Value of 'setSpec' option, may be null

   * @throws IOException

   * @throws ParserConfigurationException

   * @throws SAXException

   * @throws TransformerException

   * @throws NoSuchFieldException

   */

  public static void run(String baseURL, String dn, String password, 

                         String from, String until,

                         String metadataPrefix, String setSpec

                        )

          throws IOException, ParserConfigurationException, SAXException, 

                 TransformerException, NoSuchFieldException 

  {

    logger.info("Starting OAI-PMH Harvester.");

    if ((dn != null) && (password != null)) {

      boolean loginSuccess = metacatLogin(dn, password);

      // Terminate harvester execution if login failed

      if (!loginSuccess) { 

        logger.warn("Terminating OAI-PMH Harvester execution due to login failure.");

        return; 

      } 

    }

    else {

      logger.error("Distinguished name (-dn) and/or password (-password) " +

      		       "were not specified.");

      return;

    }

    ListIdentifiers listIdentifiers = 

             new ListIdentifiers(baseURL, from, until, metadataPrefix, setSpec);

    listIdentifiers.runVerb();

    while (listIdentifiers != null) {

      NodeList errors = listIdentifiers.getErrors();

      if (errors != null && errors.getLength() > 0) {

        logger.error("Found errors in ListIdentifier results");

        int length = errors.getLength();

        for (int i = 0; i < length; ++i) {

          Node item = errors.item(i);

          logger.error(item);

        }

        logger.error("Error record: " + listIdentifiers.toString());

        break;

      }

      String xmlString = listIdentifiers.toString();

      processListIdentifiers(baseURL, from, until, metadataPrefix, setSpec,

                             xmlString, dn);

      String resumptionToken = listIdentifiers.getResumptionToken();

      logger.debug("resumptionToken: " + resumptionToken);

      if (resumptionToken == null || resumptionToken.length() == 0) {

        listIdentifiers = null;

      } 

      else {

        listIdentifiers = new ListIdentifiers(baseURL, resumptionToken);

        listIdentifiers.runVerb();

      }

    }

    metacatLogout();

    logger.info("Harvest completed. Shutting down OAI-PMH Harvester.");

  }

  /**

   * Should a document be harvested? Compare the OAI-PMH provider datestamp to 

   * the Metacat datestamp (the 'last_updated' date). If the Metacat datestamp 

   * is unknown, or if it's less than the OAI-PMH datestamp, then the document

   * should be harvested.

   *  

   * @param docid                   The Metacat docid value.

   * @param providerDatestamp       The OAI-PMH provider datestamp.

   * @return   true if the document should be harvested into Metacat, else false

   */

  private static boolean shouldHarvestDocument(String docid, 

                                               String providerTimestamp

                                              ) {

    String providerDatestamp;

    boolean shouldHarvest = false;

    String metacatDatestamp = metacatDatestamp(docid);

    /*

     * Since Metacat stores its 'last_updated' field as a datestamp (no time),

     * we need to strip off the timestamp part of the provider timestamp

     * before doing a comparison of the Metacat datestamp to the OAI-PMH

     * provider datestamp.

     */

    if (providerTimestamp.contains("T")) {

      int tIndex = providerTimestamp.indexOf('T');

      providerDatestamp = providerTimestamp.substring(0, tIndex);

    }

    else {

      providerDatestamp = providerTimestamp;

    }

    /*

     * If we don't have a Metacat datastamp for this document, or if the

     * Metacat datestamp is older than the provider datestamp, then we

     * should harvest the document.

     */

    if (metacatDatestamp == null) {

      shouldHarvest = true;

    }

    else if (metacatDatestamp.compareTo(providerDatestamp) < 0) {

        shouldHarvest = true;

    }

    return shouldHarvest;

  }

  /**

   * Insert or update the document to Metacat. If Metacat already has this

   * document, increment the 'rev' number by 1 to update it.

   * 

   * @param   docid           The Metacat docid

   * @param   datestamp       The datestamp in the OAI-PMH provider catalog.

   * @param   metadataString  The metadata string extracted by the GetRecord 

   * @param   principal       The distinguished name of the principal

   *                          verb

   * @return  true if the upload succeeded, else false.

   */

  private static boolean uploadToMetacat(String docid,

                                         String datestamp,

                                         String metadataString,

                                         String principal) {

    String docidFull = null;

    boolean success = true;

    String metacatDatestamp = metacatDatestamp(docid);

    Integer metacatRevision = metacatRevision(docid);

    boolean insert = false;

    StringReader stringReader = null;

    boolean update = false;

    if (metadataString != null ) {

      stringReader = new StringReader(metadataString);

      /* If metacat already has this document, determine the highest revision in

       * metacat and report it to the user; else, insert or delete the document 

       * into metacat.

       */

      if (metacatDatestamp == null) {

        insert = true;

        int newRevision = 1;

        docidFull = docid + "." + newRevision;

      }

      else if (metacatDatestamp.compareTo(datestamp) < 0) {

        update = true;

        int newRevision = metacatRevision + 1;

        docidFull = docid + "." + newRevision;

      }

      else if (metacatDatestamp.compareTo(datestamp) == 0) {

        logger.warn("Attempting to update " + docid + " to datestamp " + 

            datestamp + ". Metacat has document at datestamp " +

            metacatDatestamp + ".");