jdbcconnectionpool.java

java开源的企业总线.xmlBlaster

    * This method is to be invoked when an exception occured and it can be supposed that the connection
    * is not good anymore.
    * 
    * @param conn
    * @throws XmlBlasterException
    */
   public void discardConnection(Connection conn) throws XmlBlasterException {
      if (conn == null)
         throw new XmlBlasterException(this.glob, ErrorCode.RESOURCE_DB_UNKNOWN, "discardConnection", "The connection to discard is null"); 
      if (log.isLoggable(Level.FINER)) 
         log.finer("discardConnection " + this.connections.size() + " waiting calls: " + this.waitingCalls);
      try {
         SQLWarning warns = conn.getWarnings();
         if (log.isLoggable(Level.FINE)) {
            while (warns != null) {
               log.fine("errorCode=" + warns.getErrorCode() + " state=" + warns.getSQLState() + ": " + warns.toString().trim());
               warns = warns.getNextWarning();
            }
         }
      }
      catch (Throwable e) {
         log.warning("clearWarnings() failed: " + e.toString());
      }
      
      try {
         conn.close();
      }
      catch (Throwable ex) {
         log.warning("Could not close the connection to be discarded");
      }
      try {
         addConnectionToPool(false);
      }
      catch (SQLException ex) {
         log.warning("Could not close the connection to be discarded");
      }
   }
   

   public static String isolationToString(int isolation) {
      if (isolation == Connection.TRANSACTION_READ_COMMITTED)
         return "TRANSACTION_READ_COMMITTED";
      if (isolation == Connection.TRANSACTION_READ_UNCOMMITTED)
         return "TRANSACTION_READ_UNCOMMITTED";
      if (isolation == Connection.TRANSACTION_REPEATABLE_READ)
         return "TRANSACTION_REPEATABLE_READ";
      if (isolation == Connection.TRANSACTION_SERIALIZABLE)
         return "TRANSACTION_SERIALIZABLE";
      if (isolation == Connection.TRANSACTION_NONE)
         return "TRANSACTION_NONE";
      return "" + isolation;
   }

   public static String getIsolationLevel(Connection conn) {
      try {
         int isolation = conn.getTransactionIsolation();
         return "Supports connection TRANSACTION_READ_COMMITTED-"+Connection.TRANSACTION_READ_COMMITTED+"="
               + conn.getMetaData().supportsTransactionIsolationLevel(
                     Connection.TRANSACTION_READ_COMMITTED)
               + ", TRANSACTION_READ_UNCOMMITTED-"+Connection.TRANSACTION_READ_UNCOMMITTED+"="
               + conn.getMetaData().supportsTransactionIsolationLevel(
                     Connection.TRANSACTION_READ_UNCOMMITTED)
               + ", TRANSACTION_REPEATABLE_READ-"+Connection.TRANSACTION_REPEATABLE_READ+"="
               + conn.getMetaData().supportsTransactionIsolationLevel(
                     Connection.TRANSACTION_REPEATABLE_READ)
               + ", TRANSACTION_SERIALIZABLE-"+Connection.TRANSACTION_SERIALIZABLE+"="
               + conn.getMetaData().supportsTransactionIsolationLevel(
                     Connection.TRANSACTION_SERIALIZABLE)
               + ". Using transaction isolation "
               + isolationToString(isolation) + "=" + isolation;
         // conn.setTransactionIsolation(Connection.TRANSACTION_SERIALIZABLE);
      } catch (SQLException e) {
         log.warning(e.toString());
         return "";
      }
   }
   
   private final void setInPool(Connection conn, boolean inPool) {
      //if (conn instanceof DebugConnection)
      //   ((DebugConnection)conn).setInPool(inPool); // assert code
   }
   private final boolean isInPool(Connection conn) {
      //if (conn instanceof DebugConnection)
      //   return ((DebugConnection)conn).isInPool();
      return false;
   }
   
   private synchronized void addConnectionToPool(boolean doLog) throws SQLException {
      try {
         if (this.connections.size() == this.connections.capacity()) {
            log.severe("Can't add more JDBC connections to pool, capacity="
                  + this.connections.capacity() + " is reached");
            return;
         }
         // Connection conn = DriverManager.getConnection(url, user, password);
         Connection conn = null;
         //if (this.debug)
         //   conn = new DebugConnection(DriverManager.getConnection(url, user, password));
         //else
            conn = DriverManager.getConnection(url, user, password);
         if (doLog) {
            log.info(getIsolationLevel(conn));
         }
         if (this.forceIsoaltionLevel != -1)
            conn.setTransactionIsolation(this.forceIsoaltionLevel);
         this.connections.put(conn);
         setInPool(conn, true);
         // log.info(ME, "DriverManager:" + buf.toString());
      }
      catch (InterruptedException e) {
         log.severe("connect: an interrupted exception occured " + e.getMessage());
      }
   }
   
   /**
    * Connects to the DB (so many connections as configured)
    * @param disconnectFirst if 'true' then all connections to the db are closed before reconnecting.
    */
   private void connect(boolean disconnectFirst, boolean doLog) throws SQLException {
      int oldStatus;
      I_StorageProblemListener lst = null;
      synchronized(this) {
         if (disconnectFirst) disconnect();
         for (int i = 0; i < this.capacity; i++) {
            if (log.isLoggable(Level.FINE)) log.fine("initializing DB connection "+ i + " url=" + url + " user=" + user); // + " password=" + password);
            //Logging since JDK 1.3:
            //java.io.OutputStream buf = new java.io.ByteArrayOutputStream();
            //java.io.PrintStream pr = new java.io.PrintStream(buf);
            //DriverManager.setLogStream(pr);
            addConnectionToPool((i==0)&&doLog);
            if (log.isLoggable(Level.FINE)) 
               log.fine("initialized DB connection "+ i + " success");
         }
         oldStatus = this.status;
         this.status = I_StorageProblemListener.AVAILABLE;
         lst = this.storageProblemListener;
      }
      this.isShutdown = false;
      if (lst != null) lst.storageAvailable(oldStatus);
      log.info("Successfully reconnected to database");
   }

   public String getProp(String key, String def) {
      final String prefix = "queue.persistent.";
      org.xmlBlaster.util.property.Property prop = glob.getProperty();
      def = prop.get(prefix + key, def);
      return pluginProp.getProperty(key, def);
   }
   
   public int getProp(String key, int def) {
      final String prefix = "queue.persistent.";
      org.xmlBlaster.util.property.Property prop = glob.getProperty();
      def = prop.get(prefix + key, def);
      String tmp = pluginProp.getProperty(key, "" + def);
      try {
         return Integer.parseInt(tmp.trim());
      }
      catch (Exception ex) {
         return def;
      }
   }

   /**
    * Is called after the instance is created. It reads the needed properties,
    * sets the driver and then connects to the database as many times as
    * specified in the properties. <p/>
    * Note that prefix is the string which identifies a certain database.
    * It is called prefix because it is the prefix in the xmlBlaster.properties
    * for the properties bound to the database.<br> A pool instance is created
    * for every such database. For example if you want the history queue (the
    * queue which stores the published messages) on a different database than
    * the callback queues, then you could define different databases.
    *
    * <li>prefix.driverName DEFAULTS TO "org.postgresql.Driver"</li> <li>prefix.connectionPoolSize "DEFAULTS TO 2"</li>
    * <li>prefix.url DEFAULTS TO "jdbc:postgresql://localhost/test"</li> <li>prefix.user DEFAULTS TO "postgres"</li>
    * <li>prefix.password  DEFAULTS TO ""</li> </ul>
    */
   public synchronized void initialize(Global glob, Properties pluginProperties)
      throws ClassNotFoundException, SQLException, XmlBlasterException {

      if (this.initialized) return;
      this.glob = glob;
      this.pluginProp = pluginProperties;

      if (log.isLoggable(Level.FINE)) log.fine("initialize");

      // could these also be part of the properties specific to the invoking plugin ?
      org.xmlBlaster.util.property.Property prop = glob.getProperty();
      String xmlBlasterJdbc = prop.get("JdbcDriver.drivers", "org.postgresql.Driver:sun.jdbc.odbc.JdbcOdbcDriver:ORG.as220.tinySQL.dbfFileDriver:oracle.jdbc.driver.OracleDriver");

      // in xmlBlaster.properties file we separate the drivers with a ',' but
      // in the system properties they should be separated with ':' so it may
      // be time to change that in our properties
      //(THIS IS NOT NEEDED ANYMORE (fixed 2002-10-07 Michele)
      // xmlBlasterJdbc = xmlBlasterJdbc.replace(',', ':');

      System.setProperty("jdbc.drivers", xmlBlasterJdbc);

      // Default is:
      //   queue.persistent.url=jdbc:postgresql://localhost/test
      // This has precedence: 
      //   cb.queue.persistent.url=jdbc:postgresql://localhost/test
      //   client.queue.persistent.url=jdbc:postgresql://localhost/test

      // TODO:
      //
      // Instance settings:
      // "queue/history/maxEntries" -> "queue=history, maxEntries=5"
      //
      // Class settings:
      // "plugin/QueuePlugin[JDBC][1.0]/className/org.xmlBlaster.util.queue.jdbc.JdbcQueueCommonTablePlugin"
      // "plugin/QueuePlugin[JDBC][1.0]/tableNamePrefix/XB_"

      // the old generic properties (for the defaults) outside the plugin 
      this.url = prop.get("queue.persistent.url", "jdbc:postgresql://localhost/test");
      this.user = prop.get("queue.persistent.user", "postgres");
      this.password = prop.get("queue.persistent.password", "");
      this.capacity = prop.get("queue.persistent.connectionPoolSize", MIN_POOL_SIZE);
      if (this.capacity < MIN_POOL_SIZE) {
         log.warning("queue.persistent.connectionPoolSize=" + this.capacity + " is too small, setting it to " + MIN_POOL_SIZE);
         this.capacity = MIN_POOL_SIZE;
      }
      this.queryTimeout = prop.get("queue.persistent.queryTimeout", 0);
      this.connectionBusyTimeout = prop.get("queue.persistent.connectionBusyTimeout", 60000L);
      this.maxWaitingThreads = prop.get("queue.persistent.maxWaitingThreads", 200);
      // these should be handled by the JdbcManager
      this.tableAllocationIncrement = prop.get("queue.persistent.tableAllocationIncrement", 10);
      this.tableNamePrefix = prop.get("queue.persistent.tableNamePrefix", "XB").toUpperCase(); // XB stands for XMLBLASTER
      this.colNamePrefix = prop.get("queue.persistent.colNamePrefix", "").toUpperCase();

      // the property settings specific to this plugin type / version
      this.url = pluginProp.getProperty("url", this.url);
      String dbInstanceName = glob.getStrippedId();
      this.url = ReplaceVariable.replaceFirst(this.url, "$_{xmlBlaster_uniqueId}", dbInstanceName);
      ME = "JdbcConnectionPool-" + this.url;
      this.user = pluginProp.getProperty("user", this.user);
      this.password = pluginProp.getProperty("password", this.password);

      String txt = pluginProp.getProperty("debug", "" + false);
      try {
         this.debug = (new Boolean(txt)).booleanValue();
      }
      catch (Exception ex) {
         log.warning("The property 'debug' was set to '" + txt + "' which can not be parsed as a boolean value, will set it to false");
      }
      
      String help = pluginProp.getProperty("connectionPoolSize", "" + this.capacity);
      try {
         this.capacity = Integer.parseInt(help);
      }
      catch (Exception ex) {
         log.warning("the 'connectionPoolSize' plugin-property is not parseable: '" + help + "' will be using the default '" + this.capacity + "'");
      }
      if (this.capacity < MIN_POOL_SIZE) {
         log.warning("connectionPoolSize=" + this.capacity + " is too small, setting it to " + MIN_POOL_SIZE);
         this.capacity = MIN_POOL_SIZE;
      }

      help = pluginProp.getProperty("connectionBusyTimeout", "" + this.connectionBusyTimeout);
      try {
         this.connectionBusyTimeout = Long.parseLong(help);
      }
      catch (Exception ex) {
         log.warning("the 'connectionBusyTimeout' plugin-property is not parseable: '" + help + "' will be using the default '" + this.connectionBusyTimeout + "'");
      }

      help = pluginProp.getProperty("maxWaitingThreads", "" + this.maxWaitingThreads);
      try {
         this.maxWaitingThreads = Integer.parseInt(help);