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

 *

 *   '$Author: leinfelder $'

 *     '$Date: 2011-05-05 15:14:07 -0700 (Thu, 05 May 2011) $'

 * '$Revision: 6079 $'

 *

 * 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.replication;

import java.io.BufferedInputStream;

import java.io.BufferedOutputStream;

import java.io.File;

import java.io.FileInputStream;

import java.io.FileNotFoundException;

import java.io.FileOutputStream;

import java.io.IOException;

import java.io.InputStream;

import java.io.InputStreamReader;

import java.io.OutputStream;

import java.io.StringReader;

import java.io.Writer;

import java.net.MalformedURLException;

import java.net.URL;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.SQLException;

import java.text.SimpleDateFormat;

import java.util.Date;

import java.util.Enumeration;

import java.util.Hashtable;

import java.util.Timer;

import java.util.Vector;

import javax.servlet.http.HttpServletRequest;

import javax.servlet.http.HttpServletResponse;

import org.apache.log4j.Logger;

import org.xml.sax.InputSource;

import org.xml.sax.SAXException;

import org.xml.sax.XMLReader;

import edu.ucsb.nceas.metacat.DocInfoHandler;

import edu.ucsb.nceas.metacat.DocumentImpl;

import edu.ucsb.nceas.metacat.DocumentImplWrapper;

import edu.ucsb.nceas.metacat.EventLog;

import edu.ucsb.nceas.metacat.IdentifierManager;

import edu.ucsb.nceas.metacat.McdbDocNotFoundException;

import edu.ucsb.nceas.metacat.McdbException;

import edu.ucsb.nceas.metacat.accesscontrol.AccessControlException;

import edu.ucsb.nceas.metacat.accesscontrol.AccessControlForSingleFile;

import edu.ucsb.nceas.metacat.accesscontrol.PermOrderException;

import edu.ucsb.nceas.metacat.accesscontrol.XMLAccessDAO;

import edu.ucsb.nceas.metacat.client.InsufficientKarmaException;

import edu.ucsb.nceas.metacat.database.DBConnection;

import edu.ucsb.nceas.metacat.database.DBConnectionPool;

import edu.ucsb.nceas.metacat.database.DatabaseService;

import edu.ucsb.nceas.metacat.properties.PropertyService;

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

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

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

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

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

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

import edu.ucsb.nceas.utilities.FileUtil;

import edu.ucsb.nceas.utilities.GeneralPropertyException;

import edu.ucsb.nceas.utilities.PropertyNotFoundException;

public class ReplicationService extends BaseService {

	private static ReplicationService replicationService = null;

	private long timeInterval;

	private Date firstTime;

	private boolean timedReplicationIsOn = false;

	Timer replicationDaemon;

	private static Vector<String> fileLocks = new Vector<String>();

//	private Thread lockThread = null;

	public static final String FORCEREPLICATEDELETE = "forcereplicatedelete";

	private static String TIMEREPLICATION = "replication.timedreplication";

	private static String TIMEREPLICATIONINTERVAl ="replication.timedreplicationinterval";

	private static String FIRSTTIME = "replication.firsttimedreplication";

	private static final int TIMEINTERVALLIMIT = 7200000;

	public static final String REPLICATIONUSER = "replication";

	public static final String REPLICATION_LOG_FILE_NAME = "metacatreplication.log";

	public static String METACAT_REPL_ERROR_MSG = null;

	private static Logger logReplication = Logger.getLogger("ReplicationLogging");

	private static Logger logMetacat = Logger.getLogger(ReplicationService.class);

	private ReplicationService() throws ServiceException {

		_serviceName = "ReplicationService";

		initialize();

	}

