Skip to content

Commit 5bfc8f1

Browse files
author
chengyitian
committed
Merge branch 'dev' of dolphindb.net:dolphindb/api-java
2 parents 531fca7 + e0fd818 commit 5bfc8f1

File tree

7 files changed

+473
-61
lines changed

7 files changed

+473
-61
lines changed

‎pom.xml

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,11 @@
22
<modelVersion>4.0.0</modelVersion>
33
<groupId>com.dolphindb</groupId>
44
<artifactId>dolphindb-javaapi</artifactId>
5-
<version>3.00.1.2</version>
5+
<version>3.00.1.3</version>
66
<packaging>jar</packaging>
77

88
<properties>
9-
<dolphindb.version>3.00.1.2</dolphindb.version>
9+
<dolphindb.version>3.00.1.3</dolphindb.version>
1010
</properties>
1111
<name>DolphinDB Java API</name>
1212
<description>The messaging and data conversion protocol between Java and DolphinDB server</description>
@@ -31,7 +31,7 @@
3131
<connection>scm:git:git@github.com:dolphindb/api-java.git</connection>
3232
<developerConnection>scm:git:git@github.com:dolphindb/api-java.git</developerConnection>
3333
<url>git@github.com:dolphindb/api-java.git</url>
34-
<tag>api-java-3.00.1.2</tag>
34+
<tag>api-java-3.00.1.3</tag>
3535
</scm>
3636
<dependencies>
3737
<dependency>

‎src/com/xxdb/DBConnection.java

Lines changed: 122 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,7 @@
11
package com.xxdb;
22

