Add session pool

Add tests

client_test.go

@@ -63,19 +63,19 @@ func logoutAndClose(conn *connection, sessionID int64) {
 }
 
 func TestConnection(t *testing.T) {
-	hostAdress := HostAddress{Host: address, Port: port}
-	conn := newConnection(hostAdress)
-	err := conn.open(hostAdress, testPoolConfig.TimeOut, nil)
+	hostAddress := HostAddress{Host: address, Port: port}
+	conn := newConnection(hostAddress)
+	err := conn.open(hostAddress, testPoolConfig.TimeOut, nil)
 	if err != nil {
 		t.Fatalf("fail to open connection, address: %s, port: %d, %s", address, port, err.Error())
 	}
 
-	authresp, authErr := conn.authenticate(username, password)
+	authResp, authErr := conn.authenticate(username, password)
 	if authErr != nil {
 		t.Fatalf("fail to authenticate, username: %s, password: %s, %s", username, password, authErr.Error())
 	}
 
-	sessionID := authresp.GetSessionID()
+	sessionID := authResp.GetSessionID()
 
 	defer logoutAndClose(conn, sessionID)
 
@@ -107,19 +107,19 @@ func TestConnection(t *testing.T) {
 }
 
 func TestConnectionIPv6(t *testing.T) {
-	hostAdress := HostAddress{Host: addressIPv6, Port: port}
-	conn := newConnection(hostAdress)
-	err := conn.open(hostAdress, testPoolConfig.TimeOut, nil)
+	hostAddress := HostAddress{Host: addressIPv6, Port: port}
+	conn := newConnection(hostAddress)
+	err := conn.open(hostAddress, testPoolConfig.TimeOut, nil)
 	if err != nil {
 		t.Fatalf("fail to open connection, address: %s, port: %d, %s", address, port, err.Error())
 	}
 
-	authresp, authErr := conn.authenticate(username, password)
+	authResp, authErr := conn.authenticate(username, password)
 	if authErr != nil {
 		t.Fatalf("fail to authenticate, username: %s, password: %s, %s", username, password, authErr.Error())
 	}
 
-	sessionID := authresp.GetSessionID()
+	sessionID := authResp.GetSessionID()
 
 	defer logoutAndClose(conn, sessionID)
 
@@ -151,9 +151,9 @@ func TestConnectionIPv6(t *testing.T) {
 }
 
 func TestConfigs(t *testing.T) {
-	hostAdress := HostAddress{Host: address, Port: port}
+	hostAddress := HostAddress{Host: address, Port: port}
 	hostList := []HostAddress{}
-	hostList = append(hostList, hostAdress)
+	hostList = append(hostList, hostAddress)
 
 	var configList = []PoolConfig{
 		// default
@@ -202,7 +202,7 @@ func TestConfigs(t *testing.T) {
 				username, password, err.Error())
 		}
 		defer session.Release()
-		// Excute a query
+		// Execute a query
 		resp, err := tryToExecute(session, "SHOW HOSTS;")
 		if err != nil {
 			t.Fatalf(err.Error())
@@ -231,10 +231,10 @@ func TestAuthentication(t *testing.T) {
 		password = "nebula"
 	)
 
-	hostAdress := HostAddress{Host: address, Port: port}
+	hostAddress := HostAddress{Host: address, Port: port}
 
-	conn := newConnection(hostAdress)
-	err := conn.open(hostAdress, testPoolConfig.TimeOut, nil)
+	conn := newConnection(hostAddress)
+	err := conn.open(hostAddress, testPoolConfig.TimeOut, nil)
 	if err != nil {
 		t.Fatalf("fail to open connection, address: %s, port: %d, %s", address, port, err.Error())
 	}
@@ -245,8 +245,8 @@ func TestAuthentication(t *testing.T) {
 }
 
 func TestInvalidHostTimeout(t *testing.T) {
-	hostAdress := HostAddress{Host: address, Port: port}
-	hostList := []HostAddress{hostAdress}
+	hostAddress := HostAddress{Host: address, Port: port}
+	hostList := []HostAddress{hostAddress}
 
 	invalidHostList := []HostAddress{
 		{Host: "192.168.100.125", Port: 3699}, // Invalid host
@@ -269,9 +269,9 @@ func TestInvalidHostTimeout(t *testing.T) {
 }
 
 func TestServiceDataIO(t *testing.T) {
-	hostAdress := HostAddress{Host: address, Port: port}
+	hostAddress := HostAddress{Host: address, Port: port}
 	hostList := []HostAddress{}
-	hostList = append(hostList, hostAdress)
+	hostList = append(hostList, hostAddress)
 
 	testPoolConfig = PoolConfig{
 		TimeOut:         0 * time.Millisecond,
@@ -558,9 +558,9 @@ func TestServiceDataIO(t *testing.T) {
 }
 
 func TestPool_SingleHost(t *testing.T) {
-	hostAdress := HostAddress{Host: address, Port: port}
+	hostAddress := HostAddress{Host: address, Port: port}
 	hostList := []HostAddress{}
-	hostList = append(hostList, hostAdress)
+	hostList = append(hostList, hostAddress)
 
 	testPoolConfig = PoolConfig{
 		TimeOut:         0 * time.Millisecond,
@@ -604,7 +604,7 @@ func TestPool_SingleHost(t *testing.T) {
 
 func TestPool_MultiHosts(t *testing.T) {
 	hostList := poolAddress
-	// Minimun pool size < hosts number
+	// Minimum pool size < hosts number
 	multiHostsConfig := PoolConfig{
 		TimeOut:         0 * time.Millisecond,
 		IdleTime:        0 * time.Millisecond,
@@ -722,7 +722,7 @@ func TestMultiThreads(t *testing.T) {
 	assert.Equal(t, 666, pool.getIdleConnCount(), "Total number of idle connections should be 666")
 }
 
-func TestLoadbalancer(t *testing.T) {
+func TestLoadBalancer(t *testing.T) {
 	hostList := poolAddress
 	var loadPerHost = make(map[HostAddress]int)
 	testPoolConfig := PoolConfig{
@@ -808,9 +808,9 @@ func TestIdleTimeoutCleaner(t *testing.T) {
 }
 
 func TestTimeout(t *testing.T) {
-	hostAdress := HostAddress{Host: address, Port: port}
+	hostAddress := HostAddress{Host: address, Port: port}
 	hostList := []HostAddress{}
-	hostList = append(hostList, hostAdress)
+	hostList = append(hostList, hostAddress)
 
 	testPoolConfig = PoolConfig{
 		TimeOut:         1000 * time.Millisecond,

configs.go

@@ -105,3 +105,91 @@ func openAndReadFile(path string) ([]byte, error) {
 	}
 	return b, nil
 }
+
+// SessionPoolConf is the configs of a session pool
+// Note that the space name is bound to the session pool for its lifetime
+type SessionPoolConf struct {
+	Username     string        // username for authentication
+	Password     string        // password for authentication
+	ServiceAddrs []HostAddress // service addresses for session pool
+	hostIndex    int           // index of the host in ServiceAddrs that the next new session will connect to
+	SpaceName    string        // The space name that all sessions in the pool are bound to
+	sslConfig    *tls.Config   // Optional SSL config for the connection
+
+	// Basic pool configs
+	// Socket timeout and Socket connection timeout, unit: seconds
+	TimeOut time.Duration
+	// The idleTime of the connection, unit: seconds
+	// If connection's idle time is longer than idleTime, it will be delete
+	// 0 value means the connection will not expire
+	IdleTime time.Duration
+	// The max sessions in pool for all addresses
+	MaxSize int
+	// The min sessions in pool for all addresses
+	MinSize int
+}
+
+// TODO(Aiee) add more constructors
+// NewSessionPoolConf returns a new SessionPoolConf with given parameters
+func NewSessionPoolConf(username, password string, serviceAddrs []HostAddress, spaceName string) (*SessionPoolConf, error) {
+	newPoolConf := GetDefaultSessionConf()
+	newPoolConf.Username = username
+	newPoolConf.Password = password
+	newPoolConf.ServiceAddrs = serviceAddrs
+	newPoolConf.SpaceName = spaceName
+
+	if err := newPoolConf.checkMandatoryFields(); err != nil {
+		return nil, err
+	}
+
+	return &newPoolConf, nil
+}
+
+// GetDefaultSessionConf returns the default config
+func GetDefaultSessionConf() SessionPoolConf {
+	return SessionPoolConf{
+		TimeOut:  0 * time.Millisecond,
+		IdleTime: 0 * time.Millisecond,
+		MaxSize:  10,
+		MinSize:  0,
+	}
+}
+
+func (conf *SessionPoolConf) checkMandatoryFields() error {
+	// Check mandatory fields
+	if conf.Username == "" {
+		return fmt.Errorf("invalid session pool config: Username is empty")
+	}
+	if conf.Password == "" {
+		return fmt.Errorf("invalid session pool config: Password is empty")
+	}
+	if len(conf.ServiceAddrs) == 0 {
+		return fmt.Errorf("invalid session pool config: Service address is empty")
+	}
+	if conf.SpaceName == "" {
+		return fmt.Errorf("invalid session pool config: Space name is empty")
+	}
+	return nil
+}
+
+// checkBasicFields checks the basic fields of the config and
+// sets a default value if the given field value is invalid
+func (conf *SessionPoolConf) checkBasicFields(log Logger) {
+	// Check pool related fields, use default value if the given value is invalid
+	if conf.TimeOut < 0 {
+		conf.TimeOut = 0 * time.Millisecond
+		log.Warn("Illegal Timeout value, the default value of 0 second has been applied")
+	}
+	if conf.IdleTime < 0 {
+		conf.IdleTime = 0 * time.Millisecond
+		log.Warn("Invalid IdleTime value, the default value of 0 second has been applied")
+	}
+	if conf.MaxSize < 1 {
+		conf.MaxSize = 10
+		log.Warn("Invalid MaxSize value, the default value of 10 has been applied")
+	}
+	if conf.MinSize < 0 {
+		conf.MinSize = 0
+		log.Warn("Invalid MinSize value, the default value of 0 has been applied")
+	}
+}

connection_pool.go

@@ -70,7 +70,7 @@ func NewSslConnectionPool(addresses []HostAddress, conf PoolConfig, sslConfig *t
 
 // initPool initializes the connection pool
 func (pool *ConnectionPool) initPool() error {
-	if err := pool.checkAddresses(); err != nil {
+	if err := checkAddresses(pool.conf.TimeOut, pool.addresses, pool.sslConfig); err != nil {
 		return fmt.Errorf("failed to open connection, error: %s ", err.Error())
 	}
 
@@ -167,7 +167,7 @@ func (pool *ConnectionPool) getIdleConn() (*connection, error) {
 
 	// Create a new connection if there is no idle connection and total connection < pool max size
 	newConn, err := pool.createConnection()
-	// TODO: If no idle avaliable, wait for timeout and reconnect
+	// TODO: If no idle available, wait for timeout and reconnect
 	return newConn, err
 }
 
@@ -181,15 +181,9 @@ func (pool *ConnectionPool) release(conn *connection) {
 	pool.idleConnectionQueue.PushBack(conn)
 }
 
-// Ping checks avaliability of host
+// Ping checks availability of host
 func (pool *ConnectionPool) Ping(host HostAddress, timeout time.Duration) error {
-	newConn := newConnection(host)
-	// Open connection to host
-	if err := newConn.open(newConn.severAddress, timeout, pool.sslConfig); err != nil {
-		return err
-	}
-	newConn.close()
-	return nil
+	return pingAddress(host, timeout, pool.sslConfig)
 }
 
 // Close closes all connection
@@ -259,7 +253,7 @@ func removeFromList(l *list.List, conn *connection) {
 // Compare total connection number with pool max size and return a connection if capable
 func (pool *ConnectionPool) createConnection() (*connection, error) {
 	totalConn := pool.idleConnectionQueue.Len() + pool.activeConnectionQueue.Len()
-	// If no idle avaliable and the number of total connection reaches the max pool size, return error/wait for timeout
+	// If no idle available and the number of total connection reaches the max pool size, return error/wait for timeout
 	if totalConn >= pool.conf.MaxConnPoolSize {
 		return nil, fmt.Errorf("failed to get connection: No valid connection" +
 			" in the idle queue and connection number has reached the pool capacity")
@@ -342,15 +336,28 @@ func (pool *ConnectionPool) timeoutConnectionList() (closing []*connection) {
 	return
 }
 
-func (pool *ConnectionPool) checkAddresses() error {
+// checkAddresses checks addresses availability
+// It opens a temporary connection to each address and closes it immediately.
+// If no error is returned, the addresses are available.
+func checkAddresses(confTimeout time.Duration, addresses []HostAddress, sslConfig *tls.Config) error {
 	var timeout = 3 * time.Second
-	if pool.conf.TimeOut != 0 && pool.conf.TimeOut < timeout {
-		timeout = pool.conf.TimeOut
+	if confTimeout != 0 && confTimeout < timeout {
+		timeout = confTimeout
 	}
-	for _, address := range pool.addresses {
-		if err := pool.Ping(address, timeout); err != nil {
+	for _, address := range addresses {
+		if err := pingAddress(address, timeout, sslConfig); err != nil {
 			return err
 		}
 	}
 	return nil
 }
+
+func pingAddress(address HostAddress, timeout time.Duration, sslConfig *tls.Config) error {
+	newConn := newConnection(address)
+	defer newConn.close()
+	// Open connection to host
+	if err := newConn.open(newConn.severAddress, timeout, sslConfig); err != nil {
+		return err
+	}
+	return nil
+}