supporting heartbeat in cloudevents subscribe stream

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

Author: skeeey

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

@@ -130,49 +130,19 @@ func (p *Protocol) OpenInbound(ctx context.Context) error {
 	subCtx, cancel := context.WithCancel(ctx)
 	heartbeatCh := make(chan *pbv1.CloudEvent, 1)
 
-	// start to receive the events
-	go func() {
-		// close the heartbeat channel when stopping to receive the events
-		defer close(heartbeatCh)
-
-		for {
-			evt, err := subClient.Recv()
-			if err != nil {
-				select {
-				case p.reconnectErrorChan <- fmt.Errorf("subscribe stream failed: %w", err):
-				default:
-					// error chan is full, do nothing
-				}
-				return
-			}
-
-			if evt.Type == types.HeartbeatCloudEventsType {
-				select {
-				case heartbeatCh <- evt:
-				case <-ctx.Done():
-					return
-				}
-				continue
-			}
-
-			select {
-			case p.incoming <- evt:
-			case <-ctx.Done():
-				return
-			}
-		}
-	}()
+	// start to receive the events from stream
+	go p.startEventsReceiver(subCtx, subClient, heartbeatCh)
 
 	// start to watch the stream heartbeat
-	go p.startHeartbeatWatcher(ctx, heartbeatCh)
+	go p.startHeartbeatWatcher(subCtx, heartbeatCh)
 
 	// Wait until external or internal context done
 	select {
 	case <-subCtx.Done():
 	case <-p.closeChan:
 	}
 
-	// ensure the event receiving and heartbeat checking goroutines are done
+	// ensure the event receiver  and heartbeat watcher are done
 	cancel()
 
 	logger.Infof("Close grpc client connection")
@@ -198,10 +168,36 @@ func (p *Protocol) Close(ctx context.Context) error {
 	return nil
 }
 
-func (p *Protocol) startHeartbeatWatcher(
-	ctx context.Context,
-	heartbeatCh <-chan *pbv1.CloudEvent,
-) {
+func (p *Protocol) startEventsReceiver(ctx context.Context,
+	subClient pbv1.CloudEventService_SubscribeClient, heartbeatCh chan *pbv1.CloudEvent) {
+	for {
+		evt, err := subClient.Recv()
+		if err != nil {
+			select {
+			case p.reconnectErrorChan <- fmt.Errorf("subscribe stream failed: %w", err):
+			default:
+			}
+			return
+		}
+
+		if evt.Type == types.HeartbeatCloudEventsType {
+			select {
+			case heartbeatCh <- evt:
+			case <-ctx.Done():
+				return
+			}
+			continue
+		}
+
+		select {
+		case p.incoming <- evt:
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+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 {
@@ -218,18 +214,15 @@ func (p *Protocol) startHeartbeatWatcher(
 		}
 	}
 
-	// if no heartbeat was received duration the serverHealthinessTimeout,
-	// send the timeout error to reconnectErrorChan
+	// if no heartbeat was received duration the serverHealthinessTimeout, send the
+	// timeout error to reconnectErrorChan
 	timeout := p.serverHealthinessTimeout
 	timer := time.NewTimer(timeout)
 	defer timer.Stop()
 
 	for {
 		select {
-		case msg, ok := <-heartbeatCh:
-			if !ok {
-				return
-			}
+		case msg := <-heartbeatCh:
 			logger.Debugf("heartbeat received %v", msg)
 
 			// reset timer safely
@@ -240,15 +233,12 @@ func (p *Protocol) startHeartbeatWatcher(
 				}
 			}
 			timer.Reset(timeout)
-
 		case <-timer.C:
-			// timeout reached: no heartbeat received
 			select {
-			case p.reconnectErrorChan <- fmt.Errorf("stream timeout: no message received for %v", timeout):
+			case p.reconnectErrorChan <- fmt.Errorf("stream timeout: no heartbeat received for %v", timeout):
 			default:
 			}
 			return
-
 		case <-ctx.Done():
 			return
 		}