Metacat

/**

 *  '$RCSfile$'

 *    Purpose: A class to asyncronously do delta-T replication checking

 *  Copyright: 2000 Regents of the University of California and the

 *             National Center for Ecological Analysis and Synthesis

 *    Authors: Chad Berkley

 *    Release: @release@

 *

 *   '$Author: tao $'

 *     '$Date: 2002-04-16 18:21:57 -0700 (Tue, 16 Apr 2002) $'

 * '$Revision: 1015 $'

 *

 * This program is free software; you can redistribute it and/or modify

 * it under the terms of the GNU General Public License as published by

 * the Free Software Foundation; either version 2 of the License, or

 * (at your option) any later version.

 *

 * This program is distributed in the hope that it will be useful,

 * but WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

 * GNU General Public License for more details.

 *

 * You should have received a copy of the GNU General Public License

 * along with this program; if not, write to the Free Software

 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

 */

package edu.ucsb.nceas.metacat;

import java.sql.*;

import java.util.*;

import java.lang.Thread; 

import java.io.*;

import java.net.*;

import java.text.*;

import org.xml.sax.AttributeList;

import org.xml.sax.ContentHandler;

import org.xml.sax.DTDHandler;

import org.xml.sax.EntityResolver;

import org.xml.sax.ErrorHandler;

import org.xml.sax.InputSource;

import org.xml.sax.XMLReader;

import org.xml.sax.SAXException;

import org.xml.sax.SAXParseException;

import org.xml.sax.helpers.XMLReaderFactory;

import org.xml.sax.helpers.DefaultHandler;

/**

 * This class handles deltaT replication checking.  Whenever this TimerTask

 * is fired it checks each server in xml_replication for updates and updates

 * the local db as needed.

 */

public class ReplicationHandler extends TimerTask

{

  int serverCheckCode = 1;

  MetaCatUtil util = new MetaCatUtil();

  Hashtable serverList = new Hashtable(); 

  Connection conn;

  PrintWriter out;

  public ReplicationHandler(PrintWriter o)

  {

    this.out = o;

  }

  public ReplicationHandler(PrintWriter o, int serverCheckCode)

  {

    this.out = o;

    this.serverCheckCode = serverCheckCode;

  }

  /**

   * Method that implements TimerTask.run().  It runs whenever the timer is 

   * fired.

   */

  public void run()

  {

    //find out the last_checked time of each server in the server list and

    //send a query to each server to see if there are any documents in 

    //xml_documents with an update_date > last_checked

    try

    {

      try

      { //this connection is prone to error for some reason so we 

        //try to connect it three times before quiting.

        conn = util.openDBConnection();

      }

      catch(SQLException sqle)

      {

        try

        {

          conn = util.openDBConnection();

        }

        catch(SQLException sqlee)

        {

          try

          {

            conn = util.openDBConnection();

          }

          catch(SQLException sqleee)

          {

            System.out.println("error getting db connection in " + 

                               "ReplicationHandler.run: " +

                               sqleee.getMessage());

          }

        }

      }

      serverList = buildServerList(conn);

      updateCatalog(serverList);

      update(serverList);

      conn.close();

    }

    catch (Exception e)

    {

      //System.out.println("Error in replicationHandler.run():"+e.getMessage());

    }

  }

  /**

   * Method that uses revision taging for replication instead of update_date.

   */

  private void update(Hashtable serverList)

