Skip to content

Commit a0114c4

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

12 files changed

Lines changed: 849 additions & 19 deletions

File tree

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

Lines changed: 24 additions & 5 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",
67-
config: "{\"url\":\"test\",\"keepAliveConfig\":{\"enable\":true,\"time\":10s,\"timeout\":5s,\"permitWithoutStream\":true}}",
69+
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,23 @@ func TestBuildGRPCOptionsFromFlags(t *testing.T) {
7577
PermitWithoutStream: true,
7678
},
7779
},
80+
ServerHealthinessTimeout: nil,
81+
},
82+
},
83+
{
84+
name: "customized options with server healthiness timeout",
85+
config: "{\"url\":\"test\",\"serverHealthinessTimeout\":\"1m\"}",
86+
expectedOptions: &GRPCOptions{
87+
Dialer: &GRPCDialer{
88+
URL: "test",
89+
KeepAliveOptions: KeepAliveOptions{
90+
Enable: false,
91+
Time: 30 * time.Second,
92+
Timeout: 10 * time.Second,
93+
PermitWithoutStream: false,
94+
},
95+
},
96+
ServerHealthinessTimeout: func() *time.Duration { d := time.Minute; return &d }(),
7897
},
7998
},
8099
}
@@ -85,7 +104,7 @@ func TestBuildGRPCOptionsFromFlags(t *testing.T) {
85104
if err != nil {
86105
t.Fatal(err)
87106
}
88-
defer os.Remove(file.Name())
107+
t.Cleanup(func() { _ = os.Remove(file.Name()) })
89108

90109
options, err := BuildGRPCOptionsFromFlags(file.Name())
91110
if err != nil {

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

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

Lines changed: 21 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ type Protocol struct {
3333

3434
closeChan chan struct{}
3535

36-
// errorChan is to send an error messsage to restart the connection
36+
// errorChan is to send an error message to restart the connection
3737
reconnectErrorChan chan error
3838
checkInterval time.Duration
3939
}
@@ -184,24 +184,33 @@ func (p *Protocol) healthCheck(ctx context.Context) {
184184
healthClient := healthpb.NewHealthClient(p.clientConn)
185185
stream, err := healthClient.Watch(watchCtx, &healthpb.HealthCheckRequest{Service: ""})
186186
if err != nil {
187+
logger.Errorf("failed to watch health check, %v", err)
187188
select {
188189
case p.reconnectErrorChan <- err:
189190
default:
190191
}
191192
return
192193
}
193194

194-
last := time.Now()
195195
recvErr := make(chan error, 1)
196+
healthReceived := make(chan struct{}, 1)
196197
go func() {
197198
for {
198199
resp, err := stream.Recv()
199200
if err != nil {
201+
if s, ok := status.FromError(err); ok && s.Code() == codes.Unimplemented {
202+
logger.Warnf("grpc-health not implemented on server; skipping health check")
203+
return
204+
}
205+
logger.Errorf("failed to receive health check, %v", err)
200206
recvErr <- err
201207
return
202208
}
203-
last = time.Now()
204209
logger.Infof("Received server health status %s", resp.Status)
210+
select {
211+
case healthReceived <- struct{}{}:
212+
default:
213+
}
205214
}
206215
}()
207216

@@ -218,14 +227,17 @@ func (p *Protocol) healthCheck(ctx context.Context) {
218227
default:
219228
}
220229
return
230+
case <-healthReceived:
231+
// Reset the ticker when we receive a health update
232+
ticker.Stop()
233+
ticker = time.NewTicker(p.checkInterval)
221234
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
235+
// Timeout waiting for health check
236+
select {
237+
case p.reconnectErrorChan <- fmt.Errorf("timeout waiting for health check"):
238+
default:
228239
}
240+
return
229241
}
230242
}
231243
}

0 commit comments

Comments
 (0)