JAVAOS-1340: support 'enableSeqNo' for MTW、ExclusiveDBConnectionPool; add 'setDBConnection(DBConnection conn, boolean enableSeqNo)' for DBTask;

Author: chengyitian

src/com/xxdb/BasicDBTask.java

@@ -9,6 +9,7 @@
 
 public class BasicDBTask implements DBTask {
 	private String script;
+	private boolean enableSeqNo = true;
 	private List<Entity> args;
 	private DBConnection conn;
 	private Entity result = null;
@@ -17,6 +18,9 @@ public class BasicDBTask implements DBTask {
 	private CountDownLatch latch;
 	private int timeOut = -1;
 
+	private static final int DEFAULT_PRIORITY = 4;
+	private static final int DEFAULT_PARALLELISM = 64;
+
 	public BasicDBTask(String script, List<Entity> args) {
 		this.script = script;
 		this.args = args;
@@ -30,10 +34,12 @@ public BasicDBTask(String script) {
 	@Override
 	public Entity call() {
 		try {
-			if (args != null)
-				result = conn.run(script, args);
-			else
-				result = conn.run(script);
+			if (args != null) {
+				result = conn.run(script, args, DEFAULT_PRIORITY, DEFAULT_PARALLELISM, 0, enableSeqNo);
+			} else {
+				result = conn.run(script, null, DEFAULT_PRIORITY, DEFAULT_PARALLELISM, 0, false, "", enableSeqNo);
+			}
+
 			errMsg = null;
 			synchronized (this) {
 				status = TaskStatus.SUCCESS;
@@ -55,6 +61,12 @@ public void setDBConnection(DBConnection conn) {
 		this.conn = conn;
 	}
 
+	@Override
+	public void setDBConnection(DBConnection conn, boolean enableSeqNo) {
+		this.conn = conn;
+		this.enableSeqNo = enableSeqNo;
+	}
+
 	@Override
 	public Entity getResult() {
 		if (status != TaskStatus.SUCCESS) {

src/com/xxdb/DBTask.java

@@ -5,6 +5,7 @@
 
 public interface DBTask extends Callable<Entity>{
 	void setDBConnection(DBConnection conn);
+	void setDBConnection(DBConnection conn, boolean enableSeqNo);
 	Entity getResult();
 	String getErrorMsg();
 	String getScript();

src/com/xxdb/ExclusiveDBConnectionPool.java

@@ -20,10 +20,12 @@ public class ExclusiveDBConnectionPool implements DBConnectionPool {
 
 	private class AsyncWorker implements Runnable {
 		private DBConnection conn_;
+		boolean enableSeqNo = true;
 		private final Thread workThread_;
 
-		public AsyncWorker(DBConnection conn) {
+		public AsyncWorker(DBConnection conn, boolean enableSeqNo) {
 			this.conn_ = conn;
+			this.enableSeqNo = enableSeqNo;
 			workThread_ = new Thread(this);
 			workThread_.start();
 		}
@@ -49,7 +51,7 @@ public void run() {
 						break;
 					}
 					try {
-						task.setDBConnection(conn_);
+						task.setDBConnection(conn_, enableSeqNo);
 						task.call();
 					} catch (InterruptedException e) {
 						 break;
@@ -71,10 +73,14 @@ public void run() {
 	}
 
 	public ExclusiveDBConnectionPool(String host, int port, String uid, String pwd, int count, boolean loadBalance, boolean enableHighAvailability) throws IOException {
-		this(host, port, uid, pwd, count, loadBalance, enableHighAvailability, null, "",false, false, false);
+		this(host, port, uid, pwd, count, loadBalance, enableHighAvailability, null, "",false, false, false, true);
 	}
 
 	public ExclusiveDBConnectionPool(String host, int port, String uid, String pwd, int count, boolean loadBalance, boolean enableHighAvailability, String[] highAvailabilitySites, String initialScript,boolean compress, boolean useSSL, boolean usePython) throws IOException {
+		this(host, port, uid, pwd, count, loadBalance, enableHighAvailability, highAvailabilitySites, initialScript, compress, useSSL, usePython, true);
+	}
+
+	public ExclusiveDBConnectionPool(String host, int port, String uid, String pwd, int count, boolean loadBalance, boolean enableHighAvailability, String[] highAvailabilitySites, String initialScript,boolean compress, boolean useSSL, boolean usePython, boolean enableSeqNo) throws IOException {
 		if (count <= 0)
 			throw new RuntimeException("The thread count can not be less than 0");
 		if (!loadBalance) {
@@ -90,7 +96,7 @@ public ExclusiveDBConnectionPool(String host, int port, String uid, String pwd,
 					throw new RuntimeException("Can't connect to the specified host: ", e);
 				}
 
-				workers_.add(new AsyncWorker(conn));
+				workers_.add(new AsyncWorker(conn, enableSeqNo));
 			}
 		} else {
 			BasicStringVector nodes = null;
@@ -117,7 +123,7 @@ public ExclusiveDBConnectionPool(String host, int port, String uid, String pwd,
 				DBConnection conn = new DBConnection(false, useSSL, compress, usePython);
 				if(!conn.connect(hosts[i % nodeCount], ports[i % nodeCount], uid, pwd, initialScript, enableHighAvailability, highAvailabilitySites,false,false))
 					throw new RuntimeException("Can't connect to the host " + nodes.getString(i));
-				workers_.add(new AsyncWorker(conn));
+				workers_.add(new AsyncWorker(conn, enableSeqNo));
 			}
 		}
 	}