  {

    /*

     Pseudo-algorithm

     - request a doc list from each server in xml_replication

     - check the rev number of each of those documents agains the 

       documents in the local database

     - pull any documents that have a lesser rev number on the local server

       from the remote server

     - delete any documents that still exist in the local xml_documents but

       are in the deletedDocuments tag of the remote host response.

     - update last_checked to keep track of the last time it was checked.

       (this info is theoretically not needed using this system but probably 

       should be kept anyway)

    */

    Connection conn = null;

    try

    {

      try

      { //this connection is prone to error for some reason so we 

        //try to connect it three times before quiting.

        conn = util.openDBConnection();

      }

      catch(SQLException sqle)

      {

        try

        {

          conn = util.openDBConnection();

        }

        catch(SQLException sqlee)

        {

          try

          {

            conn = util.openDBConnection();

          }

          catch(SQLException sqleee)

          {

            /*System.out.println("error getting db connection in " + 

                               "ReplicationHandler.update: " +

                               sqleee.getMessage());*/

          }

        }

      }

    }

    catch(Exception e)

    {

      /*System.out.println("error in ReplicationHandler.update: " + 

                          e.getMessage());*/

    }

    Enumeration keys;

    String server;

    String update;

    ReplMessageHandler message = new ReplMessageHandler();

    Vector responses = new Vector();

    PreparedStatement pstmt = null;

    ResultSet rs;

    boolean tablehasrows;

    boolean flag=false;

    String action = new String();

    XMLReader parser;

    URL u;

    try

    {

      MetaCatUtil.debugMessage("init parser");

      parser = initParser(message);

      keys = serverList.keys();

      while(keys.hasMoreElements())

      {

        server = (String)(keys.nextElement());

        MetacatReplication.replLog("full update started to: " + server);

        u = new URL("https://" + server + "?server="

        +util.getLocalReplicationServerName()+"&action=update");

        //System.out.println("Sending Message: " + u.toString());

        String result = MetacatReplication.getURLContent(u);

        responses.add(result);

      }

      //String srvr = util.getOption("servletpath");

      //System.out.println("responses (from srvr): " + responses.toString());

      for(int i=0; i<responses.size(); i++)

      { //check each server for updated files

        //System.out.println("parsing responses");

        parser.parse(new InputSource(

                     new StringReader(

                     (String)(responses.elementAt(i)))));

        Vector v = new Vector(message.getUpdatesVect());

        //System.out.println("v: " + v.toString());

        Vector d = new Vector(message.getDeletesVect());

        //check the revs in u to see if there are any newer ones than

        //in the local DB.

        for(int j=0; j<v.size(); j++)

        {

          Vector w = new Vector((Vector)(v.elementAt(j)));

          //System.out.println("w: " + w.toString());

          String docid = (String)w.elementAt(0);

          //System.out.println("docid: " + docid);

          int rev = Integer.parseInt((String)w.elementAt(1));

          //System.out.println("rev: " + rev);

          String docServer = (String)w.elementAt(2);

          //System.out.println("docServer: " + docServer);

          pstmt = conn.prepareStatement("select rev from xml_documents where "+

                                        "docid like '" + docid + "'");

          pstmt.execute();

          rs = pstmt.getResultSet();

          tablehasrows = rs.next();

          if(tablehasrows)

          { //check the revs for an update because this document is in the

            //local DB, it just might be out of date.

            int localrev = rs.getInt(1);

            if(localrev == rev)

            {

              flag = false;

            }

            else if(localrev < rev)

            {//this document needs to be updated so send an read request

              action = "UPDATE";

              flag = true;

            }

          }

          else

          { //insert this document as new because it is not in the local DB

            action = "INSERT";

            flag = true;

          }

          if(flag)

          { //if the document needs to be updated or inserted, this is executed

            u = new URL("https://" + docServer + "?server="+

              util.getLocalReplicationServerName()+"&action=read&docid="+docid);

            //System.out.println("Sending message: " + u.toString());

            String newxmldoc = MetacatReplication.getURLContent(u);

            DocInfoHandler dih = new DocInfoHandler();

            XMLReader docinfoParser = initParser(dih);

            URL docinfoUrl = new URL("https://" + docServer + 

                  "?server="+util.getLocalReplicationServerName()+

                  "&action=getdocumentinfo&docid="+docid);

            //System.out.println("Sending message: " + docinfoUrl.toString());

            String docInfoStr = MetacatReplication.getURLContent(docinfoUrl);

            docinfoParser.parse(new InputSource(new StringReader(docInfoStr)));

            Hashtable docinfoHash = dih.getDocInfo();

            int serverCode = MetacatReplication.getServerCode(docServer);

            //System.out.println("updating doc: " + docid +" action: "+ action);

            //docid should include rev number too

            String accnum=docid+util.getOption("accNumSeparator")+

                                              (String)docinfoHash.get("rev");

            //System.out.println("accnum: "+accnum);

            try

            {

              String newDocid = DocumentImpl.write(conn, 

                              new StringReader(newxmldoc),

                              (String)docinfoHash.get("public_access"),

                              null,  /* the dtd text */

                              action, 

                              accnum, 

                              (String)docinfoHash.get("user_owner"),

                              null, /* null for groups[] */

                              serverCode, 

                              true, /* override */

                              false /* validate */);

              MetacatReplication.replLog("wrote doc " + docid + " from " + 

                                         docServer);

              /*System.out.println("wrote doc " + docid + " from " + 

                                 docServer);*/

            }

            catch(Exception e)

            {

              System.out.println("error writing document in " + 

                                 "ReplicationHandler.update: " + docid +

                                 ": " + e.getMessage());

            }

          }

        }

        for(int k=0; k<d.size(); k++)

        { //delete the deleted documents;

          Vector w = new Vector((Vector)d.elementAt(k));

          String docid = (String)w.elementAt(0);

          if(!alreadyDeleted(docid, conn))

          {

            DocumentImpl.delete(conn, docid, null, null);

            System.out.println("Document " + docid + " deleted.");

            MetacatReplication.replLog("doc " + docid + " deleted");

          }

        }

        keys = serverList.keys();

        while(keys.hasMoreElements())

        {

          server = (String)(keys.nextElement()); 

          URL dateurl = new URL("https://" + server + "?server="+

          util.getLocalReplicationServerName()+"&action=gettime");

          String datexml = MetacatReplication.getURLContent(dateurl);

          String datestr = datexml.substring(11, datexml.indexOf('<', 11));

          StringBuffer sql = new StringBuffer();

          sql.append("update xml_replication set last_checked = to_date('");

          sql.append(datestr).append("', 'YY-MM-DD HH24:MI:SS') where ");

          sql.append("server like '").append(server).append("'");

          //System.out.println("sql: " + sql.toString());

          pstmt.close();

          pstmt = conn.prepareStatement(sql.toString());

          pstmt.executeUpdate();

          //conn.commit();

          System.out.println("last_checked updated to " + datestr + " on " +

                            server);

        }

      }

    }

    catch(Exception e)

    {

      /*System.out.println("error in ReplicationHandler.update: " + 

                          e.getMessage());*/

      e.printStackTrace(System.out);

    }

    finally

    {

      try

      {

        pstmt.close();

      }

      catch(Exception ee) {}

    }

  }

