specificdefault.java

java开源的企业总线.xmlBlaster

/*------------------------------------------------------------------------------
 Name:      SpecificDefault.java
 Project:   xmlBlaster.org
 Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
 ------------------------------------------------------------------------------*/

package org.xmlBlaster.contrib.replication.impl;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URL;
import java.net.URLClassLoader;
import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.logging.Logger;

import org.xmlBlaster.contrib.GlobalInfo;
import org.xmlBlaster.contrib.I_Info;
import org.xmlBlaster.contrib.PropertiesInfo;
import org.xmlBlaster.contrib.VersionTransformerCache;
import org.xmlBlaster.contrib.db.DbMetaHelper;
import org.xmlBlaster.contrib.db.I_DbPool;
import org.xmlBlaster.contrib.db.I_ResultCb;
import org.xmlBlaster.contrib.dbwatcher.DbWatcher;
import org.xmlBlaster.contrib.dbwatcher.convert.I_AttributeTransformer;
import org.xmlBlaster.contrib.dbwatcher.convert.ResultSetToXmlConverter;
import org.xmlBlaster.contrib.dbwriter.info.SqlInfo;
import org.xmlBlaster.contrib.dbwriter.info.SqlColumn;
import org.xmlBlaster.contrib.dbwriter.info.SqlDescription;
import org.xmlBlaster.contrib.replication.I_DbSpecific;
import org.xmlBlaster.contrib.replication.I_Mapper;
import org.xmlBlaster.contrib.replication.ReplicationConstants;
import org.xmlBlaster.contrib.replication.ReplicationConverter;
import org.xmlBlaster.contrib.replication.TableToWatchInfo;
import org.xmlBlaster.util.I_ReplaceVariable;
import org.xmlBlaster.util.ReplaceVariable;

public abstract class SpecificDefault implements I_DbSpecific /*, I_ResultCb */ {
   public final static boolean ROLLBACK_YES = true;
   public final static boolean ROLLBACK_NO = false;
   public final static boolean COMMIT_YES = true;
   public final static boolean COMMIT_NO = false;
   
   private static Logger log = Logger.getLogger(SpecificDefault.class.getName());

   private int rowsPerMessage = 10;

   protected I_Info info;

   protected I_DbPool dbPool;

   protected DbMetaHelper dbMetaHelper;

   protected String replPrefix = "repl_";
   
   protected String replVersion = "0.0";
   
   protected ReplaceVariable replaceVariable;

   protected Replacer replacer;

   protected InitialUpdater initialUpdater;
   
   protected I_AttributeTransformer transformer;

   private boolean bootstrapWarnings;
   
   private int initCount = 0;
   
   private boolean isInMaster;
   
   private Set cancelledUpdates = new HashSet();
   
   protected boolean isDbWriteable = true;
      
   class Replacer implements I_ReplaceVariable {

      private I_Info info;
      private Map additionalMap;
      
      public Replacer(I_Info info, Map additionalMap) {
         this.info = info;
         this.additionalMap = additionalMap;
         if (this.additionalMap == null)
            this.additionalMap = new HashMap();
      }
      
      public String get(String key) {
         if (key == null)
            return null;
         String repl = (String)this.additionalMap.get(key);
         if (repl != null)
            return repl.trim();
         repl = this.info.get(key, null);
         if (repl != null)
            return repl.trim();
         return null;
      }
      
      public Map getAdditionalMapClone() {
         return new HashMap(this.additionalMap);
      }
   }
   
   /**
    * Not doing anything.
    */
   public SpecificDefault() {
   }

