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

 *     '$Date: 2000-12-01 15:27:58 -0800 (Fri, 01 Dec 2000) $'

 * '$Revision: 577 $'

 */

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 org.xml.sax.*;

public class MetacatReplication extends HttpServlet implements Runnable

{  

  private String deltaT;

  Timer replicationDaemon;

  private static MetaCatUtil util = new MetaCatUtil();

  private Vector fileLocks = new Vector();

  private Thread lockThread = null;

  /**

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

    {

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

      { //stop the replication server

        replicationDaemon.cancel();

        out.println("Replication Handler Stopped");

        System.out.println("Replication Handler Stopped");

      }

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

      { //start the replication server

        int rate;

        if(params.containsKey("rate"))

        {

          rate = new Integer(

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

          if(rate < 30)

          {

            out.println("Replication deltaT rate cannot be less than 30!");

            rate = 1000;

          }

        }

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

        System.out.println("Replication Handler Started");

      }

      else if(((String[])params.get("action"))[0].equals("forcereplicate"))

      {

        handleForceReplicateRequest(out, params, response);

      }

      else if(((String[])params.get("action"))[0].equals("update"))

      { //request an update list from the server

        if(params.contains("servercheckcode"))

        { //the servercheckcode allows this server to check for updated 

          //file from other servers as well as just updated files from 

          //this server.

          System.out.println("metacatreplication: servercheckcode: " + 

                               ((int[])params.get("servercheckcode"))[0]);

          handleUpdateRequest(out, params, response, 

                              ((int[])params.get("servercheckcode"))[0]);

        }

        else

        {

          handleUpdateRequest2(out, params, response);

        }

      }

      else if(((String[])params.get("action"))[0].equals("read"))

      { //request a specific document from the server

        //note that this could be replaced by a call to metacatServlet

        //handleGetDocumentAction().

        handleGetDocumentRequest(out, params, response);

      }

      else if(((String[])params.get("action"))[0].equals("getlock"))

      {

        handleGetLockRequest(out, params, response);

      }

      else if(((String[])params.get("action"))[0].equals("getdocumentinfo"))

      {

        handleGetDocumentInfoRequest(out, params, response);

      }

      else if(((String[])params.get("action"))[0].equals("gettime"))

      {

        handleGetTimeRequest(out, params, response);

      }

    }

  }

  private void handleForceReplicateRequest(PrintWriter out, Hashtable params,

                                           HttpServletResponse response)

  {

    System.out.println("in handleforcereplicaterequest");

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

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

    String dbaction = "UPDATE";

    boolean override = false;

    int serverCode = 1;

    try

    {

      if(params.containsKey("dbaction"))

      {

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

        serverCode = MetacatReplication.getServerCode(server);

        override = true;

      }

      System.out.println("action in forcereplicate is: " + dbaction);

      System.out.println("serverCode in forcereplicate is: " + serverCode);

      int serverCheckCode = MetacatReplication.getServerCode(server);

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

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

      String xmldoc = MetacatReplication.getURLContent(u);

      URL docinfourl = new URL("http://" + server + 

                               "?action=getdocumentinfo&docid=" +

                               docid);

      System.out.println("sending message: " + docinfourl.toString());

      String docInfoStr = MetacatReplication.getURLContent(docinfourl);

      DocInfoHandler dih = new DocInfoHandler();

      XMLReader docinfoParser = ReplicationHandler.initParser(dih);

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

      Hashtable docinfoHash = dih.getDocInfo();

      String user = (String)docinfoHash.get("user_owner");

      String group = new String(user);

      Connection conn = util.openDBConnection();

      DocumentImpl.write(conn, new StringReader(xmldoc), null, dbaction, docid, 

                         user, group, serverCode, override);

      conn.close();

    }

    catch(Exception e)

    {

      System.out.println("error in metacatReplication.handleForceReplicate" +

                         "Request: " + e.getMessage());

    }

  }

  /**

   * Grants or denies a lock to a requesting host.

   * The servlet parameters of interrest are:

   * docid: the docid of the file the lock is being requested for

   * currentdate: the timestamp of the document on the remote server

   * 

   */