  /**

   * updates xml_catalog with entries from other servers.

   */

  private void updateCatalog(Hashtable serverList)

  {

    Connection conn = null;

    try

    {

      try

      { //this connection is prone to error for some reason so we 

        //try to connect it three times before quiting.

        conn = util.openDBConnection();

      }

      catch(SQLException sqle)

      {

        try

        {

          conn = util.openDBConnection();

        }

        catch(SQLException sqlee)

        {

          try

          {

            conn = util.openDBConnection();

          }

          catch(SQLException sqleee)

          {

            System.out.println("error getting db connection in " + 

                               "ReplicationHandler.update: " +

                               sqleee.getMessage());

          }

        }

      }

      String server;

      Enumeration keys = serverList.keys();

      while(keys.hasMoreElements())

      { //go through each server

        server = (String)(keys.nextElement());

        URL u = new URL("https://" + server + "?server="+

        util.getLocalReplicationServerName()+"&action=getcatalog");

        //System.out.println("sending message " + u.toString());

        String catxml = MetacatReplication.getURLContent(u);

        //System.out.println("catxml: " + catxml);

        CatalogMessageHandler cmh = new CatalogMessageHandler();

        XMLReader catparser = initParser(cmh);

        catparser.parse(new InputSource(new StringReader(catxml)));

        //parse the returned catalog xml and put it into a vector

        Vector remoteCatalog = cmh.getCatalogVect();

        String localcatxml = MetacatReplication.getCatalogXML();

        //System.out.println("localcatxml: " + localcatxml);

        cmh = new CatalogMessageHandler();

        catparser = initParser(cmh);

        catparser.parse(new InputSource(new StringReader(localcatxml)));

        Vector localCatalog = cmh.getCatalogVect();

        //now we have the catalog from the remote server and this local server

        //we now need to compare the two and merge the differences.

        //the comparison is base on the public_id fields which is the 4th

        //entry in each row vector.

        Vector publicId = new Vector();

        for(int i=0; i<localCatalog.size(); i++)

        {

          Vector v = new Vector((Vector)localCatalog.elementAt(i));

          //System.out.println("v1: " + v.toString());

          publicId.add(new String((String)v.elementAt(3)));

          //System.out.println("adding " + (String)v.elementAt(3));

        }

        for(int i=0; i<remoteCatalog.size(); i++)

        {

          Vector v = (Vector)remoteCatalog.elementAt(i);

          //System.out.println("v2: " + v.toString());

          //System.out.println("i: " + i);

          //System.out.println("remoteCatalog.size(): " + remoteCatalog.size());

          //System.out.println("publicID: " + publicId.toString());

          //System.out.println("v.elementAt(3): " + (String)v.elementAt(3));

          if(!publicId.contains(v.elementAt(3)))

          { //so we don't have this public id in our local table so we need to

            //add it.

            //System.out.println("in if");

            StringBuffer sql = new StringBuffer();

            sql.append("insert into xml_catalog (entry_type, source_doctype, ");

            sql.append("target_doctype, public_id, system_id) values (?,?,?,");

            sql.append("?,?)");

            //System.out.println("sql: " + sql.toString());

            PreparedStatement pstmt = conn.prepareStatement(sql.toString());

            pstmt.setString(1, (String)v.elementAt(0));

            pstmt.setString(2, (String)v.elementAt(1));

            pstmt.setString(3, (String)v.elementAt(2));

            pstmt.setString(4, (String)v.elementAt(3));

            pstmt.setString(5, (String)v.elementAt(4));

            pstmt.execute();

            pstmt.close();

          }

        }

      }

    }

    catch(Exception e)

    {

      System.out.println("error in ReplicationHandler.updateCatalog: " + 

                          e.getMessage());

      e.printStackTrace(System.out);

    }

  }

