This is an automated email from the ASF dual-hosted git repository. qiaojialin pushed a commit to branch fix_session_tsocket in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 6b22dfa4d483d2b51fc3655204ad1a50ec15bae8 Author: qiaojialin <[email protected]> AuthorDate: Tue Sep 17 11:49:01 2019 +0800 add reconnect and log in session --- .../main/java/org/apache/iotdb/session/Config.java | 3 + .../java/org/apache/iotdb/session/Session.java | 96 +++++++++++++++++----- 2 files changed, 80 insertions(+), 19 deletions(-) diff --git a/session/src/main/java/org/apache/iotdb/session/Config.java b/session/src/main/java/org/apache/iotdb/session/Config.java index 3bc2344..91564cd 100644 --- a/session/src/main/java/org/apache/iotdb/session/Config.java +++ b/session/src/main/java/org/apache/iotdb/session/Config.java @@ -23,4 +23,7 @@ public class Config { public static final String DEFAULT_USER = "user"; public static final String DEFAULT_PASSWORD = "password"; + public static final int RETRY_NUM = 3; + public static final long RETRY_INTERVAL = 1000; + public static int connectionTimeoutInMs = 0; } diff --git a/session/src/main/java/org/apache/iotdb/session/Session.java b/session/src/main/java/org/apache/iotdb/session/Session.java index 50e00fa..6417aa5 100644 --- a/session/src/main/java/org/apache/iotdb/session/Session.java +++ b/session/src/main/java/org/apache/iotdb/session/Session.java @@ -18,9 +18,11 @@ */ package org.apache.iotdb.session; +import java.sql.SQLException; import java.util.List; import org.apache.iotdb.rpc.IoTDBRPCException; import org.apache.iotdb.rpc.RpcUtils; +import org.apache.iotdb.rpc.TSStatusType; import org.apache.iotdb.service.rpc.thrift.*; import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; @@ -34,9 +36,12 @@ import org.apache.thrift.transport.TSocket; import org.apache.thrift.transport.TTransportException; import java.time.ZoneId; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class Session { + private static final Logger logger = LoggerFactory.getLogger(Session.class); private String host; private int port; private String username; @@ -47,6 +52,7 @@ public class Session { private TSocket transport; private boolean isClosed = true; private ZoneId zoneId; + private boolean enableRPCCompression; public Session(String host, int port) { this(host, port, Config.DEFAULT_USER, Config.DEFAULT_PASSWORD); @@ -64,23 +70,62 @@ public class Session { } public synchronized void open() throws IoTDBSessionException { - open(false, 0); + open(false); } - public synchronized void open(boolean enableRPCCompression, int connectionTimeoutInMs) + public synchronized void open(boolean enableRPCCompression) throws IoTDBSessionException { if (!isClosed) { return; } - transport = new TSocket(host, port, connectionTimeoutInMs); - if (!transport.isOpen()) { + + this.enableRPCCompression = enableRPCCompression; + + try { + openTransport(); + } catch (TTransportException e) { + throw new IoTDBSessionException(e); + } + openSession(); + isClosed = false; + client = RpcUtils.newSynchronizedClient(client); + + } + + public boolean reconnect() throws IoTDBSessionException { + boolean flag = false; + for (int i = 1; i <= Config.RETRY_NUM; i++) { try { - transport.open(); - } catch (TTransportException e) { - throw new IoTDBSessionException(e); + if (transport != null) { + transport.close(); + openTransport(); + openSession(); + client = RpcUtils.newSynchronizedClient(client); + flag = true; + break; + } + } catch (Exception e) { + try { + Thread.sleep(Config.RETRY_INTERVAL); + } catch (InterruptedException e1) { + logger.error("reconnect is interrupted.", e1); + } } } + if (!flag) { + throw new IoTDBSessionException("Fail to reconnect to server. please check server status"); + } + return flag; + } + private void openTransport() throws TTransportException { + transport = new TSocket(host, port, Config.connectionTimeoutInMs); + if (!transport.isOpen()) { + transport.open(); + } + } + + private void openSession() throws IoTDBSessionException { if(enableRPCCompression) { client = new TSIService.Client(new TCompactProtocol(transport)); } @@ -89,20 +134,26 @@ public class Session { } TSOpenSessionReq openReq = new TSOpenSessionReq(TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V1); + openReq.setUsername(username); openReq.setPassword(password); try { TSOpenSessionResp openResp = client.openSession(openReq); - RpcUtils.verifySuccess(openResp.getStatus()); - + // validate connection + try { + RpcUtils.verifySuccess(openResp.getStatus()); + } catch (IoTDBRPCException e) { + // failed to connect, disconnect from the server + transport.close(); + throw new IoTDBSessionException(e.getMessage()); + } if (protocolVersion.getValue() != openResp.getServerProtocolVersion().getValue()) { throw new TException(String .format("Protocol not supported, Client version is {}, but Server version is {}", protocolVersion.getValue(), openResp.getServerProtocolVersion().getValue())); } - sessionHandle = openResp.getSessionHandle(); if (zoneId != null) { @@ -112,14 +163,10 @@ public class Session { } } catch (TException | IoTDBRPCException e) { - transport.close(); - throw new IoTDBSessionException(String.format("Can not open session to %s:%s with user: %s.", - host, port, username), e); + throw new IoTDBSessionException(String.format("Can not establish connection to %s.", + host), e); } isClosed = false; - - client = RpcUtils.newSynchronizedClient(client); - } public synchronized void close() throws IoTDBSessionException { @@ -153,6 +200,7 @@ public class Session { try { return client.insertBatch(request); } catch (TException e) { + reconnect(); throw new IoTDBSessionException(e); } } @@ -166,8 +214,9 @@ public class Session { request.setValues(values); try { - return client.insertRow(request); + return checkAndReturn(client.insertRow(request)); } catch (TException e) { + reconnect(); throw new IoTDBSessionException(e); } } @@ -177,8 +226,9 @@ public class Session { request.setStorageGroupId(storageGroupId); try { - return client.setStorageGroup(request); + return checkAndReturn(client.setStorageGroup(request)); } catch (TException e) { + reconnect(); throw new IoTDBSessionException(e); } } @@ -191,12 +241,20 @@ public class Session { request.setCompressor(compressor.ordinal()); try { - return client.createTimeseries(request); + return checkAndReturn(client.createTimeseries(request)); } catch (TException e) { + reconnect(); throw new IoTDBSessionException(e); } } + private TSRPCResp checkAndReturn(TSRPCResp resp) { + if (resp.status.statusType.getCode() != TSStatusType.SUCCESS_STATUS.getStatusCode()) { + logger.error(resp.status.statusType.getMessage()); + } + return resp; + } + public synchronized String getTimeZone() throws TException, IoTDBRPCException { if (zoneId != null) { return zoneId.toString();