   /**
    * 
    * @param filename
    * @param method
    * @return List of String[]
    * @throws Exception
    */
   public List getContentFromClasspath(String filename, String method, String flushSeparator, String cmdSeparator) throws Exception {
      if (filename == null)
         throw new Exception(method + ": no filename specified");
      ArrayList list = new ArrayList();
      ArrayList internalList = new ArrayList();
      try {
         Enumeration enm = this.getClass().getClassLoader().getResources(filename);
         if(enm.hasMoreElements()) {
            URL url = (URL)enm.nextElement();
            log.info(method + ": : loading file '" + url.getFile() + "'");
            try {
               StringBuffer buf = new StringBuffer();
               BufferedReader reader = new BufferedReader(new InputStreamReader(url.openStream()));
               String line = null;
               while ( (line = reader.readLine()) != null) {
                  if (line.trim().startsWith(cmdSeparator)) {
                     String tmp =  buf.toString();
                     if (tmp.length() > 0)
                        internalList.add(tmp);
                     buf = new StringBuffer();
                  }
                  else if (line.trim().startsWith(flushSeparator)) {
                     if (buf.length() > 0)
                        internalList.add(buf.toString());
                     if (internalList.size() >0) {
                        String[] tmp = (String[])internalList.toArray(new String[internalList.size()]);
                        list.add(tmp);
                        internalList.clear();
                        buf = new StringBuffer();
                     }
                  }
                  else {
                     line = line.trim();
                     if (line.length() > 0 && !line.startsWith("--"))
                        buf.append(line).append("\n");
                  }
               }
               String end = buf.toString().trim();
               if (end.length() > 0)
                  internalList.add(end);
               if (internalList.size() >0) {
                  String[] tmp = (String[])internalList.toArray(new String[internalList.size()]);
                  list.add(tmp);
                  internalList.clear();
                  buf = null;
               }
               
            }
            catch(IOException ex) {
               log.warning("init: could not read properties from '" + url.getFile() + "' : " + ex.getMessage());
            }

            while(enm.hasMoreElements()) {
               url = (URL)enm.nextElement();
               log.warning("init: an additional matching file has been found in the classpath at '"
                  + url.getFile() + "' please check that the correct one has been loaded (see info above)"
               );
            }
            return list;
         }
         else {
            ClassLoader cl = this.getClass().getClassLoader();
            StringBuffer buf = new StringBuffer();
            if (cl instanceof URLClassLoader) {
               URL[] urls = ((URLClassLoader)cl).getURLs();
               for (int i=0; i < urls.length; i++) 
                  buf.append(urls[i].toString()).append("\n");
            }
            throw new Exception("init: no file found with the name '" + filename + "' : " + (buf.length() > 0 ? " classpath: " + buf.toString() : ""));
         }
      }
      catch(IOException e) {
         throw new Exception("init: an IOException occured when trying to load property file '" + filename + "'", e);
      }
   }
   
   /**
    * Replaces all tokens found in the content and returns the content with the values of the tokens.
    * @param content
    * @return
    */
   private final String replaceTokens(String content, Replacer repl) {
      return this.replaceVariable.replace(content, repl);
   }
   
   /**
    * Gets the specified object name and returns its value (name).
    * For example 'CREATE TABLE one' would return 'one'. Needed on bootstrapping.
    * NOTE: only made public for testing purposes.
    * @param op
    * @param req
    * @return
    */
   public final String getObjectName(String op, String req) {
      if (req == null) {
         log.warning("getObjectName had a null argument");
         return null;
      }  
      req = req.trim();
      if (req.length() < 1) {
         log.warning("getObjectName had an empty argument");
         return null;
      }
      String tmp = req.toUpperCase();
      if (tmp.startsWith(op)) {
         tmp = req.substring(op.length()).trim();
         int pos1 = tmp.indexOf('(');
         int pos2 = tmp.indexOf(' '); // whichever comes first
         if (pos1 < 0) {
            if (pos2 < 0) {
               log.warning("getObjectName for '" + op + "' on '" + req + "': no '(' or ' ' found.");
               return null;
            }
            return tmp.substring(0, pos2).trim();
         }
         else if (pos2 < 0) {
            return tmp.substring(0, pos1).trim();
         } 
         else {
            if (pos2 < pos1)
               pos1 = pos2;
            return tmp.substring(0, pos1).trim();
         }
      }
      return null;
   }
   
