supporting heartbeat in cloudevents subscribe stream

Signed-off-by: Wei Liu <liuweixa@redhat.com>

Author: skeeey

pkg/cloudevents/generic/options/grpc/agentoptions.go

@@ -47,7 +47,7 @@ func (o *grpcAgentOptions) Protocol(ctx context.Context, dataType types.CloudEve
 	}
 
 	if o.ServerHealthinessTimeout != nil {
-		opts = append(opts, protocol.WithServerHealthinessTimeout(*o.ServerHealthinessTimeout))
+		opts = append(opts, protocol.WithServerHealthinessTimeout(o.ServerHealthinessTimeout))
 	}
 
 	receiver, err := o.GetCloudEventsProtocol(

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

@@ -29,19 +29,21 @@ func WithSubscribeOption(subscribeOpt *SubscribeOption) Option {
 func WithReconnectErrorChan(errorChan chan error) Option {
 	return func(p *Protocol) error {
 		if errorChan == nil {
-			return fmt.Errorf("the errorChan must not be nil")
+			return fmt.Errorf("the error channel must not be nil")
 		}
 		p.reconnectErrorChan = errorChan
 		return nil
 	}
 }
 
-func WithServerHealthinessTimeout(timeout time.Duration) Option {
+func WithServerHealthinessTimeout(timeout *time.Duration) Option {
 	return func(p *Protocol) error {
-		if timeout <= 0 {
-			p.serverHealthinessTimeout = 20 * time.Second // by default, using 20s
+		if timeout != nil {
+			if *timeout <= 0 {
+				return fmt.Errorf("the server healthiness timeout must be greater than 0")
+			}
+			p.serverHealthinessTimeout = timeout
 		}
-		p.serverHealthinessTimeout = timeout
 		return nil
 	}
 }

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

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"google.golang.org/grpc/credentials/insecure"
+	"k8s.io/utils/ptr"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/test/bufconn"
@@ -86,7 +87,7 @@ func TestProtocol_Success(t *testing.T) {
 			DataType:    "io.open-cluster-management.test",
 		}),
 		WithReconnectErrorChan(reconnectErrorChan),
-		WithServerHealthinessTimeout(5*time.Second),
+		WithServerHealthinessTimeout(ptr.To(5*time.Second)),
 	)
 	if err != nil {
 		t.Fatal(err)
@@ -123,7 +124,7 @@ func TestProtocol_Timeout(t *testing.T) {
 			DataType:    "io.open-cluster-management.test",
 		}),
 		WithReconnectErrorChan(reconnectErrorChan),
-		WithServerHealthinessTimeout(2*time.Second),
+		WithServerHealthinessTimeout(ptr.To(5*time.Second)),
 	)
 	if err != nil {
 		t.Fatal(err)

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

@@ -35,8 +35,8 @@ type Protocol struct {
 	reconnectErrorChan chan error
 
 	// serverHealthinessTimeout is the max duration that client will reconnect if no server healthiness
-	// status is received in this duration, if it is 0, this will be disabled.
-	serverHealthinessTimeout time.Duration
+	// status is received in this duration.
+	serverHealthinessTimeout *time.Duration
 }
 
 var (
@@ -199,8 +199,8 @@ func (p *Protocol) startEventsReceiver(ctx context.Context,
 
 func (p *Protocol) startHeartbeatWatcher(ctx context.Context, heartbeatCh <-chan *pbv1.CloudEvent) {
 	logger := cecontext.LoggerFrom(ctx)
-	// if serverHealthinessTimeout is disable, ignore the server heartbeat timeout
-	if p.serverHealthinessTimeout == 0 {
+	// if serverHealthinessTimeout is not set, ignore the heartbeat timeout check
+	if p.serverHealthinessTimeout == nil {
 		for {
 			select {
 			case msg := <-heartbeatCh:
@@ -213,7 +213,7 @@ func (p *Protocol) startHeartbeatWatcher(ctx context.Context, heartbeatCh <-chan
 
 	// if no heartbeat was received duration the serverHealthinessTimeout, send the
 	// timeout error to reconnectErrorChan
-	timeout := p.serverHealthinessTimeout
+	timeout := *p.serverHealthinessTimeout
 	timer := time.NewTimer(timeout)
 	defer timer.Stop()
 

pkg/cloudevents/generic/options/grpc/sourceoptions.go

@@ -42,7 +42,7 @@ func (o *gRPCSourceOptions) Protocol(ctx context.Context, dataType types.CloudEv
 	}
 
 	if o.ServerHealthinessTimeout != nil {
-		opts = append(opts, protocol.WithServerHealthinessTimeout(*o.ServerHealthinessTimeout))
+		opts = append(opts, protocol.WithServerHealthinessTimeout(o.ServerHealthinessTimeout))
 	}
 
 	receiver, err := o.GetCloudEventsProtocol(