Skip to content

Commit d2c0f9e

Browse files
Rynzie02Rynzie02
authored
refactor(samples): avoid time.Sleep in tests and examples (#1029)
* refactor(samples): avoid time.Sleep in tests and examples * chore(format): run imports-formatter * remove unnecessary timer.Stop()
1 parent 6141cf3 commit d2c0f9e

File tree

10 files changed

+286
-70
lines changed

10 files changed

+286
-70
lines changed

‎async/go-client/cmd/main.go‎

Lines changed: 22 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -65,14 +65,31 @@ func main() {
6565

6666
func testAsync() {
6767
req := &user.GetUserRequest{Id: "003"}
68-
_, err := userProvider.GetUser(context.Background(), req)
69-
if err != nil {
70-
panic(err)
71-
}
68+
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
69+
defer cancel()
70+
71+
respCh := make(chan *user.GetUserResponse, 1)
72+
errCh := make(chan error, 1)
73+
74+
go func() {
75+
resp, err := userProvider.GetUser(ctx, req)
76+
if err != nil {
77+
errCh <- err
78+
return
79+
}
80+
respCh <- resp
81+
}()
7282

7383
logger.Info("non-blocking before async callback resp: do something ... ")
7484

75-
time.Sleep(time.Second)
85+
select {
86+
case err := <-errCh:
87+
panic(err)
88+
case resp := <-respCh:
89+
logger.Infof("async callback resp: %+v", resp)
90+
case <-ctx.Done():
91+
logger.Warnf("async callback timeout: %v", ctx.Err())
92+
}
7693
}
7794

7895
func testAsyncOneWay() {

‎config_center/nacos/go-client/cmd/main.go‎

Lines changed: 50 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@ package main
1919

2020
import (
2121
"context"
22+
"fmt"
23+
"strings"
2224
"time"
2325
)
2426

@@ -30,6 +32,7 @@ import (
3032
"github.com/dubbogo/gost/log/logger"
3133

3234
"github.com/nacos-group/nacos-sdk-go/v2/clients"
35+
"github.com/nacos-group/nacos-sdk-go/v2/clients/config_client"
3336
"github.com/nacos-group/nacos-sdk-go/v2/common/constant"
3437
"github.com/nacos-group/nacos-sdk-go/v2/vo"
3538
)
@@ -74,19 +77,16 @@ func main() {
7477
panic(err)
7578
}
7679

77-
success, err := configClient.PublishConfig(vo.ConfigParam{
78-
DataId: "dubbo-go-samples-configcenter-nacos-go-client",
79-
Group: "dubbo",
80-
Content: configCenterNacosClientConfig,
81-
})
82-
if err != nil {
80+
if err := publishAndWaitConfig(
81+
configClient,
82+
"dubbo-go-samples-configcenter-nacos-go-client",
83+
"dubbo",
84+
configCenterNacosClientConfig,
85+
10*time.Second,
86+
200*time.Millisecond,
87+
); err != nil {
8388
panic(err)
8489
}
85-
if !success {
86-
return
87-
}
88-
89-
time.Sleep(time.Second * 10)
9090

9191
nacosOption := config_center.WithNacos()
9292
dataIdOption := config_center.WithDataID("dubbo-go-samples-configcenter-nacos-go-client")
@@ -115,3 +115,42 @@ func main() {
115115
}
116116
logger.Infof("Server response: %s", resp)
117117
}
118+
119+
func publishAndWaitConfig(
120+
configClient config_client.IConfigClient,
121+
dataID string,
122+
group string,
123+
content string,
124+
timeout time.Duration,
125+
pollInterval time.Duration,
126+
) error {
127+
success, err := configClient.PublishConfig(vo.ConfigParam{
128+
DataId: dataID,
129+
Group: group,
130+
Content: content,
131+
})
132+
if err != nil {
133+
return err
134+
}
135+
if !success {
136+
return fmt.Errorf("publish config failed")
137+
}
138+
139+
deadline := time.Now().Add(timeout)
140+
for {
141+
current, err := configClient.GetConfig(vo.ConfigParam{
142+
DataId: dataID,
143+
Group: group,
144+
})
145+
if err == nil && strings.TrimSpace(current) == strings.TrimSpace(content) {
146+
return nil
147+
}
148+
if time.Now().After(deadline) {
149+
if err != nil {
150+
return err
151+
}
152+
return fmt.Errorf("wait for config center timeout")
153+
}
154+
time.Sleep(pollInterval)
155+
}
156+
}

‎config_center/nacos/go-server/cmd/main.go‎

Lines changed: 50 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@ package main
1919

2020
import (
2121
"context"
22+
"fmt"
23+
"strings"
2224
"time"
2325
)
2426

@@ -30,6 +32,7 @@ import (
3032
"github.com/dubbogo/gost/log/logger"
3133

3234
"github.com/nacos-group/nacos-sdk-go/v2/clients"
35+
"github.com/nacos-group/nacos-sdk-go/v2/clients/config_client"
3336
"github.com/nacos-group/nacos-sdk-go/v2/common/constant"
3437
"github.com/nacos-group/nacos-sdk-go/v2/vo"
3538
)
@@ -86,19 +89,16 @@ func main() {
8689
panic(err)
8790
}
8891

89-
success, err := configClient.PublishConfig(vo.ConfigParam{
90-
DataId: "dubbo-go-samples-configcenter-nacos-go-server",
91-
Group: "dubbo",
92-
Content: configCenterNacosServerConfig,
93-
})
94-
if err != nil {
92+
if err := publishAndWaitConfig(
93+
configClient,
94+
"dubbo-go-samples-configcenter-nacos-go-server",
95+
"dubbo",
96+
configCenterNacosServerConfig,
97+
10*time.Second,
98+
200*time.Millisecond,
99+
); err != nil {
95100
panic(err)
96101
}
97-
if !success {
98-
return
99-
}
100-
101-
time.Sleep(time.Second * 10)
102102

103103
nacosOption := config_center.WithNacos()
104104
dataIdOption := config_center.WithDataID("dubbo-go-samples-configcenter-nacos-go-server")
@@ -123,3 +123,42 @@ func main() {
123123
logger.Error(err)
124124
}
125125
}
126+
127+
func publishAndWaitConfig(
128+
configClient config_client.IConfigClient,
129+
dataID string,
130+
group string,
131+
content string,
132+
timeout time.Duration,
133+
pollInterval time.Duration,
134+
) error {
135+
success, err := configClient.PublishConfig(vo.ConfigParam{
136+
DataId: dataID,
137+
Group: group,
138+
Content: content,
139+
})
140+
if err != nil {
141+
return err
142+
}
143+
if !success {
144+
return fmt.Errorf("publish config failed")
145+
}
146+
147+
deadline := time.Now().Add(timeout)
148+
for {
149+
current, err := configClient.GetConfig(vo.ConfigParam{
150+
DataId: dataID,
151+
Group: group,
152+
})
153+
if err == nil && strings.TrimSpace(current) == strings.TrimSpace(content) {
154+
return nil
155+
}
156+
if time.Now().After(deadline) {
157+
if err != nil {
158+
return err
159+
}
160+
return fmt.Errorf("wait for config center timeout")
161+
}
162+
time.Sleep(pollInterval)
163+
}
164+
}

‎config_center/zookeeper/go-client/cmd/main.go‎

Lines changed: 22 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -47,9 +47,6 @@ func main() {
4747
}
4848
logger.Info("Successfully wrote config to ZooKeeper")
4949

50-
// wait for config write to finish
51-
time.Sleep(time.Second * 3)
52-
5350
// configure Dubbo instance
5451
zkOption := config_center.WithZookeeper()
5552
dataIdOption := config_center.WithDataID("dubbo-go-samples-configcenter-zookeeper-go-client")
@@ -132,9 +129,31 @@ func writeRuleToConfigCenter() error {
132129
logger.Info("Created new config node")
133130
}
134131

132+
if err := waitForConfigReady(c, path, valueBytes, 10*time.Second); err != nil {
133+
return perrors.Wrap(err, "wait for config ready")
134+
}
135+
135136
return nil
136137
}
137138

139+
func waitForConfigReady(c *zk.Conn, path string, expected []byte, timeout time.Duration) error {
140+
deadline := time.Now().Add(timeout)
141+
expectedStr := strings.TrimSpace(string(expected))
142+
for {
143+
data, _, err := c.Get(path)
144+
if err == nil && strings.TrimSpace(string(data)) == expectedStr {
145+
return nil
146+
}
147+
if time.Now().After(deadline) {
148+
if err != nil {
149+
return perrors.Wrap(err, "wait for config timeout")
150+
}
151+
return perrors.New("wait for config timeout")
152+
}
153+
time.Sleep(200 * time.Millisecond)
154+
}
155+
}
156+
138157
// helper function to create parent paths
139158
func createParentPaths(c *zk.Conn, path string) error {
140159
paths := strings.Split(path, "/")

‎config_center/zookeeper/go-server/cmd/main.go‎

Lines changed: 22 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -57,9 +57,6 @@ func main() {
5757
panic(err)
5858
}
5959

60-
// Wait for the configuration to take effect
61-
time.Sleep(time.Second * 10)
62-
6360
ins, err := dubbo.NewInstance(
6461
dubbo.WithConfigCenter(
6562
config_center.WithZookeeper(),
@@ -151,9 +148,31 @@ func writeRuleToConfigCenter() error {
151148
logger.Info("Created new configuration in config center")
152149
}
153150

151+
if err := waitForConfigReady(c, path, valueBytes, 10*time.Second); err != nil {
152+
return perrors.Wrap(err, "wait for config ready")
153+
}
154+
154155
return nil
155156
}
156157

158+
func waitForConfigReady(c *zk.Conn, path string, expected []byte, timeout time.Duration) error {
159+
deadline := time.Now().Add(timeout)
160+
expectedStr := strings.TrimSpace(string(expected))
161+
for {
162+
data, _, err := c.Get(path)
163+
if err == nil && strings.TrimSpace(string(data)) == expectedStr {
164+
return nil
165+
}
166+
if time.Now().After(deadline) {
167+
if err != nil {
168+
return perrors.Wrap(err, "wait for config timeout")
169+
}
170+
return perrors.New("wait for config timeout")
171+
}
172+
time.Sleep(200 * time.Millisecond)
173+
}
174+
}
175+
157176
// createParentPaths Create parent paths
158177
func createParentPaths(c *zk.Conn, path string) error {
159178
paths := strings.Split(path, "/")

‎filter/sentinel/go-client/cmd/main.go‎

Lines changed: 44 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -43,18 +43,43 @@ import (
4343
type GreetFun func(ctx context.Context, req *greet.GreetRequest, opts ...client.CallOption) (*greet.GreetResponse, error)
4444

4545
type stateChangeTestListener struct {
46+
openCh chan struct{}
47+
halfOpenCh chan struct{}
48+
closedCh chan struct{}
4649
}
4750

4851
func (s *stateChangeTestListener) OnTransformToClosed(prev circuitbreaker.State, rule circuitbreaker.Rule) {
4952
logger.Infof("rule.steategy: %+v, From %s to Closed, time: %d\n", rule.Strategy, prev.String(), util.CurrentTimeMillis())
53+
s.notify(s.closedCh)
5054
}
5155

5256
func (s *stateChangeTestListener) OnTransformToOpen(prev circuitbreaker.State, rule circuitbreaker.Rule, snapshot interface{}) {
5357
logger.Infof("rule.steategy: %+v, From %s to Open, snapshot: %.2f, time: %d\n", rule.Strategy, prev.String(), snapshot, util.CurrentTimeMillis())
58+
s.notify(s.openCh)
5459
}
5560

5661
func (s *stateChangeTestListener) OnTransformToHalfOpen(prev circuitbreaker.State, rule circuitbreaker.Rule) {
5762
logger.Infof("rule.steategy: %+v, From %s to Half-Open, time: %d\n", rule.Strategy, prev.String(), util.CurrentTimeMillis())
63+
s.notify(s.halfOpenCh)
64+
}
65+
66+
func (s *stateChangeTestListener) notify(ch chan struct{}) {
67+
if ch == nil {
68+
return
69+
}
70+
select {
71+
case ch <- struct{}{}:
72+
default:
73+
}
74+
}
75+
76+
func waitForState(ch <-chan struct{}, timeout time.Duration) bool {
77+
select {
78+
case <-ch:
79+
return true
80+
case <-time.After(timeout):
81+
return false
82+
}
5883
}
5984

6085
func main() {
@@ -69,8 +94,13 @@ func main() {
6994
if err != nil {
7095
panic(err)
7196
}
97+
listener := &stateChangeTestListener{
98+
openCh: make(chan struct{}, 1),
99+
halfOpenCh: make(chan struct{}, 1),
100+
closedCh: make(chan struct{}, 1),
101+
}
72102
// Register a state change listener so that we could observe the state change of the internal circuit breaker.
73-
circuitbreaker.RegisterStateChangeListeners(&stateChangeTestListener{})
103+
circuitbreaker.RegisterStateChangeListeners(listener)
74104
_, err = circuitbreaker.LoadRules([]*circuitbreaker.Rule{
75105
// Statistic time span=1s, recoveryTimeout=1s, maxErrorRatio=40%
76106
{
@@ -86,6 +116,7 @@ func main() {
86116
if err != nil {
87117
panic(err)
88118
}
119+
retryTimeout := 2 * time.Second
89120

90121
_, err = isolation.LoadRules([]*isolation.Rule{
91122
{
@@ -106,10 +137,19 @@ func main() {
106137

107138
logger.Info("call svc.GreetWithChanceOfError triggers circuit breaker open")
108139
CallGreetFunConcurrently(svc.GreetWithChanceOfError, "error", 1, 300)
109-
logger.Info("wait circuit breaker HalfOpen")
110-
time.Sleep(3 * time.Second)
140+
if !waitForState(listener.openCh, 5*time.Second) {
141+
logger.Warn("wait circuit breaker Open timeout")
142+
}
143+
logger.Info("wait circuit breaker HalfOpen window")
144+
timer := time.NewTimer(retryTimeout + 200*time.Millisecond)
145+
<-timer.C
111146
CallGreetFunConcurrently(svc.GreetWithChanceOfError, "hello world", 1, 300)
112-
time.Sleep(10 * time.Second)
147+
if !waitForState(listener.halfOpenCh, 5*time.Second) {
148+
logger.Warn("wait circuit breaker HalfOpen timeout")
149+
}
150+
if !waitForState(listener.closedCh, 5*time.Second) {
151+
logger.Warn("wait circuit breaker Closed timeout")
152+
}
113153
}
114154

115155
func CallGreetFunConcurrently(f GreetFun, req string, numberOfConcurrently, frequency int) (pass int64, block int64) {

0 commit comments

Comments
 (0)