Skip to content

Commit b3f743a

Browse files
qiujian16qiujian16
committed
Add tests and some refactor
Signed-off-by: Jian Qiu <jqiu@redhat.com>
1 parent 3f24eac commit b3f743a

10 files changed

Lines changed: 712 additions & 11 deletions

File tree

‎pkg/cloudevents/generic/options/grpc/options_test.go‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ func TestBuildGRPCOptionsFromFlags(t *testing.T) {
3636
name: "customized options",
3737
config: "{\"url\":\"test\"}",
3838
expectedOptions: &GRPCOptions{
39-
&GRPCDialer{
39+
Dialer: &GRPCDialer{
4040
URL: "test",
4141
KeepAliveOptions: KeepAliveOptions{
4242
Enable: false,
@@ -45,13 +45,14 @@ func TestBuildGRPCOptionsFromFlags(t *testing.T) {
4545
PermitWithoutStream: false,
4646
},
4747
},
48+
ServerHealthinessTimeout: nil,
4849
},
4950
},
5051
{
5152
name: "customized options with yaml format",
5253
config: "url: test",
5354
expectedOptions: &GRPCOptions{
54-
&GRPCDialer{
55+
Dialer: &GRPCDialer{
5556
URL: "test",
5657
KeepAliveOptions: KeepAliveOptions{
5758
Enable: false,
@@ -60,13 +61,14 @@ func TestBuildGRPCOptionsFromFlags(t *testing.T) {
6061
PermitWithoutStream: false,
6162
},
6263
},
64+
ServerHealthinessTimeout: nil,
6365
},
6466
},
6567
{
6668
name: "customized options with keepalive",
6769
config: "{\"url\":\"test\",\"keepAliveConfig\":{\"enable\":true,\"time\":10s,\"timeout\":5s,\"permitWithoutStream\":true}}",
6870
expectedOptions: &GRPCOptions{
69-
&GRPCDialer{
71+
Dialer: &GRPCDialer{
7072
URL: "test",
7173
KeepAliveOptions: KeepAliveOptions{
7274
Enable: true,
@@ -75,6 +77,7 @@ func TestBuildGRPCOptionsFromFlags(t *testing.T) {
7577
PermitWithoutStream: true,
7678
},
7779
},
80+
ServerHealthinessTimeout: nil,
7881
},
7982
},
8083
}

‎pkg/cloudevents/generic/options/grpc/protocol/option.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,3 +40,8 @@ func WithReconnectErrorOption(reconnectError chan error, interval time.Duration)
4040
return nil
4141
}
4242
}
43+
44+
// WithHealthCheck is an alias for WithReconnectErrorOption for better clarity
45+
func WithHealthCheck(interval time.Duration, errorChan chan error) Option {
46+
return WithReconnectErrorOption(errorChan, interval)
47+
}

‎pkg/cloudevents/generic/options/grpc/protocol/option_test.go‎

Lines changed: 195 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,195 @@
1+
package protocol
2+
3+
import (
4+
"net"
5+
"testing"
6+
"time"
7+
8+
"google.golang.org/grpc"
9+
"google.golang.org/grpc/test/bufconn"
10+
)
11+
12+
func TestWithSubscribeOption(t *testing.T) {
13+
lis := bufconn.Listen(1024)
14+
defer lis.Close()
15+
16+
s := grpc.NewServer()
17+
defer s.Stop()
18+
19+
conn, err := grpc.Dial("bufnet", grpc.WithInsecure(), grpc.WithDialer(func(string, time.Duration) (net.Conn, error) {
20+
return lis.Dial()
21+
}))
22+
if err != nil {
23+
t.Fatalf("Failed to create connection: %v", err)
24+
}
25+
defer conn.Close()
26+
27+
// Test valid subscribe option
28+
subscribeOpt := &SubscribeOption{
29+
Source: "test-source",
30+
ClusterName: "test-cluster",
31+
DataType: "test-type",
32+
}
33+
34+
p, err := NewProtocol(conn, WithSubscribeOption(subscribeOpt))
35+
if err != nil {
36+
t.Fatalf("Failed to create protocol: %v", err)
37+
}
38+
39+
if p.subscribeOption != subscribeOpt {
40+
t.Error("Subscribe option was not set correctly")
41+
}
42+
43+
if p.subscribeOption.Source != "test-source" {
44+
t.Errorf("Expected source 'test-source', got '%s'", p.subscribeOption.Source)
45+
}
46+
47+
// Test nil subscribe option
48+
_, err = NewProtocol(conn, WithSubscribeOption(nil))
49+
if err == nil {
50+
t.Error("Expected error for nil subscribe option")
51+
}
52+
}
53+
54+
func TestWithReconnectErrorOption(t *testing.T) {
55+
lis := bufconn.Listen(1024)
56+
defer lis.Close()
57+
58+
s := grpc.NewServer()
59+
defer s.Stop()
60+
61+
conn, err := grpc.Dial("bufnet", grpc.WithInsecure(), grpc.WithDialer(func(string, time.Duration) (net.Conn, error) {
62+
return lis.Dial()
63+
}))
64+
if err != nil {
65+
t.Fatalf("Failed to create connection: %v", err)
66+
}
67+
defer conn.Close()
68+
69+
// Test valid reconnect error option
70+
errorChan := make(chan error, 1)
71+
interval := 5 * time.Second
72+
73+
p, err := NewProtocol(conn, WithReconnectErrorOption(errorChan, interval))
74+
if err != nil {
75+
t.Fatalf("Failed to create protocol: %v", err)
76+
}
77+
78+
if p.reconnectErrorChan != errorChan {
79+
t.Error("Reconnect error channel was not set correctly")
80+
}
81+
82+
if p.checkInterval != interval {
83+
t.Errorf("Expected interval %v, got %v", interval, p.checkInterval)
84+
}
85+
86+
// Test with zero interval (should default to 20 seconds)
87+
p, err = NewProtocol(conn, WithReconnectErrorOption(errorChan, 0))
88+
if err != nil {
89+
t.Fatalf("Failed to create protocol: %v", err)
90+
}
91+
92+
if p.checkInterval != 20*time.Second {
93+
t.Errorf("Expected default interval 20s, got %v", p.checkInterval)
94+
}
95+
96+
// Test with negative interval (should default to 20 seconds)
97+
p, err = NewProtocol(conn, WithReconnectErrorOption(errorChan, -1*time.Second))
98+
if err != nil {
99+
t.Fatalf("Failed to create protocol: %v", err)
100+
}
101+
102+
if p.checkInterval != 20*time.Second {
103+
t.Errorf("Expected default interval 20s, got %v", p.checkInterval)
104+
}
105+
106+
// Test nil error channel
107+
_, err = NewProtocol(conn, WithReconnectErrorOption(nil, interval))
108+
if err == nil {
109+
t.Error("Expected error for nil reconnect error channel")
110+
}
111+
}
112+
113+
func TestWithHealthCheck(t *testing.T) {
114+
lis := bufconn.Listen(1024)
115+
defer lis.Close()
116+
117+
s := grpc.NewServer()
118+
defer s.Stop()
119+
120+
conn, err := grpc.Dial("bufnet", grpc.WithInsecure(), grpc.WithDialer(func(string, time.Duration) (net.Conn, error) {
121+
return lis.Dial()
122+
}))
123+
if err != nil {
124+
t.Fatalf("Failed to create connection: %v", err)
125+
}
126+
defer conn.Close()
127+
128+
// Test WithHealthCheck (which is an alias for WithReconnectErrorOption)
129+
errorChan := make(chan error, 1)
130+
interval := 3 * time.Second
131+
132+
p, err := NewProtocol(conn, WithHealthCheck(interval, errorChan))
133+
if err != nil {
134+
t.Fatalf("Failed to create protocol: %v", err)
135+
}
136+
137+
if p.reconnectErrorChan != errorChan {
138+
t.Error("Health check error channel was not set correctly")
139+
}
140+
141+
if p.checkInterval != interval {
142+
t.Errorf("Expected health check interval %v, got %v", interval, p.checkInterval)
143+
}
144+
145+
// Test nil error channel for health check
146+
_, err = NewProtocol(conn, WithHealthCheck(interval, nil))
147+
if err == nil {
148+
t.Error("Expected error for nil health check error channel")
149+
}
150+
}
151+
152+
func TestMultipleOptions(t *testing.T) {
153+
lis := bufconn.Listen(1024)
154+
defer lis.Close()
155+
156+
s := grpc.NewServer()
157+
defer s.Stop()
158+
159+
conn, err := grpc.Dial("bufnet", grpc.WithInsecure(), grpc.WithDialer(func(string, time.Duration) (net.Conn, error) {
160+
return lis.Dial()
161+
}))
162+
if err != nil {
163+
t.Fatalf("Failed to create connection: %v", err)
164+
}
165+
defer conn.Close()
166+
167+
// Test applying multiple options
168+
subscribeOpt := &SubscribeOption{
169+
Source: "test-source",
170+
DataType: "test-type",
171+
}
172+
errorChan := make(chan error, 1)
173+
interval := 2 * time.Second
174+
175+
p, err := NewProtocol(conn,
176+
WithSubscribeOption(subscribeOpt),
177+
WithHealthCheck(interval, errorChan),
178+
)
179+
if err != nil {
180+
t.Fatalf("Failed to create protocol: %v", err)
181+
}
182+
183+
// Verify both options were applied
184+
if p.subscribeOption != subscribeOpt {
185+
t.Error("Subscribe option was not set correctly")
186+
}
187+
188+
if p.reconnectErrorChan != errorChan {
189+
t.Error("Health check error channel was not set correctly")
190+
}
191+
192+
if p.checkInterval != interval {
193+
t.Errorf("Expected health check interval %v, got %v", interval, p.checkInterval)
194+
}
195+
}

‎pkg/cloudevents/generic/options/grpc/protocol/protocol.go‎

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -191,17 +191,20 @@ func (p *Protocol) healthCheck(ctx context.Context) {
191191
return
192192
}
193193

194-
last := time.Now()
195194
recvErr := make(chan error, 1)
195+
healthReceived := make(chan struct{}, 1)
196196
go func() {
197197
for {
198198
resp, err := stream.Recv()
199199
if err != nil {
200200
recvErr <- err
201201
return
202202
}
203-
last = time.Now()
204203
logger.Infof("Received server health status %s", resp.Status)
204+
select {
205+
case healthReceived <- struct{}{}:
206+
default:
207+
}
205208
}
206209
}()
207210

@@ -218,14 +221,17 @@ func (p *Protocol) healthCheck(ctx context.Context) {
218221
default:
219222
}
220223
return
224+
case <-healthReceived:
225+
// Reset the ticker when we receive a health update
226+
ticker.Stop()
227+
ticker = time.NewTicker(p.checkInterval)
221228
case <-ticker.C:
222-
if time.Since(last) > p.checkInterval {
223-
select {
224-
case p.reconnectErrorChan <- fmt.Errorf("timeout waiting for health check"):
225-
default:
226-
}
227-
return
229+
// Timeout waiting for health check
230+
select {
231+
case p.reconnectErrorChan <- fmt.Errorf("timeout waiting for health check"):
232+
default:
228233
}
234+
return
229235
}
230236
}
231237
}

0 commit comments

Comments
 (0)