Metacat

/**

 *  '$RCSfile$'

 *    Purpose: A Class that implements replication for metacat

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

 *             National Center for Ecological Analysis and Synthesis

 *    Authors: Chad Berkley

 *    Release: @release@

 *

 *   '$Author$'

 *     '$Date$'

 * '$Revision$'

 */

package edu.ucsb.nceas.metacat;

import java.util.*;

import java.io.*;

import java.sql.*;

import java.net.*;

import java.lang.*;

import java.text.*;

import javax.servlet.*;

import javax.servlet.http.*;

import oracle.xml.parser.v2.*;

import org.xml.sax.*;

public class MetacatReplication extends HttpServlet

{  

  private String deltaT;

  Timer replicationDaemon;

  MetaCatUtil util = new MetaCatUtil();

  /**

   * Initialize the servlet by creating appropriate database connections

   */

  public void init(ServletConfig config) throws ServletException 

  {

    //initialize db connections to handle any update requests

    MetaCatUtil util = new MetaCatUtil();

    deltaT = util.getOption("deltaT");

    //break off a thread to do the delta-T check

    replicationDaemon = new Timer(true);

    //replicationDaemon.scheduleAtFixedRate(new replicationHandler(), 0, 

    //                                      new Integer(deltaT).intValue() * 1000);

    //System.out.println("timer scheduled at: " + new Integer(deltaT).intValue() 

    //                   + " seconds");

  }

  public void destroy() 

  {

    replicationDaemon.cancel();

    System.out.println("Replication daemon cancelled.");

  }

  public void doGet (HttpServletRequest request, HttpServletResponse response)

                     throws ServletException, IOException 

  {

    // Process the data and send back the response

    handleGetOrPost(request, response);

  }

  public void doPost(HttpServletRequest request, HttpServletResponse response)

                     throws ServletException, IOException 

  {

    // Process the data and send back the response

    handleGetOrPost(request, response);

  }

  private void handleGetOrPost(HttpServletRequest request, 

                               HttpServletResponse response) 

                               throws ServletException, IOException 

  {

    PrintWriter out = response.getWriter();

    Hashtable params = new Hashtable();

    Enumeration paramlist = request.getParameterNames();

    while (paramlist.hasMoreElements()) 

    {

      String name = (String)paramlist.nextElement();

      String[] value = request.getParameterValues(name);

      params.put(name, value);  

    }

    if(params.containsKey("stop"))

    {

      if(((String[])params.get("stop"))[0].equals("true"))

      {  

        replicationDaemon.cancel();

        out.println("Replication Handler Stopped");

      }

    }

    if(params.containsKey("start"))

    {

      if(((String[])params.get("start"))[0].equals("true"))

      {

        int rate;

        if(params.containsKey("rate"))

          rate = new Integer(

                 new String(((String[])params.get("rate"))[0])).intValue();

        else

          rate = 1000;

        out.println("New rate is: " + rate + " seconds.");

        replicationDaemon.cancel();

        replicationDaemon = new Timer(true);

        replicationDaemon.scheduleAtFixedRate(new replicationHandler(out), 0, 

                                              rate * 1000);

        out.println("Replication Handler Started");

      }

    }

    if(params.containsKey("update"))

    {

      handleUpdateRequest(out, params, response);

    }

  }

  private void handleUpdateRequest(PrintWriter out, Hashtable params, 

                                   HttpServletResponse response)

  {

    System.out.println("incoming request for dt/time " + 

                       ((String[])params.get("update"))[0] +

                       " from external metacat");

    response.setContentType("text/xml");

    out.println("<replication><textmessage>In metacatReplication</textmessage>");

    StringBuffer sql = new StringBuffer();

    StringBuffer returnXML = new StringBuffer();

    String updateStr = ((String[])params.get("update"))[0];

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

    //pseudo algorithm:

    ///////////////////////////////////////////////////////////////////////

    //get the date/time from the requestor, query the db for any documents

    //that have an update date later than the requested date and send

    //those docids back to the requestor.  If there are no newer documents

    //then send back an up-to-date message.

    ///////////////////////////////////////////////////////////////////////

    //Timestamp update = Timestamp.valueOf(updateStr);

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

    java.util.Date update = new java.util.Date();

    ParsePosition pos = new ParsePosition(0);

    update = formatter.parse(updateStr, pos);

    String dateString = formatter.format(update);

    sql.append("select docid, date_updated from xml_documents where ");

    sql.append("date_updated > ");

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

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

    try

    {

      Connection conn = util.openDBConnection();

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

      pstmt.execute();

      ResultSet rs = pstmt.getResultSet();

      boolean tablehasrows = rs.next();

      returnXML.append("<server>").append(util.getOption("server"));

      returnXML.append("</server><updates>");

      while(tablehasrows)

      {

        returnXML.append("<updatedDocument><docid>").append(rs.getString(1));

        returnXML.append("</docid><date_updated>").append(rs.getString(2));

        returnXML.append("</date_updated></updatedDocument>");

        tablehasrows = rs.next();

      }

      returnXML.append("</updates></replication>");

      out.print(returnXML.toString());

    }

    catch(Exception e)

    {

      System.out.println("Exception in metacatReplication: " + e.getMessage());

    }

  }

}

