jdbcmanagercommontable.java

java开源的企业总线.xmlBlaster

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

package org.xmlBlaster.util.queue.jdbc;

import java.util.logging.Logger;
import java.util.logging.Level;

import org.xmlBlaster.util.ThreadLister;
import org.xmlBlaster.util.XmlBlasterException;
import org.xmlBlaster.util.def.ErrorCode;
import org.xmlBlaster.util.Global;

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
// import java.sql.BatchUpdateException;

import java.util.Hashtable;

import org.xmlBlaster.util.plugin.PluginInfo;
import org.xmlBlaster.util.queue.QueuePluginManager;
import org.xmlBlaster.util.queue.StorageId;
import org.xmlBlaster.util.queue.I_Entry;
import org.xmlBlaster.util.queue.I_EntryFilter;
import org.xmlBlaster.util.queue.I_Queue;
import org.xmlBlaster.util.queue.I_Storage;
import org.xmlBlaster.util.queue.I_EntryFactory;
import org.xmlBlaster.util.queue.ReturnDataHolder;
import org.xmlBlaster.util.queue.I_StorageProblemListener;
import org.xmlBlaster.util.queue.I_StorageProblemNotifier;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.Properties;
import java.util.WeakHashMap;
import java.io.ByteArrayInputStream;
import java.io.InputStream;


/**
 * Delegate class which takes care of SQL specific stuff for the JdbcQueuePlugin
 * class.
 * <p>
 * One instance of this is created by Global for each StorageId prefix,
 * so one instance for 'history', one for 'cb' etc.
 * </p>
 *
 * The tables needed for each JdbcManagerCommonTable instance are the following:<br />
 * - entriesTableName (defaults to 'ENTRIES')<br />
 * The names of such tables are constituted by the tableNamePrefix (which defaults to 'XB') plus the 
 * entriesTableName
 * 
 * @see <a href="http://www.xmlBlaster.org/xmlBlaster/doc/requirements/queue.jdbc.commontable.html">The queue.jdbc.commontable requirement</a>
 * @author <a href="mailto:[e-mail]">Marcel Ruff</a>
 * @author <a href='mailto:[e-mail]'>Michele Laghi</a>
 *
 */
public class JdbcManagerCommonTable implements I_StorageProblemListener, I_StorageProblemNotifier {

   private static final String ME = "JdbcManagerCommonTable";
   private final Global glob;
   private static Logger log = Logger.getLogger(JdbcManagerCommonTable.class.getName());
   private final JdbcConnectionPool pool;
   private final I_EntryFactory factory;
   private final WeakHashMap listener;
   private final static String DUMMY_VALUE = "A";

   private final String tableNamePrefix;
   private final String colNamePrefix;
   // the names to be used
   private String stringTxt = null;
   private String longintTxt = null;
   private String intTxt = null;
   private String blobTxt = null;
   private String booleanTxt = null;

   private int maxStatementLength = 0;
   private boolean isConnected = true;

   private static boolean first = true;
   private String entriesTableName = null;
   private String blobVarName;
   private String byteSizeColName;
   private String dataIdColName;
   private String keyAttr;
   
   private boolean isPostgres;
   private boolean isOracle;
   private boolean isDB2;
   private boolean isFirebird;
   private boolean isMicrosoftSQLServer;
   private boolean isHSQLDatabaseEngine;
   private boolean isMySql;
   private boolean isLdbc;
   private boolean isSQLite;

  // private final String managerName;
   private final I_Storage storage;

   private PreparedStatement pingPrepared = null;

   /**
    * Counts the queues using this manager.
    */
   private int queueCounter = 0;

   /**
    * tells wether the used database supports batch updates or not.
    */
    private boolean supportsBatch = true;
    
    /** forces the desactivation of batch mode when adding entries */
    private boolean enableBatchMode = true;

    /** Column index into XB_ENTRIES table */
    final static int DATA_ID = 1;
    final static int QUEUE_NAME = 2;
    final static int PRIO = 3;
    final static int TYPE_NAME = 4;
    final static int PERSISTENT = 5;
    final static int SIZE_IN_BYTES = 6;
    final static int BLOB = 7;
    