  private void handleGetLockRequest(PrintWriter out, Hashtable params,

                                    HttpServletResponse response)

  {

    System.out.println("in handlegetlockrequest");

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

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

    try

    {

      Connection conn = util.openDBConnection();

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

      String remoteDateStr = ((String[])params.get("updatedate"))[0];

      DocumentImpl requestDoc = new DocumentImpl(conn, docid);

      String localDateStr = requestDoc.getUpdateDate();

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

      ParsePosition pos = new ParsePosition(0);

      remoteDate = formatter.parse(remoteDateStr, pos);

      pos = new ParsePosition(0);

      localDate = formatter.parse(localDateStr, pos);

      if(remoteDate.compareTo(localDate) >= 0)

      {

        if(!fileLocks.contains(docid))

        { //grant the lock

          fileLocks.add(0, docid); //insert at the beginning of the queue Vector

          //send a message back to the the remote host authorizing the insert

          out.println("<lockgranted><docid>" +docid+ "</docid></lockgranted>");

          lockThread = new Thread(this);

          lockThread.setPriority(Thread.MIN_PRIORITY);

          lockThread.start();

        }

        else

        { //deny the lock

          out.println("<filelocked><docid>" + docid + "</docid></filelocked>");

        }

      }

      else

      {//deny the lock.

        out.println("<outdatedfile><docid>" + docid + "</docid></filelocked>");

      }

      conn.close();

    }

    catch(Exception e)

    {

      System.out.println("error requesting file lock: " + e.getMessage());

      e.printStackTrace(System.out);

    }

  }

  /**

   * Sends all of the xml_documents information encoded in xml to a requestor

   */

  private void handleGetDocumentInfoRequest(PrintWriter out, Hashtable params, 

                                        HttpServletResponse response)

  {

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

    StringBuffer sb = new StringBuffer();

    try

    {

      Connection conn = util.openDBConnection();

      DocumentImpl doc = new DocumentImpl(conn, docid);

      sb.append("<documentinfo><docid>").append(docid);

      sb.append("</docid><docname>").append(doc.getDocname());

      sb.append("</docname><doctype>").append(doc.getDoctype());

      sb.append("</doctype><doctitle>").append(doc.getDocTitle());

      sb.append("</doctitle><user_owner>").append(doc.getUserowner());

      sb.append("</user_owner><user_updated>").append(doc.getUserupdated());

      sb.append("</user_updated><public_access>").append(doc.getPublicaccess());

      sb.append("</public_access></documentinfo>");

      response.setContentType("text/xml");

      out.println(sb.toString());

      conn.close();

    }

    catch (Exception e)

    {

      System.out.println("error in metacatReplication.handlegetdocumentinforequest: " + 

      e.getMessage());

    }

  }

  /**

   * Sends a document to a remote host

   */

  private void handleGetDocumentRequest(PrintWriter out, Hashtable params, 

                                        HttpServletResponse response)

  {

    try

    {

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

      System.out.println("incoming get request for document: " +

                         docid);

      Connection conn = util.openDBConnection();

      DocumentImpl di = new DocumentImpl(conn, docid);

      response.setContentType("text/xml");

      out.print(di.toString());

      conn.close();

    }

    catch(Exception e)

    {

      System.out.println("error getting document: " + e.getMessage());

    }

  }

  /**

   * Does default handling for update requests

   */

  private void handleUpdateRequest(PrintWriter out, Hashtable params, 

                                   HttpServletResponse response)

  {

    handleUpdateRequest(out, params, response, 1);

  }