	private void initialize() throws ServiceException {

		// initialize db connections to handle any update requests

		// deltaT = util.getProperty("replication.deltaT");

		// the default deltaT can be set from metacat.properties

		// create a thread to do the delta-T check but don't execute it yet

		replicationDaemon = new Timer(true);

		try {

			String replLogFile = PropertyService.getProperty("replication.logdir")

				+ FileUtil.getFS() + REPLICATION_LOG_FILE_NAME;

			METACAT_REPL_ERROR_MSG = "An error occurred in replication.  Please see the " +

				"replication log at: " + replLogFile;

			String timedRepIsOnStr = 

				PropertyService.getProperty("replication.timedreplication");

			timedReplicationIsOn = (new Boolean(timedRepIsOnStr)).booleanValue();

			logReplication.info("ReplicationService.initialize - The timed replication on is" + timedReplicationIsOn);

			String timeIntervalStr = 

				PropertyService.getProperty("replication.timedreplicationinterval");

			timeInterval = (new Long(timeIntervalStr)).longValue();

			logReplication.info("ReplicationService.initialize - The timed replication time Interval is " + timeInterval);

			String firstTimeStr = 

				PropertyService.getProperty("replication.firsttimedreplication");

			logReplication.info("ReplicationService.initialize - first replication time form property is " + firstTimeStr);

			firstTime = ReplicationHandler.combinateCurrentDateAndGivenTime(firstTimeStr);

			logReplication.info("ReplicationService.initialize - After combine current time, the real first time is "

					+ firstTime.toString() + " minisec");

			// set up time replication if it is on

			if (timedReplicationIsOn) {

				replicationDaemon.scheduleAtFixedRate(new ReplicationHandler(),

						firstTime, timeInterval);

				logReplication.info("ReplicationService.initialize - deltaT handler started with rate="

						+ timeInterval + " mini seconds at " + firstTime.toString());

			}

		} catch (PropertyNotFoundException pnfe) {

			throw new ServiceException(

					"ReplicationService.initialize - Property error while instantiating "

							+ "replication service: " + pnfe.getMessage());

		} catch (HandlerException he) {

			throw new ServiceException(

					"ReplicationService.initialize - Handler error while instantiating "

							+ "replication service" + he.getMessage());

		} 

	}

	/**

	 * Get the single instance of SessionService.

	 * 

	 * @return the single instance of SessionService

	 */

	public static ReplicationService getInstance() throws ServiceException {

		if (replicationService == null) {

			replicationService = new ReplicationService();

		}

		return replicationService;

	}

	public boolean refreshable() {

		return true;

	}

	protected void doRefresh() throws ServiceException {

		return;

	}

	public void stop() throws ServiceException{

	}

	public void stopReplication() throws ServiceException {

	      //stop the replication server

	      replicationDaemon.cancel();

	      replicationDaemon = new Timer(true);

	      timedReplicationIsOn = false;

	      try {

	    	  PropertyService.setProperty("replication.timedreplication", (new Boolean(timedReplicationIsOn)).toString());

	      } catch (GeneralPropertyException gpe) {

	    	  logReplication.warn("ReplicationService.stopReplication - Could not set replication.timedreplication property: " + gpe.getMessage());

	      }

	      logReplication.info("ReplicationService.stopReplication - deltaT handler stopped");

		return;

	}

