grpc transport v2

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

Author: skeeey

pkg/cloudevents/generic/options/v2/grpc/transport.go

@@ -214,9 +214,6 @@ func (t *grpcTransport) monitorConnectionState(ctx context.Context, conn *grpc.C
 		if !conn.WaitForStateChange(ctx, state) {
 			// the ctx is closed, stop this watch
 			logger.Info("stop watching grpc connection state")
-			if err := conn.Close(); err != nil {
-				logger.Error(err, "failed to close grpc connection")
-			}
 			return
 		}
 

pkg/cloudevents/server/grpc/broker.go

@@ -368,7 +368,6 @@ func (bkr *GRPCBroker) handleRes(
 	evt *cloudevents.Event,
 	t types.CloudEventsDataType,
 	action types.EventAction) error {
-	log := klog.FromContext(ctx)
 
 	bkr.mu.RLock()
 	defer bkr.mu.RUnlock()
@@ -386,16 +385,14 @@ func (bkr *GRPCBroker) handleRes(
 	}
 	clusterName := fmt.Sprintf("%s", clusterNameValue)
 
-	for id, subscriber := range bkr.subscribers {
+	for subID, subscriber := range bkr.subscribers {
 		// checks if the event should be processed by the current instance
 		// by verifying the resource consumer name is in the subscriber list, ensuring the
 		// event will be only processed when the consumer is subscribed to the current broker.
 		if subscriber.clusterName == clusterName && subscriber.dataType == t {
-			if err := subscriber.handler(ctx, id, evt); err != nil {
+			if err := subscriber.handler(ctx, subID, evt); err != nil {
 				return err
 			}
-		} else {
-			log.V(4).Info("skip the event since the agent is not subscribed.", "subID", id, "eventContext", evt.Context)
 		}
 	}
 	return nil