  /**

   * Initiates an update of all server in the xml_replication table.

   * the remote host sends an update date.  The local host (this method)

   * queries the db for any document that was updated after the update date

   * and returns a list of those documents to the remote host.  It also 

   * sends back a list of the files that were locally deleted.

   * serverCheckCode allows the requestor to request documents from this 

   * server that reside on a different server.  Normally, this server

   * should only replicate files that it owns but in the case of updating 

   * a file that does not belong to this server, it is needed.  See 

   * DocumentImpl.write().

   */

  private void handleUpdateRequest(PrintWriter out, Hashtable params, 

                                   HttpServletResponse response, 

                                   int serverCheckCode)

  { 

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

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

                       " from external metacat");

    response.setContentType("text/xml");

    StringBuffer returnXML = new StringBuffer();

    returnXML.append("<?xml version=\"1.0\"?><replication>");

    StringBuffer sql = new StringBuffer();

    String updateStr = ((String[])params.get("date"))[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 a null 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, server_location from ");

    sql.append("xml_documents where date_updated > ");

    sql.append("to_date('").append(dateString);

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

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

    //get any recently deleted documents

    StringBuffer delsql = new StringBuffer();

    delsql.append("select docid, date_updated, server_location from ");

    delsql.append("xml_revisions where docid not in (select docid from ");

    delsql.append("xml_documents) and date_updated > to_date('");

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

    try

    {

      Connection conn = util.openDBConnection();

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

      pstmt.execute();

      ResultSet rs = pstmt.getResultSet();

      boolean tablehasrows = rs.next();

      //if a '1' should not represent localhost, add code here to query

      //xml_replication for the proper serverid number

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

      returnXML.append(util.getOption("replicationpath"));

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

      while(tablehasrows)

      {

        String docid = rs.getString(1);

        String dateUpdated = rs.getString(2);

        int serverCode = rs.getInt(3);

        if(serverCode == serverCheckCode)

        { //check that this document is from this server.

          //servers only replicate their own documents!

          returnXML.append("<updatedDocument><docid>").append(docid);

          returnXML.append("</docid><date_updated>").append(dateUpdated);

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

        }

        tablehasrows = rs.next();

      }

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

      pstmt.execute();

      rs = pstmt.getResultSet();

      tablehasrows = rs.next();

      while(tablehasrows)

      { //handle the deleted documents

        String docid = rs.getString(1);

        String dateUpdated = rs.getString(2);

        int serverCode = rs.getInt(3);

        if(serverCode == 1)

        {

          returnXML.append("<deletedDocument><docid>").append(docid);

          returnXML.append("</docid><date_updated>").append(dateUpdated);

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

        }

        tablehasrows = rs.next();

      }

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

      conn.close();

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

      out.print(returnXML.toString());

    }

    catch(Exception e)

    {

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

    }

  }

  /**

   * Sends an update list based on rev numbers instead of dates.

   */

  private void handleUpdateRequest2(PrintWriter out, Hashtable params, 

                                    HttpServletResponse response)

  {

    System.out.println("in handleUpdateRequest2");

    try

    {

      System.out.println("received update request");

      StringBuffer docsql = new StringBuffer();

      StringBuffer doclist = new StringBuffer();

      //get all docs that reside on this server

      doclist.append("<?xml version=\"1.0\"?><replication>");

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

      doclist.append(util.getOption("replicationpath"));

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

      docsql.append("select docid, rev from xml_documents where "); 

      docsql.append("server_location = 1");

      //get any deleted documents

      StringBuffer delsql = new StringBuffer();

      delsql.append("select distinct docid from ");

      delsql.append("xml_revisions where docid not in (select docid from ");

      delsql.append("xml_documents) and server_location = 1");

      Connection conn = util.openDBConnection();

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

      pstmt.execute();

      ResultSet rs = pstmt.getResultSet();

      boolean tablehasrows = rs.next();

      while(tablehasrows)

      {

        doclist.append("<updatedDocument>");

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

        doclist.append("</docid><rev>").append(rs.getInt(2));

        doclist.append("</rev>");

        doclist.append("</updatedDocument>");

        tablehasrows = rs.next();

      }

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

      pstmt.execute();

      rs = pstmt.getResultSet();

      tablehasrows = rs.next();

      while(tablehasrows)

      { //handle the deleted documents

        doclist.append("<deletedDocument><docid>").append(rs.getString(1));

        doclist.append("</docid><rev></rev></deletedDocument>");

        tablehasrows = rs.next();

      }

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

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

      conn.close();

      response.setContentType("text/xml");

      out.println(doclist.toString());

    }

    catch(Exception e)

    {

      System.out.println("error in handleupdaterequest2: " + e.getMessage());

      e.printStackTrace(System.out);

    }

  }

