Merge branch 'dev' of dolphindb.net:dolphindb/api-java

# Conflicts:
#	README.md

Author: chengyitian

README.md

@@ -111,15 +111,15 @@ boolean success = conn.connect("localhost", 8848, "admin", "123456");
 
 If the connection is established without a username and password, you only have guest privileges. To be granted with more privileges, we can log in by executing `conn.login('admin', '123456', true)`.
 
-To define and use user-defined functions in a Java program, you can pass in the user-defined scripts to the parameter initialScript. The advantages are:
-(1) These functions don't need to be defined repeatedly every time `run` is called;
+To define and use user-defined functions in a Java program, you can pass in the user-defined scripts to the parameter initialScript. The advantages are: 
+(1) These functions don't need to be defined repeatedly every time `run` is called; 
 (2) The API client can automatically connect to the server after disconnection. If the parameter *initialScript* is specified, the Java API will automatically execute the script and register the functions. The parameter can be very useful for scenarios where the network is not stable but the program needs to run continuously.
 
 ```java
 boolean success = conn.connect("localhost", 8848, "admin", "123456", "");
 ```
 
-To enable high availability, set the parameter *enableHighAvailability* to true.
+To enable high availability, set the parameter *enableHighAvailability* to true. 
 
 As of version 1.30.22.2, load balancing is automatically enabled for HA mode. Since 2.00.11.0, the `connect` method supports a new parameter *enableLoadBalance* which allows users to enable/disable load balancing in HA mode. Load balancing is only supported in HA mode and it is disabled by default.
 
@@ -233,7 +233,7 @@ The Java API provides connection pool `ExclusiveDBConnectionPool`. Users can exe
 | getConnectionCount()                                         | Get the number of connections.                               |
 | shutdown()                                                   | Shut down the connection pool.                               |
 
-**Note**: If the current `ExclusiveDBConnectionPool` is no longer in use, Java API will automatically close the connection after a while. To release the connection resources, call `shutdown()`upon the completion of thread tasks.
+**Note**: If the current `ExclusiveDBConnectionPool` is no longer in use, Java API will automatically close the connection after a while. To release the connection resources, call `shutdown()`upon the completion of thread tasks. 
 
 `BasicDBTask` wraps the functions and arguments to be executed.
 
