IDEMPIERE-120 Refactor Query timeout Implementation - move most postgresql specific code to DB_PostgreSQL

IDEMPIERE-121 Add API to support the locking of PO for update
(transplanted from e99036fafdbdc294ebb0b847d675e0aac42c1843)

org.adempiere.base/src/org/compiere/db/AdempiereDatabase.java

@@ -25,6 +25,7 @@ import java.sql.Timestamp;
 import javax.sql.DataSource;
 
 import org.compiere.dbPort.Convert;
+import org.compiere.model.PO;
 
 //import org.compiere.util.CPreparedStatement;
 
@@ -36,6 +37,10 @@ import org.compiere.dbPort.Convert;
  */
 public interface AdempiereDatabase
 {
+	
+	/** default lock timeout, 60 seconds **/
+	static final int LOCK_TIME_OUT = 60;
+	
 	/**
 	 *  Get Database Name
 	 *  @return database short name
@@ -316,6 +321,23 @@ public interface AdempiereDatabase
 	 * @return
 	 */
 	public String addPagingSQL(String sql, int start, int end);
+	
+	/**
+	 * set statement/query timeout for connection
+	 * @param conn
+	 * @param timeout
+	 * @return original timeout setting
+	 * @throws SQLException
+	 */
+	public int setStatementTimeout(Connection conn, int timeout) throws SQLException;
+	
+	/**
+	 * Lock PO for update
+	 * @param po
+	 * @param timeout
+	 * @return true if lock is granted
+	 */
+	public boolean forUpdate(PO po, int timeout);
 
 }   //  AdempiereDatabase
 

org.adempiere.base/src/org/compiere/util/DB.java

@@ -1012,50 +1012,17 @@ public final class DB
 		CPreparedStatement cs = ProxyFactory.newCPreparedStatement(ResultSet.TYPE_FORWARD_ONLY,
 			ResultSet.CONCUR_UPDATABLE, sql, trxName);	//	converted in call
 
-		boolean autoCommit = false;
+		int currentTimeout = -1;
-		int currentTimeout = 0;
 		try
 		{
 			setParameters(cs, params);
-			autoCommit = cs.getConnection().getAutoCommit();
 			//set timeout
 			if (timeOut > 0)
 			{
 				if (DB.isPostgreSQL())
-				{
+					currentTimeout = DB.getDatabase().setStatementTimeout(cs.getConnection(), timeOut);				
-					try
-					{
-						Connection conn = cs.getConnection();
-						if (autoCommit)
-						{
-							conn.setAutoCommit(false);
-						}
-						else
-						{
-							ResultSet rs = null;
-							try
-							{
-								rs = conn.createStatement().executeQuery("select current_setting('statement_timeout')");
-								if (rs.next())
-									currentTimeout = rs.getInt(1);
-							}
-							finally
-							{
-								DB.close(rs);
-							}
-						}
-						Statement timeoutStatement = conn.createStatement();
-						timeoutStatement.execute("SET LOCAL statement_timeout TO " + ( timeOut * 1000 ));
-						if (log.isLoggable(Level.FINEST))
-						{
-							log.finest("Set statement timeout to " + timeOut);
-						}
-					} catch (SQLException e) {}
-				}
 				else
-				{
 					cs.setQueryTimeout(timeOut);
-				}
 			}
 			no = cs.executeUpdate();
 			//	No Transaction - Commit
@@ -1082,31 +1049,7 @@ public final class DB
 			{
 				try
 				{
-					if (autoCommit)
+					DB.getDatabase().setStatementTimeout(cs.getConnection(), currentTimeout);
-					{
-						cs.getConnection().setAutoCommit(true);
-					}
-					else
-					{
-						if (currentTimeout > 0)
-						{
-							Statement timeoutStatement = cs.getConnection().createStatement();
-							timeoutStatement.execute("SET LOCAL statement_timeout TO " + ( currentTimeout * 1000 ));
-							if (log.isLoggable(Level.FINEST))
-							{
-								log.finest("Reset statement timeout to " + currentTimeout);
-							}
-						}
-						else
-						{
-							Statement timeoutStatement = cs.getConnection().createStatement();
-							timeoutStatement.execute("SET LOCAL statement_timeout TO Default");
-							if (log.isLoggable(Level.FINEST))
-							{
-								log.finest("Reset statement timeout to default");
-							}
-						}
-					}
 				}
 				catch (SQLException e) {}
 			}

org.compiere.db.oracle.provider/src/org/compiere/db/DB_Oracle.java

@@ -24,6 +24,7 @@ import java.sql.DatabaseMetaData;
 import java.sql.Driver;
 import java.sql.DriverManager;
 import java.sql.PreparedStatement;
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.sql.Timestamp;
@@ -40,6 +41,7 @@ import org.adempiere.exceptions.DBException;
 import org.compiere.Adempiere;
 import org.compiere.dbPort.Convert;
 import org.compiere.dbPort.Convert_Oracle;
+import org.compiere.model.PO;
 import org.compiere.util.CLogger;
 import org.compiere.util.DB;
 import org.compiere.util.DisplayType;
@@ -1206,4 +1208,54 @@ public class DB_Oracle implements AdempiereDatabase
 		catch (Exception e) {}
 		return b;
 	}
