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: jones $'

 *     '$Date: 2000-11-27 13:55:08 -0800 (Mon, 27 Nov 2000) $'

 * '$Revision: 566 $'

 */

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

{

  MetaCatUtil util = new MetaCatUtil();

  Hashtable serverList = new Hashtable(); 

  Connection conn;

  PrintWriter out;

  public ReplicationHandler(PrintWriter o)

  {

    this.out = o;

  }

  /**

   * 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

    {

      conn = util.openDBConnection();

      serverList = buildServerList(conn);

      update(serverList, conn);

      conn.close();

    }

    catch (Exception e)

    {

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

    }

  }

  /**

   * Method to check each server in the servetList for updates

   */

  private void update(Hashtable serverList, Connection conn)

  {

    /*

     Pseudo-algorithm:

     -build a list of servers that need to be checked

     -check each server to see if any documents were modified after the 

      server's update_date

     -if there are documents that need to be updated pull them here

      and update the one currently in the database or add a new one

      if it is not already here.

     -update each servers' update_date to the current time

    */

    PreparedStatement pstmt;

    Enumeration keys;

    int istreamInt;

    char istreamChar;

    StringBuffer serverResponse = new StringBuffer();

    String server;

    String update;

    Vector responses = new Vector();

    ReplMessageHandler message = new ReplMessageHandler();

    Hashtable updateDocs = new Hashtable();

    URL u;

    InputStreamReader istream;

    try

    {

      //build a list of servers with updated documents.  Choose the newest

      //one out of the list, update this server, update last_checked

      keys = serverList.keys();

      while(keys.hasMoreElements())

      { //update from one server at a time

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

        update = (String)(serverList.get(server));

        //send the server a date and it will send back any docid that has 

        //been modified after that date

        update = update.replace(' ', '+'); 

        u = new URL("http://" + server + "?action=update&date=" + update);

        istream = new InputStreamReader(u.openStream());

        serverResponse = new StringBuffer();

        while((istreamInt = istream.read()) != -1)

        {

          istreamChar = (char)istreamInt;

          serverResponse.append(istreamChar);

        }

        responses.add(serverResponse.toString()); //list of updates

      }

      //initialize the parser

      XMLReader parser = initParser(message);

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

      //System.out.println("<--end response-->");

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

      { //parse the xml and get the result

        parser.parse(new InputSource(

                     new StringReader(

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

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

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

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

        { //go through each update vector and update or insert

          //a new document

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

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

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

          //send a message to the server requesting each document

          URL getDocURL = new URL("http://" + docServer + 

                                  "?action=read&docid="+ docid);

          InputStreamReader getDocIstream = new InputStreamReader(

                                                getDocURL.openStream()); 

          //the following while loop should not be needed.  see the note

          //below before the DocumentImpl.write() call.

          serverResponse = new StringBuffer();

          while((istreamInt = getDocIstream.read()) != -1)

          {

            istreamChar = (char)istreamInt;

            serverResponse.append(istreamChar);

          }

          //System.out.println("<<<<<<document>>>>>");

          //System.out.println(serverResponse.toString());

          //System.out.println("<<<<<<end document>>>>>");

          pstmt = conn.prepareStatement("select serverid from " +

                                         "xml_replication where server " +

                                         "like '" + docServer + "'");

          pstmt.execute();

          ResultSet rs = pstmt.getResultSet();

          boolean tablehasrows = rs.next();

          int serverCode = 0;

          if(tablehasrows)

          {

            serverCode = rs.getInt(1);

            //System.out.println("servercode: " + serverCode);

          }

          else

          {

            System.out.println("error: server not registered");

          }

          if(serverCode != 0)

          {

            //update the document into the DB

            String action = getAction(docid);

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

            //note that getDocIstream is commented out below.  This should

            //work as a param to this method but it doesn'.  I don't know why

            //but putting a string reader there works but not an 

            //inputStreamReader.

            DocInfoHandler dih = new DocInfoHandler();

            XMLReader docinfoParser = initParser(dih);

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

                                     "?action=getdocumentinfo&docid=" +

                                     docid);

            InputStreamReader isr = new InputStreamReader(

                                        docinfoUrl.openStream());

            StringBuffer docInfoBuffer = new StringBuffer();

            while((istreamInt = isr.read()) != 1)

            {

              istreamChar = (char)istreamInt;

              docInfoBuffer.append(istreamChar);

            }

            docinfoParser.parse(new InputSource(new StringReader(

                                                docInfoBuffer.toString())));

            Hashtable docinfoHash = dih.getDocInfo();

            String newDocid = DocumentImpl.write(conn, 

                               new StringReader(serverResponse.toString())

                               /*getDocIstream*/, 

                               action, 

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

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

                               serverCode);

            System.out.println("newDocid: " + newDocid + " " + action + "ED");

          }

        }

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

        {

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

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

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

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

        }

      }

      //update the last_update field for each server to the current date/time      

      SimpleDateFormat formatter = new SimpleDateFormat ("yy-MM-dd HH:mm:ss");

      java.util.Date newDate = new java.util.Date(System.currentTimeMillis());

      ParsePosition pos = new ParsePosition(0);

      String dateString = formatter.format(newDate);

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

      StringBuffer sql = new StringBuffer();

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

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

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

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

      pstmt.executeUpdate();

      //conn.commit();

      System.out.println("last_checked updated: " + dateString);

    }

    catch(Exception e)

    {

      System.out.println("Error in replicationHandler.update(): " + 

                         e.getMessage());

      e.printStackTrace(System.out);

    }

  }

  /**

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

      {

        conn.close();

        return "UPDATE";

      }

      else

      {

        conn.close();

        return "INSERT";

      }

    }

    catch(Exception e)

    {

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

                          e.getMessage());

    }

    return "";

  }

  /**

   * Method to initialize the message parser

   */

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

  }

  /**

   * 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

   */

  private Hashtable buildServerList(Connection conn)

  {

    Hashtable sl = new Hashtable();

    PreparedStatement pstmt;   

    try

    {

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

                                    "xml_replication");

      pstmt.execute();

      ResultSet rs = pstmt.getResultSet();

      boolean tableHasRows = rs.next();

      while(tableHasRows)

      {

        String server = rs.getString(1);

        String last_checked = rs.getString(2);

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

        {

          sl.put(server, last_checked);

        }

        tableHasRows = rs.next();

      }

    }

    catch(Exception e)

    {

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

                         e.getMessage());

    }

    return sl;

  }

}