From 3ea9aef383aeb2e6df29fab44052d56399c8e047 Mon Sep 17 00:00:00 2001 From: HTHou Date: Thu, 29 Aug 2024 14:31:32 +0800 Subject: [PATCH 1/7] add IT --- test/e2e/e2e_test.go | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go index 5509d40..ca3392d 100644 --- a/test/e2e/e2e_test.go +++ b/test/e2e/e2e_test.go @@ -24,6 +24,7 @@ import ( "github.com/apache/iotdb-client-go/common" "log" "math/rand" + "strings" "testing" "time" @@ -74,6 +75,20 @@ func (s *e2eTestSuite) checkError(status *common.TSStatus, err error) { } } +func (s *e2eTestSuite) Test_ClusterSessionInit() { + config := &client.ClusterConfig{ + NodeUrls: strings.Split("iotdb:6668,1iotdb:6667,iotdb:6669", ","), + UserName: "root", + Password: "root", + } + var session client.Session + session = client.NewClusterSession(config) + err := session.Open(false, 0) + s.Require().NoError(err) + _, err = session.Close() + s.Require().NoError(err) +} + func (s *e2eTestSuite) Test_CreateTimeseries() { var ( path = "root.tsg1.dev1.status" From 020669834b1b5e1b28c3217744025f01179414c7 Mon Sep 17 00:00:00 2001 From: HTHou Date: Thu, 29 Aug 2024 14:33:19 +0800 Subject: [PATCH 2/7] add IT --- test/e2e/e2e_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go index ca3392d..c4d2f7b 100644 --- a/test/e2e/e2e_test.go +++ b/test/e2e/e2e_test.go @@ -77,7 +77,7 @@ func (s *e2eTestSuite) checkError(status *common.TSStatus, err error) { func (s *e2eTestSuite) Test_ClusterSessionInit() { config := &client.ClusterConfig{ - NodeUrls: strings.Split("iotdb:6668,1iotdb:6667,iotdb:6669", ","), + NodeUrls: strings.Split("iotdb:6668,iotdb:6667,iotdb:6669", ","), UserName: "root", Password: "root", } From 6bd3155baa4861a451f852094da65b0c46f3de7e Mon Sep 17 00:00:00 2001 From: HTHou Date: Thu, 29 Aug 2024 14:41:44 +0800 Subject: [PATCH 3/7] add IT --- test/e2e/e2e_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go index c4d2f7b..0ca67d1 100644 --- a/test/e2e/e2e_test.go +++ b/test/e2e/e2e_test.go @@ -83,7 +83,7 @@ func (s *e2eTestSuite) Test_ClusterSessionInit() { } var session client.Session session = client.NewClusterSession(config) - err := session.Open(false, 0) + err := session.OpenCluster(false) s.Require().NoError(err) _, err = session.Close() s.Require().NoError(err) From a70fb998917020b47cdcdeeabc2d72e79f2d4fa7 Mon Sep 17 00:00:00 2001 From: HTHou Date: Thu, 29 Aug 2024 15:27:37 +0800 Subject: [PATCH 4/7] fix IT --- test/e2e/e2e_test.go | 22 +++------------------- 1 file changed, 3 insertions(+), 19 deletions(-) diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go index 0ca67d1..d9f64ac 100644 --- a/test/e2e/e2e_test.go +++ b/test/e2e/e2e_test.go @@ -42,14 +42,12 @@ func TestE2ETestSuite(t *testing.T) { } func (s *e2eTestSuite) SetupSuite() { - config := &client.Config{ - Host: "iotdb", - Port: "6667", + clusterConfig := client.ClusterConfig{ + NodeUrls: strings.Split("0.0.0.0:6668,0.0.0.0:6667,0.0.0.0:6669", ","), UserName: "root", Password: "root", } - - s.session = client.NewSession(config) + s.session = client.NewClusterSession(&clusterConfig) err := s.session.Open(false, 0) s.Require().NoError(err) } @@ -75,20 +73,6 @@ func (s *e2eTestSuite) checkError(status *common.TSStatus, err error) { } } -func (s *e2eTestSuite) Test_ClusterSessionInit() { - config := &client.ClusterConfig{ - NodeUrls: strings.Split("iotdb:6668,iotdb:6667,iotdb:6669", ","), - UserName: "root", - Password: "root", - } - var session client.Session - session = client.NewClusterSession(config) - err := session.OpenCluster(false) - s.Require().NoError(err) - _, err = session.Close() - s.Require().NoError(err) -} - func (s *e2eTestSuite) Test_CreateTimeseries() { var ( path = "root.tsg1.dev1.status" From 96d413c8ea2aae3fc15cd74f44e9fa8c457bb84d Mon Sep 17 00:00:00 2001 From: HTHou Date: Thu, 29 Aug 2024 15:28:21 +0800 Subject: [PATCH 5/7] fix IT --- test/e2e/e2e_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go index d9f64ac..c7a971e 100644 --- a/test/e2e/e2e_test.go +++ b/test/e2e/e2e_test.go @@ -43,7 +43,7 @@ func TestE2ETestSuite(t *testing.T) { func (s *e2eTestSuite) SetupSuite() { clusterConfig := client.ClusterConfig{ - NodeUrls: strings.Split("0.0.0.0:6668,0.0.0.0:6667,0.0.0.0:6669", ","), + NodeUrls: strings.Split("iotdb:6668,iotdb:6667,iotdb:6669", ","), UserName: "root", Password: "root", } From 1a4ba32c33b56ebf10b47dc23947517ab8c66ab5 Mon Sep 17 00:00:00 2001 From: HTHou Date: Thu, 29 Aug 2024 15:34:07 +0800 Subject: [PATCH 6/7] fix bug --- client/session.go | 26 ++++++------- example/session_pool/session_pool_example.go | 39 ++++++++++---------- 2 files changed, 32 insertions(+), 33 deletions(-) diff --git a/client/session.go b/client/session.go index 0b55866..47985f8 100644 --- a/client/session.go +++ b/client/session.go @@ -1137,23 +1137,21 @@ func NewClusterSession(clusterConfig *ClusterConfig) Session { 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 }) - if err == nil { - // session.trans = thrift.NewTFramedTransport(session.trans) // deprecated - var tmp_conf = thrift.TConfiguration{MaxFrameSize: thrift.DEFAULT_MAX_FRAME_SIZE} - session.trans = thrift.NewTFramedTransportConf(session.trans, &tmp_conf) - if !session.trans.IsOpen() { - err = session.trans.Open() - if err != nil { - log.Println(err) - } else { - session.config = getConfig(e.Value.(endPoint).Host, e.Value.(endPoint).Port, - clusterConfig.UserName, clusterConfig.Password, clusterConfig.FetchSize, clusterConfig.TimeZone, clusterConfig.ConnectRetryMax) - break - } + // session.trans = thrift.NewTFramedTransport(session.trans) // deprecated + var tmp_conf = thrift.TConfiguration{MaxFrameSize: thrift.DEFAULT_MAX_FRAME_SIZE} + session.trans = thrift.NewTFramedTransportConf(session.trans, &tmp_conf) + if !session.trans.IsOpen() { + err = session.trans.Open() + if err != nil { + log.Println(err) + } else { + session.config = getConfig(e.Value.(endPoint).Host, e.Value.(endPoint).Port, + clusterConfig.UserName, clusterConfig.Password, clusterConfig.FetchSize, clusterConfig.TimeZone, clusterConfig.ConnectRetryMax) + break } } } - if err != nil { + if !session.trans.IsOpen() { log.Fatal("No Server Can Connect") } return session diff --git a/example/session_pool/session_pool_example.go b/example/session_pool/session_pool_example.go index a4641d9..459393b 100644 --- a/example/session_pool/session_pool_example.go +++ b/example/session_pool/session_pool_example.go @@ -67,6 +67,7 @@ func main() { }() } + //useNodeUrls() setStorageGroup("root.ln1") setStorageGroup("root.ln2") deleteStorageGroups("root.ln1", "root.ln2") @@ -139,6 +140,25 @@ func main() { } +// If your IoTDB is a cluster version, you can use the following code for session pool connection +func useNodeUrls() { + + config := &client.PoolConfig{ + UserName: user, + Password: password, + NodeUrls: strings.Split("127.0.0.1:6667,127.0.0.1:6668", ","), + } + sessionPool = client.NewSessionPool(config, 3, 60000, 60000, false) + defer sessionPool.Close() + session, err := sessionPool.GetSession() + defer sessionPool.PutBack(session) + if err != nil { + log.Print(err) + return + } + +} + func setStorageGroup(sg string) { session, err := sessionPool.GetSession() defer sessionPool.PutBack(session) @@ -762,22 +782,3 @@ func checkError(status *common.TSStatus, err error) { } } } - -// If your IotDB is a cluster version or doubleLive, you can use the following code for session pool connection -func useSessionPool() { - - config := &client.PoolConfig{ - UserName: user, - Password: password, - NodeUrls: strings.Split("127.0.0.1:6667,127.0.0.1:6668", ","), - } - sessionPool = client.NewSessionPool(config, 3, 60000, 60000, false) - defer sessionPool.Close() - session, err := sessionPool.GetSession() - defer sessionPool.PutBack(session) - if err != nil { - log.Print(err) - return - } - -} From 66f93c1ac26c86294aac7d4efba4568fad4ffaf1 Mon Sep 17 00:00:00 2001 From: HTHou Date: Thu, 29 Aug 2024 15:40:44 +0800 Subject: [PATCH 7/7] opt example --- example/session_example.go | 28 +++++++++++++++------------- 1 file changed, 15 insertions(+), 13 deletions(-) diff --git a/example/session_example.go b/example/session_example.go index b370435..fbfba4f 100644 --- a/example/session_example.go +++ b/example/session_example.go @@ -57,6 +57,8 @@ func main() { } defer session.Close() + //connectCluster() + setStorageGroup("root.ln1") deleteStorageGroup("root.ln1") @@ -143,6 +145,19 @@ func main() { deleteTimeseries("root.ln.device1.*") } +// If your IotDB is a cluster version, you can use the following code for multi node connection +func connectCluster() { + config := &client.ClusterConfig{ + NodeUrls: strings.Split("127.0.0.1:6667,127.0.0.1:6668,127.0.0.1:6669", ","), + UserName: "root", + Password: "root", + } + session = client.NewClusterSession(config) + if err := session.OpenCluster(false); err != nil { + log.Fatal(err) + } +} + func printDevice1(sds *client.SessionDataSet) { showTimestamp := !sds.IsIgnoreTimeStamp() if showTimestamp { @@ -667,16 +682,3 @@ func checkError(status *common.TSStatus, err error) { } } } - -// If your IotDB is a cluster version, you can use the following code for multi node connection -func connectCluster() { - config := &client.ClusterConfig{ - NodeUrls: strings.Split("127.0.0.1:6667,127.0.0.1:6668,127.0.0.1:6669", ","), - UserName: "root", - Password: "root", - } - session = client.NewClusterSession(config) - if err := session.OpenCluster(false); err != nil { - log.Fatal(err) - } -}