jdbcmanagercommontable.java

java开源的企业总线.xmlBlaster

         /*
         if (false) {  // Postgres: 1 millis   Oracle: 2 millis
            if (this.pingPrepared == null) {
               //this.pingPrepared = conn.prepareStatement("SELECT count(nodeid) from " + this.nodesTableName);
               this.pingPrepared = conn.prepareStatement("SELECT nodeid from " + this.nodesTableName + " where nodeid='bla'");
            }
            org.xmlBlaster.util.StopWatch stopWatchToBlob = new org.xmlBlaster.util.StopWatch();
            this.pingPrepared.executeQuery();
            log.info(ME, "ping on Prepared select nodeid elapsed=" + stopWatchToBlob.nice());
         }
         {  // Postgres: 1 millis   Oracle: 4 millis
            org.xmlBlaster.util.StopWatch stopWatchToBlob = new org.xmlBlaster.util.StopWatch();
            Statement st = null;
            st = conn.createStatement();
            st.setQueryTimeout(this.pool.getQueryTimeout());
            st.execute("SELECT nodeid from " + this.nodesTableName + " where nodeid='bla'");// + this.tablesTxt);
            log.info(ME, "ping on select nodeid elapsed=" + stopWatchToBlob.nice());
         }
         {  // Postgres: 6 millis    Oracle: 9 millis
            org.xmlBlaster.util.StopWatch stopWatchToBlob = new org.xmlBlaster.util.StopWatch();
            ResultSet rs = conn.getMetaData().getTables("xyx", "xyz", "xyz", null);
            log.info(ME, "ping xy elapsed=" + stopWatchToBlob.nice());
         }
         {  // Postgres: 14 millis   Oracle: 2 sec 527
            org.xmlBlaster.util.StopWatch stopWatchToBlob = new org.xmlBlaster.util.StopWatch();
            conn.getMetaData().getTables(null, null, null, null);
            log.info(ME, "ping null elapsed=" + stopWatchToBlob.nice());
         }
         */
         if (log.isLoggable(Level.FINE)) log.fine("ping successful");
         return true;
      }
      catch (Throwable ex) {
         if (log.isLoggable(Level.FINE)) log.fine("ping to DB failed. DB may be down. Reason " + ex.toString());
         return false;
      }