    private int maxNumStatements;
    private int maxSelectLimit;
    
   /**
    * @param storage TODO
    * @param JdbcConnectionPool the pool to be used for the connections to
    *        the database. IMPORTANT: The pool must have been previously
    *        initialized.
    */
   public JdbcManagerCommonTable(JdbcConnectionPool pool, I_EntryFactory factory, String managerName, I_Storage storage)
      throws XmlBlasterException {
   //   this.managerName = managerName;
      this.pool = pool;
      this.glob = this.pool.getGlobal();

      this.storage = storage;
      if (log.isLoggable(Level.FINER)) log.finer("Constructor called");

      this.factory = factory;

      if (!this.pool.isInitialized())
         throw new XmlBlasterException(glob, ErrorCode.INTERNAL_UNKNOWN, ME, "constructor: the Connection pool is not properly initialized");

      // get the necessary metadata
      Connection conn = null;
      boolean success = true;
      try {
         conn = this.pool.getConnection();
         DatabaseMetaData dbmd = conn.getMetaData();
         this.maxStatementLength = dbmd.getMaxStatementLength();
         if (this.maxStatementLength < 1) {
            this.maxStatementLength = glob.getProperty().get("queue.persistent.maxStatementLength", 2048);
            if (first) {
               log.info("The maximum SQL statement length is not defined in JDBC meta data, we set it to " + this.maxStatementLength);
               first = false;
            }
         }

         if (!dbmd.supportsTransactions()) {
            String dbName = dbmd.getDatabaseProductName();
            log.severe("the database '" + dbName + "' does not support transactions, unpredicted results may happen");
         }

         if (!dbmd.supportsBatchUpdates()) {
            String dbName = dbmd.getDatabaseProductName();
            this.supportsBatch = false;
            log.fine("the database '" + dbName + "' does not support batch mode. No problem I will work whitout it");
         }
         
         // zero means not limit (to be sure we also check negative Values
         boolean logWarn = false;
         int defaultMaxNumStatements = dbmd.getMaxStatements();
         if (defaultMaxNumStatements < 1) {
            defaultMaxNumStatements = 50;
            logWarn = true;
         }
         // -queue.persistent.maxNumStatements 50
         this.maxNumStatements = this.pool.getProp("maxNumStatements", defaultMaxNumStatements);
         log.info("The maximum Number of statements for this database instance are '" + this.maxNumStatements + "'");
         if (logWarn && this.pool.getProp("maxNumStatements",-1)==-1)
            log.warning("The maxStatements returned fromt the database metadata is '" + defaultMaxNumStatements + "', will set the default to 50 unless you explicitly set '-queue.persistent.maxNumStatements <num>'");

         // -queue.persistent.maxSelectLimit -1 (off)
         this.maxSelectLimit = this.pool.getProp("maxSelectLimit", -1);
         if (this.maxSelectLimit > 0)
            log.info("The maximum results returned by a select is set to '" + this.maxSelectLimit + "' (MSSQLerver only)");
         
         log.info("Using DB " + dbmd.getDatabaseProductName() + " " + dbmd.getDatabaseProductVersion() + " " + dbmd.getDriverName());
      }
      catch (XmlBlasterException ex) {
         success = false;
         throw ex;
      }
      catch (Throwable ex) {
         success = false;
         throw new XmlBlasterException(glob, ErrorCode.RESOURCE_CONFIGURATION, ME, "constructor: failed to get the metadata", ex);
      }
      finally {
         if (conn != null) 
            this.pool.releaseConnection(conn, success);
      }

      Hashtable names = pool.getMapping();
      this.listener = new WeakHashMap();

      this.stringTxt = (String)names.get("string");
      if (this.stringTxt == null) this.stringTxt = "text";

      this.longintTxt = (String)names.get("longint");
      if (this.longintTxt == null) this.longintTxt = "bigint";

      this.intTxt = (String)names.get("int");
      if (this.intTxt == null) this.intTxt = "integer";

      this.booleanTxt = (String)names.get("boolean");
      if (this.booleanTxt == null) this.booleanTxt = "char(1)";

      this.blobTxt = (String)names.get("blob");
      if (this.blobTxt == null) this.blobTxt = "bytea";

      this.blobVarName = (String)names.get("blobVarName");
      if (this.blobVarName == null) this.blobVarName = "blob";

      this.keyAttr = (String)names.get("keyAttr");
      if (this.keyAttr == null) this.keyAttr = ""; // could be "not null" for MySQL

      this.tableNamePrefix = this.pool.getTableNamePrefix();
      this.colNamePrefix = this.pool.getColNamePrefix();
      // this.queueIncrement = this.pool.getTableAllocationIncrement(); // 2

      this.entriesTableName = this.tableNamePrefix + 
                            pool.getPluginProperties().getProperty("entriesTableName", "ENTRIES");
      this.entriesTableName = this.entriesTableName.toUpperCase();

      // byteSize and dataId are reserved in MS-SQLServer, prefixing other column names are not yet coded
      this.byteSizeColName = this.colNamePrefix + "byteSize";

      this.dataIdColName = this.colNamePrefix + "dataId";
      
      this.enableBatchMode = this.pool.isBatchModeEnabled();

      this.isPostgres = this.pool.getUrl().startsWith("jdbc:postgresql:");
      this.isOracle = this.pool.getUrl().startsWith("jdbc:oracle:");
      this.isDB2 = this.pool.getUrl().startsWith("jdbc:db2:");
      this.isFirebird = this.pool.getUrl().startsWith("jdbc:firebirdsql:");
      this.isMicrosoftSQLServer = this.pool.getUrl().startsWith("jdbc:microsoft:sqlserver:") // 2000
         || this.pool.getUrl().startsWith("jdbc:sqlserver:"); // 2005
      this.isHSQLDatabaseEngine = this.pool.getUrl().startsWith("jdbc:hsqldb:");
      this.isMySql = this.pool.getUrl().startsWith("jdbc:mysql:");
      this.isLdbc = this.pool.getUrl().startsWith("jdbc:ldbc:");
      this.isSQLite = this.pool.getUrl().startsWith("jdbc:sqlite:");

      this.pool.registerManager(this);
   }