	protected void startReplication(Hashtable<String, String[]> params) throws ServiceException {

	       String firstTimeStr = "";

	      //start the replication server

	       if ( params.containsKey("rate") ) {

	        timeInterval = new Long(

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

	        if(timeInterval < TIMEINTERVALLIMIT) {

	            //deltaT<30 is a timing mess!

	            timeInterval = TIMEINTERVALLIMIT;

	            throw new ServiceException("Replication deltaT rate cannot be less than "+

	                    TIMEINTERVALLIMIT + " millisecs and system automatically setup the rate to "+TIMEINTERVALLIMIT);

	        }

	      } else {

	        timeInterval = TIMEINTERVALLIMIT ;

	      }

	      logReplication.info("ReplicationService.startReplication - New rate is: " + timeInterval + " mini seconds.");

	      if ( params.containsKey("firsttime"))

	      {

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

	         try

	         {

	           firstTime = ReplicationHandler.combinateCurrentDateAndGivenTime(firstTimeStr);

	           logReplication.info("ReplicationService.startReplication - The first time setting is "+firstTime.toString());

	         }

	         catch (HandlerException e)

	         {

	            throw new ServiceException(e.getMessage());

	         }

	         logReplication.warn("After combine current time, the real first time is "

	                                  +firstTime.toString()+" minisec");

	      }

	      else

	      {

	    	  logMetacat.error("ReplicationService.startReplication - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

	          logReplication.error("ReplicationService.startReplication - You should specify the first time " +

	                                  "to start a time replication");

	          return;

	      }

	      timedReplicationIsOn = true;

	      try {

	      // save settings to property file

	      PropertyService.setProperty(TIMEREPLICATION, (new Boolean(timedReplicationIsOn)).toString());

	      // note we couldn't use firstTime object because it has date info

	      // we only need time info such as 10:00 PM

	      PropertyService.setProperty(FIRSTTIME, firstTimeStr);

	      PropertyService.setProperty(TIMEREPLICATIONINTERVAl, (new Long(timeInterval)).toString());

	      } catch (GeneralPropertyException gpe) {

	    	  logReplication.warn("ReplicationService.startReplication - Could not set property: " + gpe.getMessage());

	      }

	      replicationDaemon.cancel();

	      replicationDaemon = new Timer(true);

	      replicationDaemon.scheduleAtFixedRate(new ReplicationHandler(), firstTime,

	                                            timeInterval);

	      logReplication.info("ReplicationService.startReplication - deltaT handler started with rate=" +

	                                    timeInterval + " milliseconds at " +firstTime.toString());

	}

	public void runOnce() throws ServiceException {

	      //updates this server exactly once

	      replicationDaemon.schedule(new ReplicationHandler(), 0);

	}

	/**

	 * This method can add, delete and list the servers currently included in

	 * xml_replication.

	 * action           subaction            other needed params

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

	 * servercontrol    add                  server

	 * servercontrol    delete               server

	 * servercontrol    list

	 */

	protected static void handleServerControlRequest(

			Hashtable<String, String[]> params, HttpServletResponse response) {

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

		DBConnection dbConn = null;

		int serialNumber = -1;

		PreparedStatement pstmt = null;

		String replicate = null;

		String server = null;

		String dataReplicate = null;

		String hub = null;

		Writer out = null;

		try {

			response.setContentType("text/xml");

			out = response.getWriter();

			//conn = util.openDBConnection();

			dbConn = DBConnectionPool

					.getDBConnection("MetacatReplication.handleServerControlRequest");

			serialNumber = dbConn.getCheckOutSerialNumber();

			// add server to server list

			if (subaction.equals("add")) {

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

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

				//Get data replication value

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

				//Get hub value

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

				String toDateSql = DatabaseService.getInstance().getDBAdapter().toDate("01/01/1980","MM/DD/YYYY");

				String sql = "INSERT INTO xml_replication "

						+ "(server, last_checked, replicate, datareplicate, hub) "

						+ "VALUES (?," + toDateSql + ",?,?,?)";

				pstmt = dbConn.prepareStatement(sql);

				pstmt.setString(1, server);

				pstmt.setInt(2, Integer.parseInt(replicate));

				pstmt.setInt(3, Integer.parseInt(dataReplicate));

				pstmt.setInt(4, Integer.parseInt(hub));

				String sqlReport = "XMLAccessAccess.getXMLAccessForDoc - SQL: " + sql;

				sqlReport += " [" + server + "," + replicate + 

					"," + dataReplicate + "," + hub + "]";

				logMetacat.info(sqlReport);

				pstmt.execute();

				pstmt.close();

				dbConn.commit();

				out.write("Server " + server + " added");

				response.setContentType("text/html");

				out.write("<html><body><table border=\"1\">");

				out.write("<tr><td><b>server</b></td><td><b>last_checked</b></td><td>");

				out.write("<b>replicate</b></td>");

				out.write("<td><b>datareplicate</b></td>");

				out.write("<td><b>hub</b></td></tr>");

				pstmt = dbConn.prepareStatement("SELECT * FROM xml_replication");

				//increase dbconnection usage

				dbConn.increaseUsageCount(1);

				pstmt.execute();

				ResultSet rs = pstmt.getResultSet();

				boolean tablehasrows = rs.next();

				while (tablehasrows) {

					out.write("<tr><td>" + rs.getString(2) + "</td><td>");

					out.write(rs.getString(3) + "</td><td>");

					out.write(rs.getString(4) + "</td><td>");

					out.write(rs.getString(5) + "</td><td>");

					out.write(rs.getString(6) + "</td></tr>");

					tablehasrows = rs.next();

				}

				out.write("</table></body></html>");

				// download certificate with the public key on this server

				// and import it as a trusted certificate

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

				if (certURL != null && !certURL.equals("")) {

					downloadCertificate(certURL);

				}

				// delete server from server list

			} else if (subaction.equals("delete")) {

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

				pstmt = dbConn.prepareStatement("DELETE FROM xml_replication "

						+ "WHERE server LIKE '" + server + "'");

				pstmt.execute();

				pstmt.close();

				dbConn.commit();

				out.write("Server " + server + " deleted");

				response.setContentType("text/html");

				out.write("<html><body><table border=\"1\">");

				out.write("<tr><td><b>server</b></td><td><b>last_checked</b></td><td>");

				out.write("<b>replicate</b></td>");

				out.write("<td><b>datareplicate</b></td>");

				out.write("<td><b>hub</b></td></tr>");

				pstmt = dbConn.prepareStatement("SELECT * FROM xml_replication");

				//increase dbconnection usage

				dbConn.increaseUsageCount(1);

				pstmt.execute();

				ResultSet rs = pstmt.getResultSet();

				boolean tablehasrows = rs.next();

				while (tablehasrows) {

					out.write("<tr><td>" + rs.getString(2) + "</td><td>");

					out.write(rs.getString(3) + "</td><td>");

					out.write(rs.getString(4) + "</td><td>");

					out.write(rs.getString(5) + "</td><td>");

					out.write(rs.getString(6) + "</td></tr>");

					tablehasrows = rs.next();

				}

				out.write("</table></body></html>");

				// list servers in server list

			} else if (subaction.equals("list")) {

				response.setContentType("text/html");

				out.write("<html><body><table border=\"1\">");

				out.write("<tr><td><b>server</b></td><td><b>last_checked</b></td><td>");

				out.write("<b>replicate</b></td>");

				out.write("<td><b>datareplicate</b></td>");

				out.write("<td><b>hub</b></td></tr>");

				pstmt = dbConn.prepareStatement("SELECT * FROM xml_replication");

				pstmt.execute();

				ResultSet rs = pstmt.getResultSet();

				boolean tablehasrows = rs.next();

				while (tablehasrows) {

					out.write("<tr><td>" + rs.getString(2) + "</td><td>");

					out.write(rs.getString(3) + "</td><td>");

					out.write(rs.getString(4) + "</td><td>");

					out.write(rs.getString(5) + "</td><td>");

					out.write(rs.getString(6) + "</td></tr>");

					tablehasrows = rs.next();

				}

				out.write("</table></body></html>");

			} else {

				out.write("<error>Unkonwn subaction</error>");

			}

			pstmt.close();

			//conn.close();

		} catch (Exception e) {

			logMetacat.error("ReplicationService.handleServerControlRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleServerControlRequest - Error in "

					+ "MetacatReplication.handleServerControlRequest " + e.getMessage());

			e.printStackTrace(System.out);

		} finally {

			try {

				pstmt.close();

			}//try

			catch (SQLException ee) {

				logMetacat.error("ReplicationService.handleServerControlRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

				logReplication.error("ReplicationService.handleServerControlRequest - Error in MetacatReplication.handleServerControlRequest to close pstmt "

						+ ee.getMessage());

			}//catch

			finally {

				DBConnectionPool.returnDBConnection(dbConn, serialNumber);

			}//finally

			if (out != null) {

				try {

					out.close();

				} catch (IOException e) {

					logMetacat.error(e.getMessage(), e);

				}

			}

		}//finally

	}

	// download certificate with the public key from certURL and

	// upload it onto this server; it then must be imported as a

	// trusted certificate

	private static void downloadCertificate(String certURL) throws FileNotFoundException,

			IOException, MalformedURLException, PropertyNotFoundException {

		// the path to be uploaded to

		String certPath = SystemUtil.getContextDir();

		// get filename from the URL of the certificate

		String filename = certURL;

		int slash = Math.max(filename.lastIndexOf('/'), filename.lastIndexOf('\\'));

		if (slash > -1) {

			filename = filename.substring(slash + 1);

		}

		// open file output strem to write the input into it

		File f = new File(certPath, filename);

		synchronized (f) {

			try {

				if (f.exists()) {

					throw new IOException("File already exist: " + f.getCanonicalFile());

					// if ( f.exists() && !f.canWrite() ) {

					// throw new IOException("Not writable: " +

					// f.getCanonicalFile());

				}

			} catch (SecurityException se) {

				// if a security manager exists,

				// its checkRead method is called for f.exist()

				// or checkWrite method is called for f.canWrite()

				throw se;

			}

			// create a buffered byte output stream

			// that uses a default-sized output buffer

			FileOutputStream fos = new FileOutputStream(f);

			BufferedOutputStream out = new BufferedOutputStream(fos);

			// this should be http url

			URL url = new URL(certURL);

			BufferedInputStream bis = null;

			try {

				bis = new BufferedInputStream(url.openStream());

				byte[] buf = new byte[4 * 1024]; // 4K buffer

				int b = bis.read(buf);

				while (b != -1) {

					out.write(buf, 0, b);

					b = bis.read(buf);

				}

			} finally {

				if (bis != null)

					bis.close();

			}

			// the input and the output streams must be closed

			bis.close();

			out.flush();

			out.close();

			fos.close();

		} // end of synchronized(f)

	}

	/**

	 * when a forcereplication request comes in, local host sends a read request

	 * to the requesting server (remote server) for the specified docid. Then

	 * store it in local database.

	 */

	protected static void handleForceReplicateRequest(

			Hashtable<String, String[]> params, HttpServletResponse response,

			HttpServletRequest request) {

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

		String docid = ((String[]) params.get("docid"))[0]; // sent the document

		String dbaction = "UPDATE"; // the default action is UPDATE

		//    boolean override = false;

		//    int serverCode = 1;

		DBConnection dbConn = null;

		int serialNumber = -1;

		String docName = null;

		try {

			//if the url contains a dbaction then the default action is overridden

			if (params.containsKey("dbaction")) {

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

				//serverCode = MetacatReplication.getServerCode(server);

				//override = true; //we are now overriding the default action

			}

			logReplication.info("ReplicationService.handleForceReplicateRequest - Force replication request from: " + server);

			logReplication.info("ReplicationService.handleForceReplicateRequest - Force replication docid: " + docid);

			logReplication.info("ReplicationService.handleForceReplicateRequest - Force replication action: " + dbaction);

			// sending back read request to remote server

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

					+ MetacatUtil.getLocalReplicationServerName() + "&action=read&docid="

					+ docid);

			String xmldoc = ReplicationService.getURLContent(u);

			// get the document info from server

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

					+ MetacatUtil.getLocalReplicationServerName()

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

			String docInfoStr = ReplicationService.getURLContent(docinfourl);

			//dih is the parser for the docinfo xml format

			DocInfoHandler dih = new DocInfoHandler();

			XMLReader docinfoParser = ReplicationHandler.initParser(dih);

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

			//      Hashtable<String,Vector<AccessControlForSingleFile>> docinfoHash = dih.getDocInfo();

			Hashtable<String, String> docinfoHash = dih.getDocInfo();

			// Get home server of this docid

			String homeServer = (String) docinfoHash.get("home_server");

			String guid = (String) docinfoHash.get("guid");

			logReplication.info("XXXXXXXXXXXXXXXX GUID found in dociinfoHash: " + guid);

			logReplication.info("Processing guid " + guid + 

			  " information from handleForceReplicationRequest: " + 

			  docinfoHash.toString());

      IdentifierManager idman = IdentifierManager.getInstance();

      if(guid != null)

      { //if the guid was passed in, put it in the identifiers table

        logReplication.info("YYYYYYYYYYYYYY Creating guid/docid mapping for docid " + 

          docinfoHash.get("docid") + " and guid: " + guid);

        docName = (String) docinfoHash.get("docname");

        logReplication.info("ZZZZZZZZZZZZ docName: " + docName);

        if(docName.trim().equals("systemMetadata"))

        {

        	if (dbaction.equalsIgnoreCase("UPDATE")) {

        		logReplication.info("update mapping for systemMetadata: guid: " + guid + " localId: " + docinfoHash.get("docid"));

	            idman.updateSystemMetadataMapping(guid, docinfoHash.get("docid"));

	            logReplication.info("handleForceReplicateRequest: updating SM with additional info");

        	} else { // insert

	            logReplication.info("creating mapping for systemMetadata: guid: " + guid + " localId: " + docinfoHash.get("docid"));

	            idman.createSystemMetadataMapping(guid, docinfoHash.get("docid"));

        	}

        	logReplication.info("handleForceReplicateRequest: updating SM with additional info");

        	Long dateUploadedLong = new Long(docinfoHash.get("date_uploaded"));

	        Long dateModifiedLong = new Long(docinfoHash.get("date_modified"));

            idman.insertAdditionalSystemMetadataFields(

                    dateUploadedLong.longValue(), 

                    docinfoHash.get("rights_holder"),

                    docinfoHash.get("checksum"), 

                    docinfoHash.get("checksum_algorithm"), 

                    docinfoHash.get("origin_member_node"),

                    docinfoHash.get("authoritive_member_node"), 

                    dateModifiedLong.longValue(),

                    docinfoHash.get("submitter"),

                    docinfoHash.get("guid"),

                    docinfoHash.get("object_format"),

                    new Long(docinfoHash.get("size")).longValue());

        }

        else

        {

        	if (dbaction.equalsIgnoreCase("UPDATE")) {

        		logReplication.info("updating mapping: guid: " + guid + " localId: " + docinfoHash.get("docid"));

	            idman.updateMapping(guid, docinfoHash.get("docid"));

        	} else { // insert

	            logReplication.info("creating mapping: guid: " + guid + " localId: " + docinfoHash.get("docid"));

	            idman.createMapping(guid, docinfoHash.get("docid"));

        	}

        }

      }

      else

      {

        logReplication.debug("No guid information was included with the replicated document");

      }

			String createdDate = (String) docinfoHash.get("date_created");

			String updatedDate = (String) docinfoHash.get("date_updated");

			logReplication.info("ReplicationService.handleForceReplicateRequest - homeServer: " + homeServer);

			// Get Document type

			String docType = (String) docinfoHash.get("doctype");

			logReplication.info("ReplicationService.handleForceReplicateRequest - docType: " + docType);

			String parserBase = null;

			// this for eml2 and we need user eml2 parser

			if (docType != null

					&& (docType.trim()).equals(DocumentImpl.EML2_0_0NAMESPACE)) {

				logReplication.warn("ReplicationService.handleForceReplicateRequest - This is an eml200 document!");

				parserBase = DocumentImpl.EML200;

			} else if (docType != null

					&& (docType.trim()).equals(DocumentImpl.EML2_0_1NAMESPACE)) {

				logReplication.warn("ReplicationService.handleForceReplicateRequest - This is an eml2.0.1 document!");

				parserBase = DocumentImpl.EML200;

			} else if (docType != null

					&& (docType.trim()).equals(DocumentImpl.EML2_1_0NAMESPACE)) {

				logReplication.warn("ReplicationService.handleForceReplicateRequest - This is an eml2.1.0 document!");

				parserBase = DocumentImpl.EML210;

			} else if (docType != null

					&& (docType.trim()).equals(DocumentImpl.EML2_1_1NAMESPACE)) {

				logReplication.warn("ReplicationService.handleForceReplicateRequest - This is an eml2.1.1 document!");

				parserBase = DocumentImpl.EML210;

			}

			logReplication.warn("ReplicationService.handleForceReplicateRequest - The parserBase is: " + parserBase);

			// Get DBConnection from pool

			dbConn = DBConnectionPool

					.getDBConnection("MetacatReplication.handleForceReplicateRequest");

			serialNumber = dbConn.getCheckOutSerialNumber();

			// write the document to local database

			DocumentImplWrapper wrapper = new DocumentImplWrapper(parserBase, false);

			//try this independently so we can set access even if the update action is invalid (i.e docid has not changed)

			try {

				wrapper.writeReplication(dbConn, xmldoc, null, null,

						dbaction, docid, null, null, homeServer, server, createdDate,

						updatedDate);

			} finally {

				//process extra access rules before dealing with the write exception (doc exist already)			

		        Vector<XMLAccessDAO> accessControlList = dih.getAccessControlList();

		        if (accessControlList != null) {

		        	AccessControlForSingleFile acfsf = new AccessControlForSingleFile(docid);

		        	for (XMLAccessDAO xmlAccessDAO : accessControlList) {

		        		if (!acfsf.accessControlExists(xmlAccessDAO)) {

		        			acfsf.insertPermissions(xmlAccessDAO);

							logReplication.info("ReplicationService.handleForceReplicateRequest - document " + docid

									+ " permissions added to DB");

		        		}

		            }

		        }

		        // process the real owner and updater

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

				String updated = (String) docinfoHash.get("user_updated");

		        updateUserOwner(dbConn, docid, user, updated);

				logReplication.info("ReplicationService.handleForceReplicateRequest - document " + docid + " added to DB with "

						+ "action " + dbaction);

				if(guid != null)

                {

                    if(!docName.trim().equals("systemMetadata"))

                    {

                        logReplication.info("replicate D1GUID:" + guid + ":D1SCIMETADATA:" + 

                                docid + ":");

                    }

                    else

                    {

                        logReplication.info("replicate D1GUID:" + guid + ":D1SYSMETADATA:" + 

                                docid + ":");

                    }

                }

				EventLog.getInstance().log(request.getRemoteAddr(), REPLICATIONUSER, docid, dbaction);

			}

		} catch (SQLException sqle) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - SQL error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + sqle.getMessage());

		} catch (MalformedURLException mue) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - URL error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + mue.getMessage());

		} catch (SAXException se) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - SAX parsing error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + se.getMessage());

		} catch (HandlerException he) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - Handler error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + he.getMessage());

		} catch (IOException ioe) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - I/O error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + ioe.getMessage());

		} catch (PermOrderException poe) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - Permissions order error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + poe.getMessage());

		} catch (AccessControlException ace) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - Permissions order error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + ace.getMessage());

		} catch (Exception e) {

			logMetacat.error("ReplicationService.handleForceReplicateRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("ReplicationService.handleForceReplicateRequest - General error when adding doc " + docid + 

					" to DB with action " + dbaction + ": " + e.getMessage());

		} finally {

			// Return the checked out DBConnection

			DBConnectionPool.returnDBConnection(dbConn, serialNumber);

		}//finally

	}

	/*

	 * when a forcereplication delete request comes in, local host will delete this

	 * document

	 */

	protected static void handleForceReplicateDeleteRequest(

			Hashtable<String, String[]> params, HttpServletResponse response,

			HttpServletRequest request) {

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

		String docid = ((String[]) params.get("docid"))[0]; // sent the document

		try {

			logReplication.info("ReplicationService.handleForceReplicateDeleteRequest - force replication delete request from " + server);

			logReplication.info("ReplicationService.handleForceReplicateDeleteRequest - force replication delete docid " + docid);

			logReplication.info("ReplicationService.handleForceReplicateDeleteRequest - Force replication delete request from: " + server);

			logReplication.info("ReplicationService.handleForceReplicateDeleteRequest - Force replication delete docid: " + docid);

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

			logReplication.info("ReplicationService.handleForceReplicateDeleteRequest - document " + docid + " was successfully deleted ");

			EventLog.getInstance().log(request.getRemoteAddr(), REPLICATIONUSER, docid,

					"delete");

			logReplication.info("ReplicationService.handleForceReplicateDeleteRequest - document " + docid + " was successfully deleted ");

		} catch (McdbDocNotFoundException e) {

			logMetacat.error("ReplicationService.handleForceReplicateDeleteRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("document " + docid

					+ " failed to delete because " + e.getMessage());

			logReplication.error("ReplicationService.handleForceReplicateDeleteRequest - error: " + e.getMessage());

		} catch (InsufficientKarmaException e) {

			logMetacat.error("ReplicationService.handleForceReplicateDeleteRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("document " + docid

					+ " failed to delete because " + e.getMessage());

			logReplication.error("ReplicationService.handleForceReplicateDeleteRequest - error: " + e.getMessage());

		} catch (SQLException e) {

			logMetacat.error("ReplicationService.handleForceReplicateDeleteRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("document " + docid

					+ " failed to delete because " + e.getMessage());

			logReplication.error("ReplicationService.handleForceReplicateDeleteRequest - error: " + e.getMessage());

		} catch (Exception e) {

			logMetacat.error("ReplicationService.handleForceReplicateDeleteRequest - " + ReplicationService.METACAT_REPL_ERROR_MSG);                         

			logReplication.error("document " + docid

					+ " failed to delete because " + e.getMessage());

			logReplication.error("ReplicationService.handleForceReplicateDeleteRequest - error: " + e.getMessage());

		}//catch

	}

	/**

	 * when a forcereplication data file request comes in, local host sends a

	 * readdata request to the requesting server (remote server) for the specified

	 * docid. Then store it in local database and file system

	 */

	protected static void handleForceReplicateDataFileRequest(Hashtable<String, String[]> params,

			HttpServletRequest request) {

		//make sure there is some parameters

		if (params.isEmpty()) {

			return;

		}

		// Get remote server

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

		// the docid should include rev number

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