/*
      finally {
         try {
            if (st != null) st.close();
         }
         catch (Throwable e) {
            log.warn(ME, "ping exception when closing the statement " + e.toString());
         }
      }
*/
   }

  /**
   * Adds (registers) a listener for connection/disconnection events.
   */
   public boolean registerStorageProblemListener(I_StorageProblemListener entry) {
      synchronized (this.listener) {
         this.listener.put(entry, DUMMY_VALUE);  // use DUMMY_VALUE to support check in unregisterListener()
      }
      return true;
   }

  /**
   * Adds (registers) a listener for connection/disconnection events.
   * @return boolean true if the entry was successfully removed.
   */
   public boolean unRegisterStorageProblemListener(I_StorageProblemListener entry) {
      if (entry == null) return false;
      synchronized (this.listener) {
         return this.listener.remove(entry) != null;
      }
   }

   /**
    * @see I_StorageProblemListener#storageUnavailable(int)
    */
   public void storageUnavailable(int oldStatus) {
      if (log.isLoggable(Level.FINER)) log.finer("storageUnavailable (old status '" + oldStatus + "')");
      this.isConnected = false;

      I_StorageProblemListener[] listenerArr = getStorageProblemListenerArr();
      for(int i=0; i<listenerArr.length; i++) {
         if (this.isConnected == true) break;
         I_StorageProblemListener singleListener = listenerArr[i];
         singleListener.storageUnavailable(oldStatus);
      }
   }

   /**
    * @see I_StorageProblemListener#storageAvailable(int)
    */
   public void storageAvailable(int oldStatus) {
      if (log.isLoggable(Level.FINER)) log.finer("storageAvailable (old status '" + oldStatus + "')");
      this.isConnected = true;
      //change this once this class implements I_StorageProblemNotifier
      if (oldStatus == I_StorageProblemListener.UNDEF) return;
      I_StorageProblemListener[] listenerArr = getStorageProblemListenerArr();
      for(int i=0; i<listenerArr.length; i++) {
         if (this.isConnected == false) break;
         I_StorageProblemListener singleListener = listenerArr[i];
         singleListener.storageAvailable(oldStatus);
      }
   }

   /**
    * @return A current snapshot of the connection listeners where we can work on (unsynchronized) and remove
    * listeners without danger
    */
   public I_StorageProblemListener[] getStorageProblemListenerArr() {
      synchronized (this.listener) {
         return (I_StorageProblemListener[])this.listener.keySet().toArray(new I_StorageProblemListener[this.listener.size()]);
      }
   }

   /**
    * @see #checkIfDBLoss(Connection, String, Throwable, String)
    */
   protected final boolean checkIfDBLoss(Connection conn, String location, Throwable ex) {
      return checkIfDBLoss(conn, location, ex, null);
   }

   /**
    * Handles the SQLException. 
    * If it is a communication exception like the
    * connection has been broken it will inform the connection pool.
    * @param location where the exception occured.
    * @param ex the exception which has to be handled.
    * @param trace additional information to put in the logetLogId(tableName, storageId, "getEntries") logging trace.
    * @return boolean true if it was a communication exception
    * 
    */
   protected final boolean checkIfDBLoss(Connection conn, String location, Throwable ex, String trace) {
      boolean ret = false;

      if (conn != null) ret = !ping(conn);
      else ret = !ping();

      if (ret) {
         log.severe(location + ": the connection to the DB has been lost. Going in polling modus");
         this.pool.setConnectionLost();
      }
      return ret;
   }

   /**
    * Gets the names of all the tables used by XmlBlaster. 
    * This information is retrieved via the database's metadata.
    * @param conn the connection on which to retrieve the metadata.
    * @return HashSet the set containing all the existing tablenames.
    */
   synchronized private HashSet getXbTableNames(Connection conn) throws SQLException {
      String[] types = { "TABLE" };
      ResultSet rs = conn.getMetaData().getTables(null, null, null, types);
      HashSet ret = new HashSet();
      while (rs.next()) { // retrieve the result set ...
         String table = rs.getString(3).toUpperCase();
         // if (table.startsWith(this.tablePrefix))
         // we currently add everything since I don't know what's better: speed here or when searching
         if (log.isLoggable(Level.FINE)) log.fine("getXbTableNames found table '" + table + "': adding to the set of found tables"); 
         ret.add(table);
      }
      return ret;
   }


   /**
    * Checks if all necessary tables exist. 
    * If a table does not exist and 'createTables' true, then the 
    * table is created.
    * @return boolean 'true' if the tables are all there after the invocation to this method, 'false' if at
    *         least one of the required tables is missing after the invocation to this method.
    */
   public final boolean tablesCheckAndSetup(boolean createTables)
      throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) log.finer("tablesCheckAndSetup");

      if (!this.isConnected)
         throw new XmlBlasterException(glob, ErrorCode.RESOURCE_DB_UNAVAILABLE, ME, "tablesCheckAndSetup: the DB disconnected. Handling is currently not possible");

      boolean entriesTableExists = false;

      Connection conn = null;
      boolean success = true;
      String req = "retrieving metadata";
      try {
         conn = this.pool.getConnection();
         conn.setAutoCommit(false);
         HashSet set = getXbTableNames(conn);

         if (set.contains(this.entriesTableName.toUpperCase())) entriesTableExists = true;
         if (log.isLoggable(Level.FINE)) log.fine("entries table exists : " + entriesTableExists);

         if (!createTables) return entriesTableExists;

         if (!entriesTableExists) {
            log.info("adding table '" + this.entriesTableName + "' as the 'entries' table");
            req = "CREATE TABLE " + this.entriesTableName.toUpperCase() + 
                  " (" + this.dataIdColName + " " + this.longintTxt + " " + this.keyAttr +
                  //", nodeId " + this.stringTxt + " " + this.keyAttr +
                  ", queueName " + this.stringTxt + " " + this.keyAttr +
                  ", prio " + this.intTxt +
                  ", flag " + this.stringTxt +
                  ", durable " + this.booleanTxt +
                  ", " + this.byteSizeColName + " " + this.longintTxt +
                  ", " + this.blobVarName + " " + this.blobTxt +
                  ", PRIMARY KEY (" + this.dataIdColName + ", queueName)";
            if (this.pool.isCascadeDeleteSuppported()) req  += " ON DELETE CASCADE)";
            else req += ")";
            if (log.isLoggable(Level.FINE)) 
               log.fine("Request: '" + req + "'");
            update(req, conn);
         }
         if (!conn.getAutoCommit()) 
            conn.commit();
         return true;
      }
      catch (XmlBlasterException ex) {
         success = false;
         try {
            if (!conn.getAutoCommit()) 
               conn.rollback();
         }
         catch (Throwable e) {
            log.severe("tablesCheckAndSetup: exception occured when rolling back: " + e.toString());
         }
         throw ex;
      }
      catch (Throwable ex) {
         success = false;
         try {
            if (conn != null && !conn.getAutoCommit()) 
               conn.rollback();
         }
         catch (Throwable e) {
            log.severe("tablesCheckAndSetup: exception occured when rolling back: " + e.toString());
         }

         if (checkIfDBLoss(conn, getLogId(null, "tablesCheckAndSetup"), ex, "SQL request giving problems: " + req))
            throw new XmlBlasterException(this.glob, ErrorCode.RESOURCE_DB_UNAVAILABLE, ME + ".tablesCheckAndSetup", "", ex); 
         else throw new XmlBlasterException(this.glob, ErrorCode.RESOURCE_DB_UNKNOWN, ME + ".tablesCheckAndSetup", "", ex); 
      }
      finally {
         if (conn != null) {
            try {
               conn.setAutoCommit(true);
            }
            catch (Throwable e) {
               log.severe("tablesCheckAndSetup: exception occured when setting back autocommit flag, reason: " + e.toString());
            }
            this.pool.releaseConnection(conn, success);
         }
      }
   }

   /**
    *
    * modifies a row in the specified queue table
    * @param queueName The name of the queue on which to perform the operation
    * @param entry the object to be stored.
    *
    * @return true on success
    *
    * @throws XmlBlasterException if an error occurred when trying to get a connection or an SQLException
    *         occurred.
    */
   public final boolean modifyEntry(String queueName, I_Entry entry)
      throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) log.finer("Entering");

      if (!this.isConnected) {
         if (log.isLoggable(Level.FINE)) log.fine("For entry '" + entry.getUniqueId() + "' currently not possible. No connection to the DB");
         throw new XmlBlasterException(this.glob, ErrorCode.RESOURCE_DB_UNAVAILABLE, ME + ".modifyEntry", " the connection to the DB is unavailable already before trying to add an entry"); 
      }

      Connection conn = null;
      boolean success = true;
      PreparedStatement preStatement = null;
      Statement exStatement = null;
      boolean ret = false;

      long dataId = entry.getUniqueId();
      int prio = entry.getPriority();
      byte[] blob = this.factory.toBlob(entry);
      String typeName = entry.getEmbeddedType();
      boolean persistent = entry.isPersistent();
      long sizeInBytes = entry.getSizeInBytes();
      
      if (log.isLoggable(Level.FINEST))
         log.finest("modification dataId: " + dataId + ", prio: " + prio + ", typeName: " + typeName + ", byteSize in bytes: " + sizeInBytes);

      try {
         conn = this.pool.getConnection();
         String req = "UPDATE " + this.entriesTableName + " SET prio = ? , flag = ? , durable = ? , " +
                      this.byteSizeColName + " = ? , " + this.blobVarName + " = ? WHERE  " + this.dataIdColName +
                      " = ? AND queueName = ?";

         if (log.isLoggable(Level.FINE)) log.fine(req);
         preStatement = conn.prepareStatement(req);
         preStatement.setQueryTimeout(this.pool.getQueryTimeout());
         preStatement.setInt(1, prio);
         preStatement.setString(2, typeName);
         if (persistent == true) preStatement.setString(3, "T");
         else preStatement.setString(3, "F");
         preStatement.setLong(4, sizeInBytes);
         
         ByteArrayInputStream blob_stream = new ByteArrayInputStream(blob);
         preStatement.setBinaryStream(5, blob_stream, blob.length); //(int)sizeInBytes);
         // preStatement.setBytes(5, blob);

         preStatement.setLong(6, dataId);
         preStatement.setString(7, queueName);

         if (log.isLoggable(Level.FINE)) log.fine(preStatement.toString());

         int num = preStatement.executeUpdate();
         if (log.isLoggable(Level.FINE)) log.fine("Modified " + num + " entries, entryId='" + entry.getUniqueId() + "'");
         ret = true;
      }
      catch (Throwable ex) {
         success = false;
         if (log.isLoggable(Level.FINE)) {
            if (ex instanceof SQLException) {
               log.fine("modifyEntry: sql exception, the sql state: '" + ((SQLException)ex).getSQLState() );
               log.fine("modifyEntry: sql exception, the error code: '" + ((SQLException)ex).getErrorCode() );