@@ -563,7 +563,7 @@ public void testVoid() throws IOException{
 There are 2 types of DolphinDB tables:
 
 - In-memory table: it has the fastest access speed, but if the node shuts down the data will be lost.
-- DFS table: data are distributed across disks of multiple nodes.
+- DFS table: data are distributed across disks of multiple nodes. 
 
 ### 7.1. Write to an In-Memory Table
 
@@ -610,7 +610,7 @@ The example above uses partial application in DolphinDB to embed a table in `tab
 
 #### 7.1.3. Save BasicTable Objects With Function `tableInsert`
 
-Function `tableInsert` can also accept a BasicTable object in Java as a parameter to append data to a table in batches.
+Function `tableInsert` can also accept a BasicTable object in Java as a parameter to append data to a table in batches. 
 
 ```java
 public void test_save_table(BasicTable table1) throws IOException {
@@ -762,7 +762,7 @@ public void test_loop_basicTable(BasicTable table1) throws Exception{
 
 ### 7.4. Append Data Asynchronously
 
-You can use methods of `MultithreadedTableWriter` class to asynchronously append data to a DolphinDB in-memory table, dimension table, or a DFS table. The class maintains a buffer queue. Even when the server is fully occupied with network I/O operations, the writing threads of the API client will not be blocked.
+You can use methods of `MultithreadedTableWriter` class to asynchronously append data to a DolphinDB in-memory table, dimension table, or a DFS table. The class maintains a buffer queue. Even when the server is fully occupied with network I/O operations, the writing threads of the API client will not be blocked. 
 
 For asynchronous writes:
 
@@ -791,7 +791,7 @@ Parameters:
 - **port**: port number
 - **userId** / **password**: username and password
 - **dbPath**: a STRING indicating the DFS database path. Leave it unspecified for an in-memory table.
-- **tableName**: a STRING indicating the in-memory or DFS table name.
+- **tableName**: a STRING indicating the in-memory or DFS table name. 
 
 **Note:** For API 1.30.17 or lower versions, when writing to an in-memory table, please specify the in-memory table name for *dbPath* and leave *tableName* empty.
 
@@ -857,7 +857,7 @@ List<List<Entity>> unwrittenData = multithreadedTableWriter_.getUnwrittenData();
 ErrorCodeInfo insertUnwrittenData(List<List<Entity>> records)
 ```
 
-**Details:**
+**Details:** 
 
 Insert unwritten data. The result is in the same format as `insert`. The difference is that `insertUnwrittenData` can insert multiple records at a time.
 
@@ -877,7 +877,7 @@ ErrorCodeInfo ret = multithreadedTableWriter_.insertUnwrittenData(unwrittenData)
 Status getStatus()
 ```
 
-**Details:**
+**Details:** 
 
 Get the current status of the `MultithreadedTableWriter` object.
 
@@ -911,14 +911,14 @@ writeStatus = multithreadedTableWriter_.getStatus();
 - `hasError()`: return true if an error occurred, false otherwise.
 - `succeed()`: return true if the data is written successfully, false otherwise.
 
-
+ 
 (5) waitForThreadCompletion
 
 ```java
 waitForThreadCompletion()
 ```
 
-**Details:**
+**Details:** 
 
 After calling the method, `MultithreadedTableWriter` will wait until all working threads complete their tasks. If you call `insert` or `insertUnwrittenData` after the execution of `waitForThreadCompletion`, an error "thread is exiting" will be raised.
 
@@ -1000,7 +1000,7 @@ Output:
 The above example calls method `writer.insert()` to write data to writer, and obtains the status with `writer.getStatus()`. Please note that the method `writer.waitForThreadCompletion()` will wait for `MultithreadedTableWriter` to finish the data writes, and then terminate all working threads with the last status retained. A new MTW object must be created to write data again.
 
 As shown in the above example, `MultithreadedTableWriter` applies multiple threads to data conversion and writes. The API client also uses multiple threads to call `MultithreadedTableWriter`, and the implementation is thread-safe.
-
+ 
 #### 7.4.2. Exceptions Raised by MultithreadedTableWriter
 
 When calling method `insert` of class `MultithreadedTableWriter`:

pom.xml

@@ -2,11 +2,11 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.dolphindb</groupId>
     <artifactId>dolphindb-javaapi</artifactId>
-    <version>3.00.1.0</version>
+    <version>3.00.1.1</version>
     <packaging>jar</packaging>
 
     <properties>
-        <dolphindb.version>3.00.1.0</dolphindb.version>
+        <dolphindb.version>3.00.1.1</dolphindb.version>
     </properties>
     <name>DolphinDB Java API</name>
     <description>The messaging and data conversion protocol between Java and DolphinDB server</description>
@@ -31,7 +31,7 @@
         <connection>scm:git:git@github.com:dolphindb/api-java.git</connection>
         <developerConnection>scm:git:git@github.com:dolphindb/api-java.git</developerConnection>
         <url>git@github.com:dolphindb/api-java.git</url>
-        <tag>api-java-3.00.1.0</tag>
+        <tag>api-java-3.00.1.1</tag>
     </scm>
     <dependencies>
         <dependency>

src/com/xxdb/DBConnection.java

@@ -5,6 +5,7 @@
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.Socket;
+import java.rmi.RemoteException;
 import java.security.PublicKey;
 import java.security.cert.CertificateException;
 import java.security.cert.X509Certificate;
@@ -55,6 +56,7 @@ public class DBConnection {
     private long runSeqNo_ = 0;
     private int[] serverVersion_;
     private boolean isReverseStreaming_ = false;
+    private int tryReconnectNums = -1;
 
     private static final Logger log = LoggerFactory.getLogger(DBConnection.class);
 
@@ -189,7 +191,6 @@ private boolean connect()throws IOException{
                     socket_.connect(new InetSocketAddress(hostName_,port_), 3000);
                 }
             } catch (ConnectException ex) {
-                log.error("com.xxdb.DBConnection.DBConnectionImpl.connect() has exception. Current node hostName: " + this.hostName_ + ", port: " + this.port_);
                 log.error("Connect to " + this.hostName_ + ":" + this.port_ + " failed.");
                 throw ex;
             }
@@ -668,6 +669,11 @@ public boolean connect(String hostName, int port, int timeout, boolean reconnect
         return connect(hostName, port, "", "", null, false, null, reconnect);
     }
 
+    public boolean connect(String hostName, int port, int timeout, boolean reconnect, int tryReconnectNums) throws IOException {
+        this.connTimeout_ = timeout;
+        return connect(hostName, port, "", "", null, false, null, reconnect, tryReconnectNums);
+    }
+
     public boolean connect(String hostName, int port, String initialScript) throws IOException {
         return connect(hostName, port, "", "", initialScript, false, null);
     }
@@ -723,14 +729,30 @@ public boolean connect(String hostName, int port, String userId, String password
             return connect(hostName, port, userId, password, initialScript, enableHighAvailability, highAvailabilitySites, reconnect, false);
     }
 
+    public boolean connect(String hostName, int port, String userId, String password, String initialScript, boolean enableHighAvailability, String[] highAvailabilitySites, boolean reconnect, int tryReconnectNums) throws IOException {
+        if (enableHighAvailability)
+            return connect(hostName, port, userId, password, initialScript, enableHighAvailability, highAvailabilitySites, reconnect, true, tryReconnectNums);
+        else
+            return connect(hostName, port, userId, password, initialScript, enableHighAvailability, highAvailabilitySites, reconnect, false, tryReconnectNums);
+    }
+
     public boolean connect(String hostName, int port, String userId, String password, String initialScript, boolean enableHighAvailability, String[] highAvailabilitySites, boolean reconnect, boolean enableLoadBalance) throws IOException {
+        return connect(hostName, port, userId, password, initialScript, enableHighAvailability, highAvailabilitySites, reconnect, enableLoadBalance, -1);
+    }
+
+    public boolean connect(String hostName, int port, String userId, String password, String initialScript, boolean enableHighAvailability, String[] highAvailabilitySites, boolean reconnect, boolean enableLoadBalance, int tryReconnectNums) throws IOException {
         mutex_.lock();
         try {
             this.uid_ = userId;
             this.pwd_ = password;
             this.initialScript_ = initialScript;
             this.enableHighAvailability_ = enableHighAvailability;
             this.loadBalance_ = enableLoadBalance;
+            if (tryReconnectNums <= 0)
+                this.tryReconnectNums = -1;
+            else
+                this.tryReconnectNums = tryReconnectNums;
+
             if (this.loadBalance_ && !this.enableHighAvailability_)
                 throw new RuntimeException("Cannot only enable loadbalance but not enable highAvailablity.");
 
@@ -746,20 +768,48 @@ public boolean connect(String hostName, int port, String userId, String password
 
                 Node connectedNode = new Node();
                 BasicTable bt = null;
+
                 while (!closed_) {
+                    int totalConnectAttemptNums = this.tryReconnectNums * nodes_.size();
+                    int attempt = 0;
                     while (!conn_.isConnected() && !closed_) {
-                        for (Node one : nodes_) {
-                            if (connectNode(one)) {
-                                connectedNode = one;
-                                break;
+                        if (this.tryReconnectNums > 0) {
+                            // finite try to connect.
+                            for (Node one : nodes_) {
+                                attempt ++;
+                                // System.out.println("Current init connect node: " + one.hostName + ":" + one.port);
+                                if (connectNode(one)) {
+                                    connectedNode = one;
+                                    break;
+                                }
+
+                                try {
+                                    Thread.sleep(100);
+                                } catch (Exception e){
+                                    e.printStackTrace();
+                                    return false;
+                                }
                             }
 
-                            try {
-                                Thread.sleep(100);
-                            } catch (Exception e){
-                                e.printStackTrace();
+                            if (attempt == totalConnectAttemptNums) {
+                                log.error("Connect failed after " + tryReconnectNums + " reconnect attemps for every node in high availability sites.");
                                 return false;
                             }
+                        } else {
+                            // infinite try to connect.
+                            for (Node one : nodes_) {
+                                if (connectNode(one)) {
+                                    connectedNode = one;
+                                    break;
+                                }
+
+                                try {
+                                    Thread.sleep(100);
+                                } catch (Exception e){
+                                    e.printStackTrace();
+                                    return false;
+                                }
+                            }
                         }
                     }
 
@@ -882,30 +932,43 @@ private void initConnection() throws IOException{
     }
 
     public void switchDataNode(Node node) throws IOException{
+        int attempt = 0;
+        boolean isConnected = false;
         do {
-            if (node.hostName != null && node.hostName.length() > 0){
-                if (connectNode(node)){
+            attempt ++;
+            if (node.hostName != null && node.hostName.length() > 0) {
+                if (connectNode(node)) {
                     log.info("Switch to node: " + node.hostName + ":" + node.port + " successfully.");
+                    isConnected = true;
                     break;
                 }
             }
+
             if (nodes_.isEmpty()){
-                log.error("com.xxdb.DBConnection.switchDataNode nodes_ is empty. Current node hostName: " + node.hostName + ", port: " + node.port);
                 log.error("Connect to " + node.hostName + ":" + node.port + " failed.");
                 throw new RuntimeException("Connect to " + node.hostName + ":" + node.port + " failed.");
             }
-            int index = nodeRandom_.nextInt(nodes_.size());
-            if (connectNode(nodes_.get(index))){
-                log.info("Switch to node: " + nodes_.get(index).hostName + ":" + nodes_.get(index).port + " successfully.");
-                break;
+
+            if (nodes_.size() > 1) {
+                int index = nodeRandom_.nextInt(nodes_.size());
+                if (connectNode(nodes_.get(index))){
+                    log.info("Switch to node: " + nodes_.get(index).hostName + ":" + nodes_.get(index).port + " successfully.");
+                    isConnected = true;
+                    break;
+                }
             }
+
             try {
                 Thread.sleep(1000);
-            }catch (Exception e){
+            } catch (Exception e){
                 e.printStackTrace();
                 return;
             }
-        }while (!closed_);
+        } while (!closed_ && (tryReconnectNums == -1 || attempt < tryReconnectNums));
+
+        if (!closed_ && !isConnected)
+            throw new RuntimeException("Connect to " + node.hostName + ":" + node.port + " failed after " + attempt + " reconnect attemps.");
+
         if (initialScript_!=null && initialScript_.length() > 0){
             run(initialScript_);
         }
@@ -918,7 +981,10 @@ public boolean connectNode(Node node) throws IOException{
                 return conn_.connect(node.hostName, node.port, uid_, pwd_, connTimeout_);
             }catch (Exception e){
                 if (isConnected()){
-                    ExceptionType type = parseException(e.getMessage(), node);
+                    Node tmpNode = new Node();
+                    tmpNode.hostName = node.hostName;
+                    tmpNode.port = node.port;
+                    ExceptionType type = parseException(e.getMessage(), tmpNode);
                     if (type != ExceptionType.ET_NEWLEADER){
                         if (type == ExceptionType.ET_IGNORE)
                             return true;