This is an automated email from the ASF dual-hosted git repository.
wangchao316 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new f35a087 [IOTDB-1463] Implement builder pattern for Session and
SessionPool (#3502)
f35a087 is described below
commit f35a087bcf3c1bb5f4f9fbca4eea7d6541d3247e
Author: J.J. Liu <[email protected]>
AuthorDate: Wed Jul 7 09:09:16 2021 +0800
[IOTDB-1463] Implement builder pattern for Session and SessionPool (#3502)
* Implement builder pattern for Session and SessionPool
* Modify tests
* Fix
* Change ut name
* Change default params for SessionPol
* Make more params optional
* Fix test bug
* Add docs
* Fix spotless
* Remove zoneId
* Remove zoneId
* Remove zoneId in docs
---
docs/UserGuide/API/Programming-Java-Native-API.md | 20 ++-
.../UserGuide/API/Programming-Java-Native-API.md | 18 ++-
.../main/java/org/apache/iotdb/SessionExample.java | 3 +-
.../java/org/apache/iotdb/SessionPoolExample.java | 9 +-
.../main/java/org/apache/iotdb/session/Config.java | 4 +
.../java/org/apache/iotdb/session/Session.java | 65 ++++++++++
.../org/apache/iotdb/session/pool/SessionPool.java | 142 ++++++++++++++++++---
.../session/{SessionUT.java => SessionTest.java} | 25 +++-
.../apache/iotdb/session/pool/SessionPoolTest.java | 26 ++++
9 files changed, 284 insertions(+), 28 deletions(-)
diff --git a/docs/UserGuide/API/Programming-Java-Native-API.md
b/docs/UserGuide/API/Programming-Java-Native-API.md
index a09a067..bfa2079 100644
--- a/docs/UserGuide/API/Programming-Java-Native-API.md
+++ b/docs/UserGuide/API/Programming-Java-Native-API.md
@@ -56,11 +56,21 @@ Here we show the commonly used interfaces and their
parameters in the Native API
* Initialize a Session
```java
-Session(String host, int rpcPort)
-
-Session(String host, String rpcPort, String username, String password)
-
-Session(String host, int rpcPort, String username, String password)
+ // use default configuration
+ session = new Session.Builder.build();
+
+ // configure all fields
+ session =
+ new Session.Builder()
+ .host(String host)
+ .port(int port)
+ .fetchSize(int fetchSize)
+ .username(String username)
+ .password(String password)
+ .thriftDefaultBufferSize(int thriftDefaultBufferSize)
+ .thriftMaxFrameSize(int thriftMaxFrameSize)
+ .enableCacheLeader(boolean enableCacheLeader)
+ .build();
```
* Open a Session
diff --git a/docs/zh/UserGuide/API/Programming-Java-Native-API.md
b/docs/zh/UserGuide/API/Programming-Java-Native-API.md
index db91d13..7bc6cab 100644
--- a/docs/zh/UserGuide/API/Programming-Java-Native-API.md
+++ b/docs/zh/UserGuide/API/Programming-Java-Native-API.md
@@ -59,9 +59,21 @@ mvn clean install -pl session -am -Dmaven.test.skip=true
* 初始化Session
```java
-Session(String host, int rpcPort)
-Session(String host, String rpcPort, String username, String password)
-Session(String host, int rpcPort, String username, String password)
+ // 全部使用默认配置
+ session = new Session.Builder.build();
+
+ // 自行配置参数
+ session =
+ new Session.Builder()
+ .host(String host)
+ .port(int port)
+ .fetchSize(int fetchSize)
+ .username(String username)
+ .password(String password)
+ .thriftDefaultBufferSize(int thriftDefaultBufferSize)
+ .thriftMaxFrameSize(int thriftMaxFrameSize)
+ .enableCacheLeader(boolean enableCacheLeader)
+ .build();
```
* 开启Session
diff --git a/example/session/src/main/java/org/apache/iotdb/SessionExample.java
b/example/session/src/main/java/org/apache/iotdb/SessionExample.java
index a705c24..a8294f1 100644
--- a/example/session/src/main/java/org/apache/iotdb/SessionExample.java
+++ b/example/session/src/main/java/org/apache/iotdb/SessionExample.java
@@ -53,7 +53,8 @@ public class SessionExample {
public static void main(String[] args)
throws IoTDBConnectionException, StatementExecutionException {
- session = new Session(LOCAL_HOST, 6667, "root", "root");
+ session =
+ new
Session.Builder().host(LOCAL_HOST).port(6667).username("root").password("root").build();
session.open(false);
// set session fetchSize
diff --git
a/example/session/src/main/java/org/apache/iotdb/SessionPoolExample.java
b/example/session/src/main/java/org/apache/iotdb/SessionPoolExample.java
index d2e4977..230849d 100644
--- a/example/session/src/main/java/org/apache/iotdb/SessionPoolExample.java
+++ b/example/session/src/main/java/org/apache/iotdb/SessionPoolExample.java
@@ -37,7 +37,14 @@ public class SessionPoolExample {
public static void main(String[] args)
throws StatementExecutionException, IoTDBConnectionException,
InterruptedException {
- pool = new SessionPool("127.0.0.1", 6667, "root", "root", 3);
+ pool =
+ new SessionPool.Builder()
+ .host("127.0.0.1")
+ .port(6667)
+ .user("root")
+ .password("root")
+ .maxSize(3)
+ .build();
service = Executors.newFixedThreadPool(10);
insertRecord();
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 02e7e70..7b1a636 100644
--- a/session/src/main/java/org/apache/iotdb/session/Config.java
+++ b/session/src/main/java/org/apache/iotdb/session/Config.java
@@ -20,6 +20,8 @@ package org.apache.iotdb.session;
public class Config {
+ public static final String DEFAULT_HOST = "localhost";
+ public static final int DEFAULT_PORT = 6667;
public static final String DEFAULT_USER = "root";
public static final String DEFAULT_PASSWORD = "root";
public static final int DEFAULT_FETCH_SIZE = 5000;
@@ -34,4 +36,6 @@ public class Config {
/** thrift max frame size (16384000 bytes by default), we change it to 64MB
*/
public static final int DEFAULT_MAX_FRAME_SIZE = 67108864;
+
+ public static final int DEFAULT_SESSION_POOL_MAX_SIZE = 5;
}
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 1b2f812..cfe11a7 100644
--- a/session/src/main/java/org/apache/iotdb/session/Session.java
+++ b/session/src/main/java/org/apache/iotdb/session/Session.java
@@ -1839,4 +1839,69 @@ public class Session {
public void setEnableQueryRedirection(boolean enableQueryRedirection) {
this.enableQueryRedirection = enableQueryRedirection;
}
+
+ public static class Builder {
+ private String host = Config.DEFAULT_HOST;
+ private int rpcPort = Config.DEFAULT_PORT;
+ private String username = Config.DEFAULT_USER;
+ private String password = Config.DEFAULT_PASSWORD;
+ private int fetchSize = Config.DEFAULT_FETCH_SIZE;
+ private ZoneId zoneId = null;
+ private int thriftDefaultBufferSize =
Config.DEFAULT_INITIAL_BUFFER_CAPACITY;
+ private int thriftMaxFrameSize = Config.DEFAULT_MAX_FRAME_SIZE;
+ private boolean enableCacheLeader = Config.DEFAULT_CACHE_LEADER_MODE;
+
+ public Builder host(String host) {
+ this.host = host;
+ return this;
+ }
+
+ public Builder port(int port) {
+ this.rpcPort = port;
+ return this;
+ }
+
+ public Builder username(String username) {
+ this.username = username;
+ return this;
+ }
+
+ public Builder password(String password) {
+ this.password = password;
+ return this;
+ }
+
+ public Builder fetchSize(int fetchSize) {
+ this.fetchSize = fetchSize;
+ return this;
+ }
+
+ public Builder thriftDefaultBufferSize(int thriftDefaultBufferSize) {
+ this.thriftDefaultBufferSize = thriftDefaultBufferSize;
+ return this;
+ }
+
+ public Builder thriftMaxFrameSize(int thriftMaxFrameSize) {
+ this.thriftMaxFrameSize = thriftMaxFrameSize;
+ return this;
+ }
+
+ public Builder enableCacheLeader(boolean enableCacheLeader) {
+ this.enableCacheLeader = enableCacheLeader;
+ return this;
+ }
+
+ public Session build() {
+ return new Session(
+ host,
+ rpcPort,
+ username,
+ password,
+ fetchSize,
+ zoneId,
+ thriftDefaultBufferSize,
+ thriftMaxFrameSize,
+ enableCacheLeader);
+ }
+ }
}
diff --git
a/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
b/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
index 79a539d..fe8cb25 100644
--- a/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
+++ b/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
@@ -70,7 +70,7 @@ public class SessionPool {
private ConcurrentMap<Session, Session> occupied = new ConcurrentHashMap<>();
private int size = 0;
private int maxSize = 0;
- private String ip;
+ private String host;
private int port;
private String user;
private String password;
@@ -83,9 +83,9 @@ public class SessionPool {
private boolean closed; // whether the queue is closed.
- public SessionPool(String ip, int port, String user, String password, int
maxSize) {
+ public SessionPool(String host, int port, String user, String password, int
maxSize) {
this(
- ip,
+ host,
port,
user,
password,
@@ -98,9 +98,9 @@ public class SessionPool {
}
public SessionPool(
- String ip, int port, String user, String password, int maxSize, boolean
enableCompression) {
+ String host, int port, String user, String password, int maxSize,
boolean enableCompression) {
this(
- ip,
+ host,
port,
user,
password,
@@ -113,7 +113,7 @@ public class SessionPool {
}
public SessionPool(
- String ip,
+ String host,
int port,
String user,
String password,
@@ -121,7 +121,7 @@ public class SessionPool {
boolean enableCompression,
boolean enableCacheLeader) {
this(
- ip,
+ host,
port,
user,
password,
@@ -134,9 +134,9 @@ public class SessionPool {
}
public SessionPool(
- String ip, int port, String user, String password, int maxSize, ZoneId
zoneId) {
+ String host, int port, String user, String password, int maxSize, ZoneId
zoneId) {
this(
- ip,
+ host,
port,
user,
password,
@@ -150,7 +150,7 @@ public class SessionPool {
@SuppressWarnings("squid:S107")
public SessionPool(
- String ip,
+ String host,
int port,
String user,
String password,
@@ -161,7 +161,7 @@ public class SessionPool {
ZoneId zoneId,
boolean enableCacheLeader) {
this.maxSize = maxSize;
- this.ip = ip;
+ this.host = host;
this.port = port;
this.user = user;
this.password = password;
@@ -198,9 +198,9 @@ public class SessionPool {
if (canCreate) {
// create a new one.
if (logger.isDebugEnabled()) {
- logger.debug("Create a new Session {}, {}, {}, {}", ip, port, user,
password);
+ logger.debug("Create a new Session {}, {}, {}, {}", host, port,
user, password);
}
- session = new Session(ip, port, user, password, fetchSize, zoneId,
enableCacheLeader);
+ session = new Session(host, port, user, password, fetchSize, zoneId,
enableCacheLeader);
try {
session.open(enableCompression);
// avoid someone has called close() the session pool
@@ -244,7 +244,7 @@ public class SessionPool {
logger.warn(
"the SessionPool has wait for {} seconds to get a new
connection: {}:{} with {}, {}",
(System.currentTimeMillis() - start) / 1000,
- ip,
+ host,
port,
user,
password);
@@ -255,7 +255,7 @@ public class SessionPool {
size);
if (System.currentTimeMillis() - start > timeout) {
throw new IoTDBConnectionException(
- String.format("timeout to get a connection from %s:%s",
ip, port));
+ String.format("timeout to get a connection from %s:%s",
host, port));
}
}
} catch (InterruptedException e) {
@@ -337,7 +337,7 @@ public class SessionPool {
@SuppressWarnings({"squid:S2446"})
private synchronized void removeSession() {
- logger.warn("Remove a broken Session {}, {}, {}", ip, port, user);
+ logger.warn("Remove a broken Session {}, {}, {}", host, port, user);
size--;
// we do not need to notifyAll as any waited thread can continue to work
after waked up.
this.notify();
@@ -365,7 +365,7 @@ public class SessionPool {
throw new IoTDBConnectionException(
String.format(
"retry to execute statement on %s:%s failed %d times: %s",
- ip, port, RETRY, e.getMessage()),
+ host, port, RETRY, e.getMessage()),
e);
}
}
@@ -1161,4 +1161,112 @@ public class SessionPool {
// never go here
return null;
}
+
+ public int getMaxSize() {
+ return maxSize;
+ }
+
+ public String getHost() {
+ return host;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public String getUser() {
+ return user;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ public int getFetchSize() {
+ return fetchSize;
+ }
+
+ public long getTimeout() {
+ return timeout;
+ }
+
+ public boolean isEnableCompression() {
+ return enableCompression;
+ }
+
+ public boolean isEnableCacheLeader() {
+ return enableCacheLeader;
+ }
+
+ public static class Builder {
+ private String host = Config.DEFAULT_HOST;
+ private int port = Config.DEFAULT_PORT;
+ private int maxSize = Config.DEFAULT_SESSION_POOL_MAX_SIZE;
+ private String user = Config.DEFAULT_USER;
+ private String password = Config.DEFAULT_PASSWORD;
+ private int fetchSize = Config.DEFAULT_FETCH_SIZE;
+ private long timeout = 60_000;
+ private boolean enableCompression = false;
+ private ZoneId zoneId = null;
+ private boolean enableCacheLeader = Config.DEFAULT_CACHE_LEADER_MODE;
+
+ public Builder host(String host) {
+ this.host = host;
+ return this;
+ }
+
+ public Builder port(int port) {
+ this.port = port;
+ return this;
+ }
+
+ public Builder maxSize(int maxSize) {
+ this.maxSize = maxSize;
+ return this;
+ }
+
+ public Builder user(String user) {
+ this.user = user;
+ return this;
+ }
+
+ public Builder password(String password) {
+ this.password = password;
+ return this;
+ }
+
+ public Builder fetchSize(int fetchSize) {
+ this.fetchSize = fetchSize;
+ return this;
+ }
+
+ public Builder timeout(long timeout) {
+ this.timeout = timeout;
+ return this;
+ }
+
+ public Builder enableCompression(boolean enableCompression) {
+ this.enableCompression = enableCompression;
+ return this;
+ }
+
+ public Builder enableCacheLeader(boolean enableCacheLeader) {
+ this.enableCacheLeader = enableCacheLeader;
+ return this;
+ }
+
+ public SessionPool build() {
+ return new SessionPool(
+ host,
+ port,
+ user,
+ password,
+ maxSize,
+ fetchSize,
+ timeout,
+ enableCompression,
+ zoneId,
+ enableCacheLeader);
+ }
+ }
}
diff --git a/session/src/test/java/org/apache/iotdb/session/SessionUT.java
b/session/src/test/java/org/apache/iotdb/session/SessionTest.java
similarity index 92%
rename from session/src/test/java/org/apache/iotdb/session/SessionUT.java
rename to session/src/test/java/org/apache/iotdb/session/SessionTest.java
index 8ec1e2e..290c6be 100644
--- a/session/src/test/java/org/apache/iotdb/session/SessionUT.java
+++ b/session/src/test/java/org/apache/iotdb/session/SessionTest.java
@@ -43,9 +43,10 @@ import java.util.List;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-public class SessionUT {
+public class SessionTest {
private Session session;
@@ -244,4 +245,26 @@ public class SessionUT {
"template1", schemaNames, measurementList, dataTypeList, encodingList,
compressionTypes);
session.setSchemaTemplate("template1", "root.sg.1");
}
+
+ @Test
+ public void testBuilder() {
+ session =
+ new Session.Builder()
+ .host("localhost")
+ .port(1234)
+ .fetchSize(1)
+ .username("abc")
+ .password("123456")
+ .thriftDefaultBufferSize(2)
+ .thriftMaxFrameSize(3)
+ .enableCacheLeader(true)
+ .build();
+
+ assertEquals(session.fetchSize, 1);
+ assertEquals(session.username, "abc");
+ assertEquals(session.password, "123456");
+ assertEquals(session.thriftDefaultBufferSize, 2);
+ assertEquals(session.thriftMaxFrameSize, 3);
+ assertTrue(session.enableCacheLeader);
+ }
}
diff --git
a/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java
b/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java
index 6b7d68f..9a5c4a3 100644
--- a/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java
+++ b/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java
@@ -361,4 +361,30 @@ public class SessionPoolTest {
// e.g., thread A created a new session, but not returned; thread B close
the pool; A get the
// session.
}
+
+ @Test
+ public void testBuilder() {
+ SessionPool pool =
+ new SessionPool.Builder()
+ .host("localhost")
+ .port(1234)
+ .maxSize(10)
+ .user("abc")
+ .password("123")
+ .fetchSize(1)
+ .timeout(2)
+ .enableCacheLeader(true)
+ .enableCompression(true)
+ .build();
+
+ assertEquals(pool.getHost(), "localhost");
+ assertEquals(pool.getPort(), 1234);
+ assertEquals(pool.getUser(), "abc");
+ assertEquals(pool.getPassword(), "123");
+ assertEquals(pool.getMaxSize(), 10);
+ assertEquals(pool.getFetchSize(), 1);
+ assertEquals(pool.getTimeout(), 2);
+ assertTrue(pool.isEnableCacheLeader());
+ assertTrue(pool.isEnableCompression());
+ }
}
