grpc transport v2

Signed-off-by: Wei Liu <[email protected]>

Author: skeeey

pkg/cloudevents/server/grpc/broker.go

@@ -217,7 +217,6 @@ func (bkr *GRPCBroker) Subscribe(subReq *pbv1.SubscriptionRequest, subServer pbv
 		// WARNING: don't use "pbEvt, err := pb.ToProto(evt)" to convert cloudevent to protobuf
 		pbEvt := &pbv1.CloudEvent{}
 		if err := grpcprotocol.WritePBMessage(context.TODO(), binding.ToMessage(evt), pbEvt); err != nil {
-			// return the error to requeue the event if converting to protobuf fails (e.g., due to invalid cloudevent).
 			return fmt.Errorf("failed to convert cloudevent to protobuf for resource(%s): %v", evt.ID(), err)
 		}
 
@@ -226,7 +225,12 @@ func (bkr *GRPCBroker) Subscribe(subReq *pbv1.SubscriptionRequest, subServer pbv
 		select {
 		case eventCh <- pbEvt:
 		case <-ctx.Done():
-			return status.Error(codes.Unavailable, "stream context canceled")
+			// The context of the stream has been canceled or completed.
+			// This could happen if:
+			// - The client closed the connection or canceled the stream.
+			// - The server closed the stream, potentially due to a shutdown.
+			// No error is returned here because the stream closure is expected.
+			return nil
 		}
 
 		return nil