   /**
    * Checks if the table has to be created. 
    * If it is a 'CREATE TABLE' operation a non-negative value is returned,
    * if it is another kind of operation, -1 is returned.
    * If the table already exists, it returns zero.
    * If the table does not exist, it returns 1.
    * NOTE: only made public for testing purposes.
    * @param creationRequest the sql request to analyze.
    * @return
    * @throws Exception
    */
   public final int checkTableForCreation(String creationRequest) throws Exception {
      String tmp = getObjectName("CREATE TABLE", creationRequest);
      if (tmp == null)
         return -1;
      String name = this.dbMetaHelper.getIdentifier(tmp);
      if (name == null)
         return -1;
      
      Connection conn = null;
      try {
         conn = this.dbPool.reserve();
         conn.setAutoCommit(true);
         ResultSet rs = conn.getMetaData().getTables(null, getOwnSchema(), name, null);
         boolean exists = rs.next();
         rs.close();
         if (exists) {
            log.info("table '" +  name + "' exists, will not create it");
            return 0; 
         }
         else {
            log.info("table '" +  name + "' does not exist, will create it");
            return 1;
         }
      }
      catch (Exception ex) {
         conn = removeFromPool(conn, ROLLBACK_NO);
         throw ex;
      }
      finally {
         conn = releaseIntoPool(conn, COMMIT_NO);
      }
   }

   protected abstract boolean sequenceExists(Connection conn, String sequenceName) throws Exception;
   
   protected abstract boolean triggerExists(Connection conn, String triggerName) throws Exception;
   /**
    * Checks if the sequence has to be created. 
    * If it is a 'CREATE SEQUENCE' operation a non-negative value is returned,
    * if it is another kind of operation, -1 is returned.
    * If the sequence already exists, it returns zero.
    * If the sequence does not exist, it returns 1.
    * @param creationRequest the sql request to analyze.
    * NOTE: only made public for testing purposes.
    * @return 
    * @throws Exception
    */
   public int checkSequenceForCreation(String creationRequest) throws Exception {
      String tmp = getObjectName("CREATE SEQUENCE", creationRequest);
      if (tmp == null)
         return -1;
      String name = this.dbMetaHelper.getIdentifier(tmp);
      if (name == null)
         return -1;
      
      Connection conn = null;
      try {
         conn = this.dbPool.reserve();
         conn.setAutoCommit(true);
         try {
            if (sequenceExists(conn, name)) {
               log.info("sequence '" +  name + "' exists, will not create it");
               return 0;
            }
            else {
               log.info("sequence '" +  name + "' does not exist, will create it");
               return 1;
            }
            //incrementReplKey(conn);
            //log.info("sequence '" +  name + "' exists, will not create it");
            //return 0; 
         }
         catch (Exception ex) {
            log.info("table '" +  name + "' does not exist (an exception occured), will create it");
            return 1;
         }
      }
      catch (Exception ex) {
         conn = removeFromPool(conn, ROLLBACK_NO);
         throw ex;
      }
      finally {
         conn = releaseIntoPool(conn, COMMIT_NO);
      }
   }
   
   /**
    * Checks if the trigger has to be created.
    * If it is a 'CREATE TRIGGER' operation a non-negative value is returned,
    * if it is another kind of operation, -1 is returned.
    * If the triggger already exists, it returns zero.
    * If the trigger does not exist, it returns 1.
    * @param creationRequest the sql request to analyze.
    * NOTE: only made public for testing purposes.
    * @return 
    * @throws Exception
    */
   public final int checkTriggerForCreation(String creationRequest) throws Exception {
      String tmp = getObjectName("CREATE TRIGGER", creationRequest);
      if (tmp == null)
         return -1;
      String name = this.dbMetaHelper.getIdentifier(tmp);
      if (name == null)
         return -1;
      
      Connection conn = null;
      try {
         conn = this.dbPool.reserve();