33
import java.io.*;
4-
import java.net.ConnectException;
5-
import java.net.InetAddress;
6-
import java.net.InetSocketAddress;
7-
import java.net.Socket;
8-
import java.rmi.RemoteException;
4+
import java.net.*;
95
import java.security.PublicKey;
106
import java.security.cert.CertificateException;
117
import java.security.cert.X509Certificate;
@@ -50,6 +46,8 @@ public class DBConnection {
5046
private List<Node> nodes_ = new ArrayList<>();
5147
private Random nodeRandom_ = new Random();
5248
private int connTimeout_ = 0;
49+
private int connectTimeout_ = 0;
50+
private int readTimeout_ = 0;
5351
private boolean closed_ = false;
5452
private boolean loadBalance_ = false;
5553
private String runClientId_ = null;
@@ -66,7 +64,9 @@ private enum ExceptionType{
6664
ET_NEWLEADER(2),
6765
ET_NODENOTAVAIL(3),
6866
ET_NOINITIALIZED(4),
69-
ET_NOTLEADER(5);
67+
ET_NOTLEADER(5),
68+
ET_READTIMEDOUT(6),
69+
ET_NORESPONSEHEADER(7);
7070

7171
public int value;
7272
ExceptionType(int value){
@@ -147,6 +147,8 @@ private class DBConnectionImpl{
147147
private boolean compress_ = false;
148148
private boolean ifUrgent_ = false;
149149
private int connTimeout_ = 0;
150+
private int connectTimeout_ = 0;
151+
private int readTimeout_ = 0;
150152
private ExtendedDataInput in_;
151153
private ExtendedDataOutput out_;
152154
private boolean remoteLittleEndian_;
@@ -168,35 +170,49 @@ private DBConnectionImpl(boolean asynTask, boolean sslEnable, boolean compress,
168170
this.lock_ = new ReentrantLock();
169171
}
170172

171-
private boolean connect(String hostName, int port, String userId, String password, int connTimeout) throws IOException{
173+
private boolean connect(String hostName, int port, String userId, String password, int connTimeout, int connectTimeout, int readTimeout) throws IOException{
172174
this.hostName_ = hostName;
173175
this.port_ = port;
174176
this.userId_ = userId;
175177
this.pwd_ = password;
176178
this.connTimeout_ = connTimeout;
179+
this.connectTimeout_ = connectTimeout;
180+
this.readTimeout_ = readTimeout;
177181
return connect();
178182
}
179183

180-
private boolean connect()throws IOException{
184+
private boolean connect() throws IOException {
181185
this.isConnected_ = false;
182186

183187
try {
184-
if(sslEnable_)
188+
if (sslEnable_)
185189
socket_ = getSSLSocketFactory().createSocket();
186190
else
187191
socket_ = new Socket();
188-
if (this.connTimeout_ > 0){
189-
socket_.connect(new InetSocketAddress(hostName_,port_), connTimeout_);
190-
}else {
191-
socket_.connect(new InetSocketAddress(hostName_,port_), 3000);
192-
}
193-
} catch (ConnectException ex) {
192+
193+
// set 'connectTimeout' param to connect()
194+
if (this.connTimeout_ > 0 && this.connectTimeout_ == 0)
195+
socket_.connect(new InetSocketAddress(hostName_, port_), connTimeout_);
196+
else if (this.connTimeout_ > 0 && this.connectTimeout_ > 0)
197+
socket_.connect(new InetSocketAddress(hostName_, port_), connectTimeout_);
198+
else if (this.connTimeout_ == 0 && this.connectTimeout_ > 0)
199+
socket_.connect(new InetSocketAddress(hostName_, port_), connectTimeout_);
200+
else if (this.connTimeout_ == 0 && this.connectTimeout_ == 0)
201+
socket_.connect(new InetSocketAddress(hostName_, port_), 3000);
202+
} catch (IOException ex) {
194203
log.error("Connect to " + this.hostName_ + ":" + this.port_ + " failed.");
195204
throw ex;
196205
}
197-
if (this.connTimeout_ > 0) {
206+
207+
// set 'readTimeout' param to setSoTimeout
208+
if (this.connTimeout_ > 0 && this.readTimeout_ == 0)
198209
socket_.setSoTimeout(this.connTimeout_);
199-
}
210+
else if (this.connTimeout_ > 0 && this.readTimeout_ > 0)
211+
socket_.setSoTimeout(this.readTimeout_);
212+
else if (this.connTimeout_ == 0 && this.readTimeout_ > 0)
213+
socket_.setSoTimeout(this.readTimeout_);
214+
215+
200216
socket_.setKeepAlive(true);
201217
socket_.setTcpNoDelay(true);
202218
out_ = new LittleEndianDataOutputStream(new BufferedOutputStream(socket_.getOutputStream()));
@@ -441,9 +457,14 @@ private Entity run(String script, String scriptType, ProgressListener listener,
441457
header = in_.readLine();
442458
}
443459
}catch (IOException ex){
444-
isConnected_ = false;
445-
socket_ = null;
446-
throw new IOException("Failed to read response header from the socket with IO error " + ex.getMessage());
460+
if (ex instanceof SocketTimeoutException) {
461+
// isConnected_ = true;
462+
throw ex;
463+
} else {
464+
isConnected_ = false;
465+
socket_ = null;
466+
throw new IOException("Failed to read response header from the socket with IO error " + ex.getMessage());
467+
}
447468
}
448469

449470
String[] headers = header.split(" ");
@@ -664,16 +685,49 @@ public boolean connect(String hostName, int port, int timeout) throws IOExceptio
664685
return connect(hostName, port, "", "", null, false, null);
665686
}
666687

688+
public boolean connect(String hostName, int port, int connectTimeout, int readTimeout) throws IOException {
689+
if (connectTimeout < 0 || readTimeout < 0) {
690+
log.error("The param connectTimeout or readTimeout cannot less than zero.");
691+
return false;
692+
}
693+
694+
this.connectTimeout_ = connectTimeout;
695+
this.readTimeout_ = readTimeout;
696+
return connect(hostName, port, "", "", null, false, null);
697+
}
698+
667699
public boolean connect(String hostName, int port, int timeout, boolean reconnect) throws IOException {
668700
this.connTimeout_ = timeout;
669701
return connect(hostName, port, "", "", null, false, null, reconnect);
670702
}
671703

704+
public boolean connect(String hostName, int port, int connectTimeout, int readTimeout, boolean reconnect) throws IOException {
705+
if (connectTimeout < 0 || readTimeout < 0) {
706+
log.error("The param connectTimeout or readTimeout cannot less than zero.");
707+
return false;
708+
}
709+
710+
this.connectTimeout_ = connectTimeout;
711+
this.readTimeout_ = readTimeout;
712+
return connect(hostName, port, "", "", null, false, null, reconnect);
713+
}
714+
672715
public boolean connect(String hostName, int port, int timeout, boolean reconnect, int tryReconnectNums) throws IOException {
673716
this.connTimeout_ = timeout;
674717
return connect(hostName, port, "", "", null, false, null, reconnect, tryReconnectNums);
675718
}
676719

720+
public boolean connect(String hostName, int port, int connectTimeout, int readTimeout, boolean reconnect, int tryReconnectNums) throws IOException {
721+
if (connectTimeout < 0 || readTimeout < 0) {
722+
log.error("The param connectTimeout or readTimeout cannot less than zero.");
723+
return false;
724+
}
725+
726+
this.connectTimeout_ = connectTimeout;
727+
this.readTimeout_ = readTimeout;
728+
return connect(hostName, port, "", "", null, false, null, reconnect, tryReconnectNums);
729+
}
730+
677731
public boolean connect(String hostName, int port, String initialScript) throws IOException {
678732
return connect(hostName, port, "", "", initialScript, false, null);
679733
}
@@ -900,6 +954,7 @@ public boolean connect(String hostName, int port, String userId, String password
900954
}
901955
} else {
902956
if (reconnect) {
957+
nodes_.clear();
903958
nodes_.add(new Node(hostName, port));
904959
switchDataNode(new Node(hostName, port));
905960
} else {
@@ -938,7 +993,8 @@ public void switchDataNode(Node node) throws IOException{
938993
attempt ++;
939994
if (node.hostName != null && node.hostName.length() > 0) {
940995
if (connectNode(node)) {
941-
log.info("Switch to node: " + node.hostName + ":" + node.port + " successfully.");
996+
if (nodes_.size() > 1)
997+
log.info("Switch to node: " + node.hostName + ":" + node.port + " successfully.");
942998
isConnected = true;
943999
break;
9441000
}
@@ -974,12 +1030,12 @@ public void switchDataNode(Node node) throws IOException{
9741030
}
9751031
}
9761032

977-
public boolean connectNode(Node node) throws IOException{
1033+
public boolean connectNode(Node node) throws IOException {
9781034
log.info("Connect to " + node.hostName + ":" + node.port + ".");
9791035
while (!closed_){
9801036
try {
981-
return conn_.connect(node.hostName, node.port, uid_, pwd_, connTimeout_);
982-
}catch (Exception e){
1037+
return conn_.connect(node.hostName, node.port, uid_, pwd_, connTimeout_, connectTimeout_, readTimeout_);
1038+
} catch (Exception e) {
9831039
if (isConnected()){
9841040
Node tmpNode = new Node();
9851041
tmpNode.hostName = node.hostName;
@@ -993,11 +1049,13 @@ else if (type == ExceptionType.ET_NODENOTAVAIL)
9931049
else
9941050
throw e;
9951051
}
996-
}else {
997-
log.error(e.getMessage());
1052+
} else {
1053+
if (Objects.nonNull(e.getMessage()))
1054+
log.error(e.getMessage());
9981055
return false;
9991056
}
10001057
}
1058+
10011059
try {
10021060
Thread.sleep(100);
10031061
}catch (Exception e){
@@ -1008,8 +1066,7 @@ else if (type == ExceptionType.ET_NODENOTAVAIL)
10081066
return false;
10091067
}
10101068

1011-
public ExceptionType parseException(String msg, Node node){
1012-
log.info("com.xxdb.DBConnection.parseException msg: " + msg);
1069+
public ExceptionType parseException(String msg, Node node) {
10131070
if(msg==null){
10141071
node.hostName = "";
10151072
node.port = 0;
@@ -1042,7 +1099,13 @@ public ExceptionType parseException(String msg, Node node){
10421099
node.hostName = "";
10431100
node.port = 0;
10441101
return ExceptionType.ET_NOINITIALIZED;
1045-
}else {
1102+
} else if (msg.contains("Read timed out")) {
1103+
conn_.getNode(node);
1104+
return ExceptionType.ET_READTIMEDOUT;
1105+
} else if (msg.contains("Failed to read response header from the socket with IO error null")) {
1106+
conn_.getNode(node);
1107+
return ExceptionType.ET_NORESPONSEHEADER;
1108+
} else {
10461109
node.hostName = "";
10471110
node.port = 0;
10481111
return ExceptionType.ET_UNKNOW;
@@ -1184,9 +1247,14 @@ public Entity run(String script, ProgressListener listener, int priority, int pa
11841247
return new Void();
11851248
else if (type == ExceptionType.ET_UNKNOW)
11861249
throw e;
1187-
}else {
1188-
parseException(e.getMessage(), node);
1250+
else if (type == ExceptionType.ET_READTIMEDOUT)
1251+
cancelConsoleJob(enableSeqNo, currentSeqNo, e);
1252+
} else {
1253+
ExceptionType type = parseException(e.getMessage(), node);
1254+
if (type == ExceptionType.ET_READTIMEDOUT)
1255+
cancelConsoleJob(enableSeqNo, currentSeqNo, e);
11891256
}
1257+
11901258
switchDataNode(node);
11911259
}
11921260
}
@@ -1199,6 +1267,29 @@ else if (type == ExceptionType.ET_UNKNOW)
11991267
}
12001268
}
12011269

1270+
private void cancelConsoleJob(boolean enableSeqNo, long currentSeqNo, IOException e) throws IOException {
1271+
String cancelConsoleJobScript =
1272+
"jobs = exec rootJobId from getConsoleJobs() where sessionId = " + conn_.sessionID_ + "\n" +
1273+
(conn_.python_ ? "if size(jobs):\n" : "if (size(jobs))\n") +
1274+
" cancelConsoleJob(jobs)\n";
1275+
conn_.ifUrgent_ = true;
1276+
// conn_.asynTask_ = true;
1277+
1278+
if (enableSeqNo)
1279+
currentSeqNo = newSeqNo();
1280+
1281+
try {
1282+
conn_.run(cancelConsoleJobScript, currentSeqNo);
1283+
conn_.ifUrgent_ = false;
1284+
} catch (IOException ioe) {
1285+
conn_.ifUrgent_ = false;
1286+
throw new RuntimeException("Execute cancelConsoleJob failed after current connnection read timed out. ");
1287+
}
1288+
1289+
log.error(e.getMessage());
1290+
throw e;
1291+
}
1292+
12021293
public Entity run(String script, ProgressListener listener, int priority, int parallelism, int fetchSize, boolean clearSessionMemory, String tableName) throws IOException{
12031294
return run(script, listener, priority, parallelism, fetchSize, clearSessionMemory, tableName, true);
12041295
}

‎src/com/xxdb/data/Utils.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515

1616
public class Utils {
1717

18-
public static final String JAVA_API_VERSION = "3.00.1.2";
18+
public static final String JAVA_API_VERSION = "3.00.1.3";
1919

2020
public static final int DISPLAY_ROWS = 20;
2121
public static final int DISPLAY_COLS = 100;

0 commit comments

Comments
 (0)