+
+	@Override
+	public int setStatementTimeout(Connection conn, int timeout) throws SQLException {
+		//not supported by oracle
+		return -1;
+	}
+	
+	@Override
+	public boolean forUpdate(PO po, int timeout) {
+    	//only can lock for update if using trx
+    	if (po.get_TrxName() == null)
+    		return false;
+    	
+    	String[] keyColumns = po.get_KeyColumns();
+		if (keyColumns != null && keyColumns.length > 0 && !po.is_new()) {
+			StringBuffer sqlBuffer = new StringBuffer(" SELECT ");
+			sqlBuffer.append(keyColumns[0])
+				.append(" FROM ")
+				.append(po.get_TableName())
+				.append(" WHERE ");
+			for(int i = 0; i < keyColumns.length; i++) {
+				if (i > 0)
+					sqlBuffer.append(" AND ");
+				sqlBuffer.append(keyColumns[i]).append(" = ? ");
+			}
+			sqlBuffer.append(" FOR UPDATE ");
+			sqlBuffer.append(" WAIT ").append((timeout > 0 ? timeout : LOCK_TIME_OUT));
+			
+			PreparedStatement stmt = null;
+			ResultSet rs = null;
+			try {
+				stmt = DB.prepareStatement(sqlBuffer.toString(),
+					ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_UPDATABLE, po.get_TrxName());
+				for(int i = 0; i < keyColumns.length; i++) {
+					stmt.setObject(i+1, po.get_Value(keyColumns[i]));
+				}
+				rs = stmt.executeQuery();
+				if (rs.next()) {
+					return true;
+				} else {
+					return false;
+				}
+			} catch (Exception e) {
+				log.log(Level.INFO, e.getLocalizedMessage(), e);				
+			} finally {
+				DB.close(rs, stmt);
+			}			
+		}
+		return false;
+	}
 }   //  DB_Oracle

org.compiere.db.postgresql.provider/src/org/compiere/db/DB_PostgreSQL.java

@@ -23,6 +23,7 @@ import java.math.BigDecimal;
 import java.net.URL;
 import java.sql.Connection;
 import java.sql.DriverManager;
+import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
@@ -39,6 +40,7 @@ import org.adempiere.db.postgresql.PostgreSQLBundleActivator;
 import org.adempiere.exceptions.DBException;
 import org.compiere.dbPort.Convert;
 import org.compiere.dbPort.Convert_PostgreSQL;
+import org.compiere.model.PO;
 import org.compiere.util.CCache;
 import org.compiere.util.CLogger;
 import org.compiere.util.DB;
@@ -928,4 +930,78 @@ public class DB_PostgreSQL implements AdempiereDatabase
 		catch (Exception e) {}
 		return b;
 	}
+
+	@Override
+	public int setStatementTimeout(Connection conn, int timeOut) throws SQLException {
+		int currentTimeout = 0;		
+		boolean autoCommit = conn.getAutoCommit();
+		ResultSet rs = null;
+		try
+		{
+			rs = conn.createStatement().executeQuery("select current_setting('statement_timeout')");
+			if (rs.next())
+				currentTimeout = rs.getInt(1) / 1000;
+		}
+		finally
+		{
+			DB.close(rs);
+		}
+		
+		Statement timeoutStatement = conn.createStatement();
+		String sql = "SET " + (autoCommit ? "SESSION" : "LOCAL") + " statement_timeout TO " + ( timeOut > 0 ? Integer.toString(timeOut * 1000) : " DEFAULT ");
+		timeoutStatement.execute(sql);
+		if (log.isLoggable(Level.FINEST))
+		{
+			log.finest("Set statement timeout to " + timeOut);
+		}
+		return currentTimeout;
+	}
+	
+	@Override
+	public boolean forUpdate(PO po, int timeout) {
+    	//only can lock for update if using trx
+    	if (po.get_TrxName() == null)
+    		return false;
+    	
+    	String[] keyColumns = po.get_KeyColumns();
+		if (keyColumns != null && keyColumns.length > 0 && !po.is_new()) {
+			StringBuffer sqlBuffer = new StringBuffer(" SELECT ");
+			sqlBuffer.append(keyColumns[0])
+				.append(" FROM ")
+				.append(po.get_TableName())
+				.append(" WHERE ");
+			for(int i = 0; i < keyColumns.length; i++) {
+				if (i > 0)
+					sqlBuffer.append(" AND ");
+				sqlBuffer.append(keyColumns[i]).append(" = ? ");
+			}
+			sqlBuffer.append(" FOR UPDATE ");
+			PreparedStatement stmt = null;
+			ResultSet rs = null;
+			int currentTimeout = -1;
+			try {
+				stmt = DB.prepareStatement(sqlBuffer.toString(),
+					ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_UPDATABLE, po.get_TrxName());
+				for(int i = 0; i < keyColumns.length; i++) {
+					stmt.setObject(i+1, po.get_Value(keyColumns[i]));
+				}
+				currentTimeout = setStatementTimeout(stmt.getConnection(), (timeout > 0 ? timeout : LOCK_TIME_OUT));
+				
+				rs = stmt.executeQuery();
+				if (rs.next()) {
+					return true;
+				} else {
+					return false;
+				}
+			} catch (Exception e) {
+				log.log(Level.INFO, e.getLocalizedMessage(), e);				
+			} finally {
+				try {
+					setStatementTimeout(stmt.getConnection(), currentTimeout);
+				} catch (SQLException e) {}
+				DB.close(rs, stmt);
+			}			
+		}
+		return false;
+	}
 }   //  DB_PostgreSQL