From 6025f25374ca40ef57e7faba2ed443594e06f78f Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Fri, 21 Mar 2025 14:47:38 +0800 Subject: [PATCH 1/2] remove global endPointList --- client/session.go | 16 ++++++++++------ test/e2e/e2e_test.go | 10 ++++++++++ 2 files changed, 20 insertions(+), 6 deletions(-) diff --git a/client/session.go b/client/session.go index e5f2184..de5631d 100644 --- a/client/session.go +++ b/client/session.go @@ -63,6 +63,7 @@ type Session struct { trans thrift.TTransport requestStatementId int64 protocolFactory thrift.TProtocolFactory + endPointList *list.List } type endPoint struct { @@ -70,8 +71,6 @@ type endPoint struct { Port string } -var endPointList = list.New() - func (s *Session) Open(enableRPCCompression bool, connectionTimeoutInMs int) error { if s.config.FetchSize <= 0 { s.config.FetchSize = DefaultFetchSize @@ -1078,23 +1077,28 @@ func (s *Session) GetSessionId() int64 { } func NewSession(config *Config) Session { + endPointList := list.New() endPoint := endPoint{} endPoint.Host = config.Host endPoint.Port = config.Port endPointList.PushBack(endPoint) - return Session{config: config} + return Session{ + config: config, + endPointList: endPointList, + } } func NewClusterSession(clusterConfig *ClusterConfig) (Session, error) { session := Session{} node := endPoint{} + session.endPointList = list.New() for i := 0; i < len(clusterConfig.NodeUrls); i++ { node.Host = strings.Split(clusterConfig.NodeUrls[i], ":")[0] node.Port = strings.Split(clusterConfig.NodeUrls[i], ":")[1] - endPointList.PushBack(node) + session.endPointList.PushBack(node) } var err error - for e := endPointList.Front(); e != nil; e = e.Next() { + for e := session.endPointList.Front(); e != nil; e = e.Next() { session.trans = thrift.NewTSocketConf(net.JoinHostPort(e.Value.(endPoint).Host, e.Value.(endPoint).Port), &thrift.TConfiguration{ ConnectTimeout: time.Duration(0), // Use 0 for no timeout }) @@ -1181,7 +1185,7 @@ func (s *Session) reconnect() bool { var connectedSuccess = false for i := 0; i < s.config.ConnectRetryMax; i++ { - for e := endPointList.Front(); e != nil; e = e.Next() { + for e := s.endPointList.Front(); e != nil; e = e.Next() { err = s.initClusterConn(e.Value.(endPoint)) if err == nil { connectedSuccess = true diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go index e4b91ea..3ad5c2e 100644 --- a/test/e2e/e2e_test.go +++ b/test/e2e/e2e_test.go @@ -75,6 +75,16 @@ func (s *e2eTestSuite) checkError(status *common.TSStatus, err error) { } } +func (s *e2eTestSuite) Test_WrongURL() { + clusterConfig := client.ClusterConfig{ + NodeUrls: strings.Split("iotdb1:6667", ","), + UserName: "root", + Password: "root", + } + _, err := client.NewClusterSession(&clusterConfig) + s.Require().Error(err) +} + func (s *e2eTestSuite) Test_CreateTimeseries() { var ( path = "root.tsg1.dev1.status" From 316db5d45cfa69ec98ceeff6271d72d69141085b Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Fri, 21 Mar 2025 18:16:31 +0800 Subject: [PATCH 2/2] modify type of endpointList --- client/session.go | 32 ++++++++++++++++---------------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/client/session.go b/client/session.go index de5631d..aa778fb 100644 --- a/client/session.go +++ b/client/session.go @@ -21,7 +21,6 @@ package client import ( "bytes" - "container/list" "context" "encoding/binary" "errors" @@ -63,7 +62,7 @@ type Session struct { trans thrift.TTransport requestStatementId int64 protocolFactory thrift.TProtocolFactory - endPointList *list.List + endPointList []endPoint } type endPoint struct { @@ -1077,11 +1076,10 @@ func (s *Session) GetSessionId() int64 { } func NewSession(config *Config) Session { - endPointList := list.New() - endPoint := endPoint{} - endPoint.Host = config.Host - endPoint.Port = config.Port - endPointList.PushBack(endPoint) + endPointList := []endPoint{{ + Host: config.Host, + Port: config.Port, + }} return Session{ config: config, endPointList: endPointList, @@ -1090,16 +1088,17 @@ func NewSession(config *Config) Session { func NewClusterSession(clusterConfig *ClusterConfig) (Session, error) { session := Session{} - node := endPoint{} - session.endPointList = list.New() + session.endPointList = make([]endPoint, len(clusterConfig.NodeUrls)) for i := 0; i < len(clusterConfig.NodeUrls); i++ { + node := endPoint{} node.Host = strings.Split(clusterConfig.NodeUrls[i], ":")[0] node.Port = strings.Split(clusterConfig.NodeUrls[i], ":")[1] - session.endPointList.PushBack(node) + session.endPointList[i] = node } var err error - for e := session.endPointList.Front(); e != nil; e = e.Next() { - session.trans = thrift.NewTSocketConf(net.JoinHostPort(e.Value.(endPoint).Host, e.Value.(endPoint).Port), &thrift.TConfiguration{ + for i := range session.endPointList { + ep := session.endPointList[i] + session.trans = thrift.NewTSocketConf(net.JoinHostPort(ep.Host, ep.Port), &thrift.TConfiguration{ ConnectTimeout: time.Duration(0), // Use 0 for no timeout }) // session.trans = thrift.NewTFramedTransport(session.trans) // deprecated @@ -1110,7 +1109,7 @@ func NewClusterSession(clusterConfig *ClusterConfig) (Session, error) { if err != nil { log.Println(err) } else { - session.config = getConfig(e.Value.(endPoint).Host, e.Value.(endPoint).Port, + session.config = getConfig(ep.Host, ep.Port, clusterConfig.UserName, clusterConfig.Password, clusterConfig.FetchSize, clusterConfig.TimeZone, clusterConfig.ConnectRetryMax) break } @@ -1185,13 +1184,14 @@ func (s *Session) reconnect() bool { var connectedSuccess = false for i := 0; i < s.config.ConnectRetryMax; i++ { - for e := s.endPointList.Front(); e != nil; e = e.Next() { - err = s.initClusterConn(e.Value.(endPoint)) + for i := range s.endPointList { + ep := s.endPointList[i] + err = s.initClusterConn(ep) if err == nil { connectedSuccess = true break } else { - log.Println("Connection refused:", e.Value.(endPoint)) + log.Println("Connection refused:", ep) } } if connectedSuccess {