  /**

   * Sends the current system date to the remote server.  Using this action

   * for replication gets rid of any problems with syncronizing clocks 

   * because a time specific to a document is always kept on its home server.

   */

  private void handleGetTimeRequest(PrintWriter out, Hashtable params, 

                                    HttpServletResponse response)

  {

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

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

    String dateString = formatter.format(localtime);

    response.setContentType("text/xml");

    out.println("<timestamp>" + dateString + "</timestamp>");

  }

  /**

   * This method is what is run when a seperate thread is broken off to handle

   * inserting a document from a remote replicated server.

   */

  public void run()

  {

    try

    {

      System.out.println("thread started for docid: " + 

                         (String)fileLocks.elementAt(0));

      Thread.sleep(30000); //the lock will expire in 30 seconds

      System.out.println("thread for docid: " + 

                         (String)fileLocks.elementAt(fileLocks.size() - 1) + 

                         " exiting.");

      fileLocks.remove(fileLocks.size() - 1);

      //fileLocks is treated as a FIFO queue.  If there are more than one lock

      //in the vector, the first one inserted will be removed.

    }

    catch(Exception e)

    {

      System.out.println("error in file lock thread: " + e.getMessage());

    }

  }

  /**

   * Returns the name of a server given a serverCode

   * @param serverCode the serverid of the server

   * @return the servername or null if the specified serverCode does not

   *         exist.

   */

  public static String getServer(int serverCode)

  {

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

    try

    {

      Connection conn = util.openDBConnection();

      String sql = new String("select server from " +

                              "xml_replication where serverid = " + 

                              serverCode);

      PreparedStatement pstmt = conn.prepareStatement(sql);

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

      pstmt.execute();

      ResultSet rs = pstmt.getResultSet();

      boolean tablehasrows = rs.next();

      if(tablehasrows)

      {

        //System.out.println("server: " + rs.getString(1));

        return rs.getString(1);

      }

      conn.close();

    }

    catch(Exception e)

    {

      System.out.println("Error in MetacatReplication.getServer: " + 

                          e.getMessage());

    }

    return null;

      //return null if the server does not exist

  }

  /**

   * Returns a server code given a server name

   * @param server the name of the server

   * @return integer > 0 representing the code of the server, 0 if the server

   *  does not exist.

   */

  public static int getServerCode(String server) throws Exception

  {

    try

    {

      Connection conn = util.openDBConnection();

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

                                         "xml_replication where server " +

                                         "like '" + server + "'");

      pstmt.execute();

      ResultSet rs = pstmt.getResultSet();

      boolean tablehasrows = rs.next();

      int serverCode = 0;

      if(tablehasrows)

      {

         return rs.getInt(1);

      }

      else

      {

        return 0;

      }

    }

    catch(Exception e)

    {

      throw e;

    }

  }

  /**

   * This method returns the content of a url

   * @param u the url to return the content from

   * @return a string representing the content of the url

   * @throws java.io.IOException

   */

  public static String getURLContent(URL u) throws java.io.IOException

  {

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

    char istreamChar;

    int istreamInt;

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

    StringBuffer serverResponse = new StringBuffer();

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

    {

      istreamChar = (char)istreamInt;

      serverResponse.append(istreamChar);

    }

    return serverResponse.toString();

  }

}