   /**
    * pings the jdbc connection to check if the DB is up and running. It returns
    * 'true' if the connection is OK, false otherwise. The ping is done by invocation 
    */
   public boolean ping() {
      Connection conn = null;
      boolean success = true;
      try {
         conn = this.pool.getConnection();
         boolean ret = ping(conn);
         return ret;
      }
      catch (XmlBlasterException ex) {
         success = false;
         log.warning("ping failed due to problems with the pool. Check the jdbc pool size in 'xmlBlaster.properties'. Reason :" + ex.getMessage());
         return false;
      }
      finally {
         try {
            if (conn != null) 
               this.pool.releaseConnection(conn, success);
         }
         catch (XmlBlasterException e) {
            log.severe("ping: releaseConnection failed: " + e.getMessage());
         }
      }
   }


   /**
    * pings the jdbc connection to check if the DB is up and running. It returns
    * 'true' if the connection is OK, false otherwise. The ping is done by invocation 
    */
// isClosed() does not work
   private boolean ping(Connection conn) {
      if (log.isLoggable(Level.FINER)) log.finer("ping");
      if (conn == null) return false; // this could occur if it was not possible to create the connection

//      Statement st = null;
      try {
         // conn.isClosed();

         if (this.pingPrepared == null) {
            synchronized (this) {
               if (this.pingPrepared == null) {
                  this.pingPrepared = conn.prepareStatement("select min("+this.dataIdColName+") from " + this.entriesTableName);
               }
            }
         }
         this.pingPrepared.executeQuery();

         // Until v1.5.1+: Did not work with MSSQLServer
         //if (log.isLoggable(Level.FINE)) log.fine("Trying ping ...");
         //conn.getMetaData().getTables("xyx", "xyz", "xyz", null);