/**

 *  '$RCSfile$'

 *    Purpose: A class that handles xml messages  passed by  the replication handler

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

 *             National Center for Ecological Analysis and Synthesis

 *    Authors: Chad Berkley

 *    Release: @release@

 *

 *   '$Author$'

 *     '$Date$'

 * '$Revision$'

 */

package edu.ucsb.nceas.metacat;

import java.sql.*;

import java.util.Stack;

import java.util.Vector;

import java.util.Enumeration;

import java.util.EmptyStackException;

import org.xml.sax.Attributes;

import org.xml.sax.SAXException;

import org.xml.sax.SAXParseException;

import org.xml.sax.ext.DeclHandler;

import org.xml.sax.ext.LexicalHandler;

import org.xml.sax.helpers.DefaultHandler;

/** 

 * A database aware Class implementing callback bethods for the SAX parser to

 * call when processing the XML messages from the replication handler

 */

public class ReplMessageHandler extends DefaultHandler 

{

  private Vector updates = new Vector();

  private Vector indivUpdate = new Vector();

  String currentTag = new String();

  public ReplMessageHandler()

  {

  }

  /**

   * This method starts a new vector for each updatedDocument tag.

   */

  public void startElement(String uri, String localName, String qName, 

                           Attributes attributes) throws SAXException

  {

    currentTag = localName;

    if(localName.equals("updatedDocument"))

    {

      indivUpdate = new Vector();

    }

  }

  /**

   * This method write the indivUpdate to updates when it finds the end of

   */

  public void endElement(String uri, String localName, String qName) 

              throws SAXException

  {

    if(localName.equals("updatedDocument"))

    {

      updates.add(new Vector(indivUpdate));

    }

  }

  /**

   * Take the data out of the docid and date_updated fields

   */

  public void characters(char[] ch, int start, int length) throws SAXException

  {

    if(currentTag.equals("docid"))

    {

      indivUpdate.add(new String(ch, start, length));

    }

    if(currentTag.equals("date_updated"))

    {

      indivUpdate.add(new String(ch, start, length));

    }

  }

  public Vector getResultVect()

  {

    return updates;

  }

}

/**

 *  '$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$'

 *     '$Date$'

 * '$Revision$'

 */

package edu.ucsb.nceas.metacat;

import java.sql.*;

import java.util.*;

import java.lang.Thread; 

import java.io.*;

import java.net.*;

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;

/**

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

  {

    System.out.println("replicationHandler is running");

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

      System.out.println("Server list: " + serverList.toString());

      update(serverList, conn);

    }

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

    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(' ', '+');

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

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

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

        {

          istreamChar = (char)istreamInt;

          serverResponse.append(istreamChar);

        }

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

      }

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

      //initialize the parser

      XMLReader parser = initParser(message);

      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.getResultVect());

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

        {

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

          System.out.print("param " + j + ": " + w.toString());

          //so now we have a list of the documents that need to be updated, so 

          //now we need to request them. from the server and update them here  

        }

        System.out.println("");

      }

    }

    catch(Exception e)

    {

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

                         e.getMessage());

    }

  }

  /**

   * Method to initialize the message parser

   */

  private static XMLReader initParser(ReplMessageHandler rmh)

          throws Exception

  {

    XMLReader parser = null;

    try {

      ContentHandler chandler = rmh;

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

      {

        sl.put(rs.getString(1), rs.getString(2));

        tableHasRows = rs.next();

      }

    }

    catch(Exception e)

    {

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

                         e.getMessage());

    }

    return sl;

  }

}