  /**

   * Method that returns true if docid has already been "deleted" from metacat.

   * This method really implements a truth table for deleted documents

   * The table is (a docid in one of the tables is represented by the X):

   * xml_docs      xml_revs      deleted?

   * ------------------------------------

   *   X             X             FALSE

   *   X             _             FALSE

   *   _             X             TRUE

   *   _             _             TRUE

   */

  private static boolean alreadyDeleted(String docid, Connection conn)

  {

    try

    {

      boolean xml_docs = false;

      boolean xml_revs = false;

      StringBuffer sb = new StringBuffer();

      sb.append("select docid from xml_revisions where docid like '");

      sb.append(docid).append("'");

      PreparedStatement pstmt = conn.prepareStatement(sb.toString());

      pstmt.execute();

      ResultSet rs = pstmt.getResultSet();

      boolean tablehasrows = rs.next();

      if(tablehasrows)

      {

        xml_revs = true;

      }

      sb = new StringBuffer();

      sb.append("select docid from xml_documents where docid like '");

      sb.append(docid).append("'");

      pstmt.close();

      pstmt = conn.prepareStatement(sb.toString());

      pstmt.execute();

      rs = pstmt.getResultSet();

      tablehasrows = rs.next();

      pstmt.close();

      if(tablehasrows)

      {

        xml_docs = true;

      }

      if(xml_docs && xml_revs)

      {

        return false;

      }

      else if(xml_docs && !xml_revs)

      {

        return false;

      }

      else if(!xml_docs && xml_revs)

      {

        return true;

      }

      else if(!xml_docs && !xml_revs)

      {

        return true;

      }

    }

    catch(Exception e)

    {

      System.out.println("error in ReplicationHandler.alreadyDeleted: " + 

                          e.getMessage());

      e.printStackTrace(System.out);

    }

    finally

    {

    }

    return false;

  }

  /**

   * Checks to see if a document is already in the DB.  Returns

   * "UPDATE" if it is, "INSERT" if it isn't

   */

  private static String getAction(String docid)

  {

    try

    {

      MetaCatUtil util = new MetaCatUtil();

      StringBuffer sql = new StringBuffer();

      sql.append("select docid from xml_documents where docid like '");

      sql.append(docid).append("'");

      Connection conn = util.openDBConnection();

      PreparedStatement pstmt = conn.prepareStatement(sql.toString());

      pstmt.execute();

      ResultSet rs = pstmt.getResultSet();

      if(rs.next())

      {

        pstmt.close();

        conn.close();

        return "UPDATE";

      }

      else

      {

        pstmt.close();

        conn.close();

        return "INSERT";

      }

    }

    catch(Exception e)

    {

      System.out.println("error in replicationHandler.getAction: " + 

                          e.getMessage());

    }

    return "";

  }

  /**

   * Method to query xml_replication and build a hashtable of each server

   * and it's last update time.

   * @param conn a connection to the database

   */

  public static Hashtable buildServerList(Connection conn)

  {

    Hashtable sl = new Hashtable();

    PreparedStatement pstmt;   

    try

    {

      pstmt = conn.prepareStatement("select server, last_checked, replicate " +

                                    "from xml_replication");

      pstmt.execute();

      ResultSet rs = pstmt.getResultSet();

      boolean tableHasRows = rs.next();

      while(tableHasRows)

      {

        if(rs.getInt(3) == 1)

        {//only put the server in the list if the replicate flag is true

          String server = rs.getString(1);

          String last_checked = rs.getString(2);

          if(!server.equals("localhost"))

          {

            sl.put(server, last_checked);

          }

        }

        tableHasRows = rs.next();   

      }

      pstmt.close();

    }

    catch(Exception e)

    {

      System.out.println("error in replicationHandler.buildServerList(): " +

                         e.getMessage());

    }

    return sl;

  }

  /**

   * Method to initialize the message parser

   */

  public static XMLReader initParser(DefaultHandler dh)

          throws Exception

  {

    XMLReader parser = null;

    try {

      ContentHandler chandler = dh;

      // Get an instance of the parser

      MetaCatUtil util = new MetaCatUtil();

      String parserName = util.getOption("saxparser");

      parser = XMLReaderFactory.createXMLReader(parserName);

      // Turn off validation

      parser.setFeature("http://xml.org/sax/features/validation", false);

      parser.setContentHandler((ContentHandler)chandler);

      parser.setErrorHandler((ErrorHandler)chandler);

    } catch (Exception e) {

      throw e;

    }

    return parser;

  }

}