address comments.

Signed-off-by: morvencao <[email protected]>

Author: morvencao

pkg/cloudevents/generic/options/builder/optionsbuilder_test.go

@@ -106,6 +106,11 @@ func TestBuildCloudEventsSourceOptions(t *testing.T) {
 					AgentEvents:    "projects/test-project/subscriptions/agentevents-source1",
 					AgentBroadcast: "projects/test-project/subscriptions/agentbroadcast-source1",
 				},
+				KeepaliveSettings: &pubsub.KeepaliveSettings{
+					Time:                5 * time.Minute,
+					Timeout:             20 * time.Second,
+					PermitWithoutStream: false,
+				},
 			},
 		},
 	}
@@ -167,6 +172,11 @@ func TestBuildCloudEventsAgentOptions(t *testing.T) {
 					SourceEvents:    "projects/test-project/subscriptions/sourceevents-cluster1",
 					SourceBroadcast: "projects/test-project/subscriptions/sourcebroadcast-cluster1",
 				},
+				KeepaliveSettings: &pubsub.KeepaliveSettings{
+					Time:                5 * time.Minute,
+					Timeout:             20 * time.Second,
+					PermitWithoutStream: false,
+				},
 			},
 		},
 	}

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

@@ -1,44 +1,14 @@
 package pubsub
 
 import (
-	"context"
-	"fmt"
-
-	"cloud.google.com/go/pubsub/v2"
-	cloudevents "github.com/cloudevents/sdk-go/v2"
-	"google.golang.org/api/option"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
-	"k8s.io/klog/v2"
-
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options"
-	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 )
 
-var _ options.CloudEventTransport = &pubsubAgentTransport{}
-
-// pubsubAgentTransport is a CloudEventTransport implementation for Pub/Sub for agents.
-type pubsubAgentTransport struct {
-	PubSubOptions
-	clusterName string
-	client      *pubsub.Client
-	// Publisher for status updates
-	publisher *pubsub.Publisher
-	// Publisher for resync broadcasts
-	resyncPublisher *pubsub.Publisher
-	// Subscriber for spec updates
-	subscriber *pubsub.Subscriber
-	// Subscriber for resync broadcasts
-	resyncSubscriber *pubsub.Subscriber
-	// TODO: handle error channel
-	errorChan chan error
-}
-
 // NewAgentOptions creates a new CloudEventsAgentOptions for Pub/Sub.
 func NewAgentOptions(pubsubOptions *PubSubOptions,
 	clusterName, agentID string) *options.CloudEventsAgentOptions {
 	return &options.CloudEventsAgentOptions{
-		CloudEventsTransport: &pubsubAgentTransport{
+		CloudEventsTransport: &pubsubTransport{
 			PubSubOptions: *pubsubOptions,
 			clusterName:   clusterName,
 			errorChan:     make(chan error),
@@ -47,119 +17,3 @@ func NewAgentOptions(pubsubOptions *PubSubOptions,
 		ClusterName: clusterName,
 	}
 }
-
-func (o *pubsubAgentTransport) Connect(ctx context.Context) error {
-	options := []option.ClientOption{}
-	if o.CredentialsFile != "" {
-		options = append(options, option.WithCredentialsFile(o.CredentialsFile))
-	}
-	if o.Endpoint != "" {
-		options = append(options, option.WithEndpoint(o.Endpoint))
-		if o.CredentialsFile == "" {
-			// use the insecure connection for local development/testing when no credentials are provided
-			pubsubConn, err := grpc.NewClient(o.Endpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))
-			if err != nil {
-				return err
-			}
-			options = append(options, option.WithGRPCConn(pubsubConn))
-		}
-	}
-
-	client, err := pubsub.NewClient(ctx, o.ProjectID, options...)
-	if err != nil {
-		return err
-	}
-
-	// initialize pubsub client, publisher and subscriber
-	o.client = client
-	o.publisher = client.Publisher(o.Topics.AgentEvents)
-	o.resyncPublisher = client.Publisher(o.Topics.AgentBroadcast)
-	o.subscriber = client.Subscriber(o.Subscriptions.SourceEvents)
-	o.resyncSubscriber = client.Subscriber(o.Subscriptions.SourceBroadcast)
-
-	return nil
-}
-
-func (o *pubsubAgentTransport) Send(ctx context.Context, evt cloudevents.Event) error {
-	msg, err := Encode(evt)
-	if err != nil {
-		return err
-	}
-
-	eventType, err := types.ParseCloudEventsType(evt.Context.GetType())
-	if err != nil {
-		return fmt.Errorf("unsupported event type %s, %v", eventType, err)
-	}
-
-	// determine publisher based on event type
-	var result *pubsub.PublishResult
-	if eventType.Action == types.ResyncRequestAction {
-		result = o.resyncPublisher.Publish(ctx, msg)
-	} else {
-		result = o.publisher.Publish(ctx, msg)
-	}
-
-	// block until the result is returned
-	_, err = result.Get(ctx)
-	return err
-}
-
-func (o *pubsubAgentTransport) Receive(ctx context.Context, fn options.ReceiveHandlerFn) error {
-	// create error channels for both subscribers
-	subscriberErrChan := make(chan error, 1)
-	resyncSubscriberErrChan := make(chan error, 1)
-
-	// start the subscriber for spec updates
-	go func() {
-		err := o.subscriber.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
-			evt, err := Decode(msg)
-			if err != nil {
-				// also send ACK on decode error since redelivery won't fix it.
-				klog.Errorf("failed to decode pubsub message to resource spec event: %v", err)
-			} else {
-				fn(evt)
-			}
-			// send ACK after all receiver handlers complete.
-			msg.Ack()
-		})
-		if err != nil {
-			subscriberErrChan <- fmt.Errorf("subscriber is interrupted by error: %w", err)
-		}
-	}()
-
-	// start the resync subscriber for broadcast messages
-	go func() {
-		err := o.resyncSubscriber.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
-			evt, err := Decode(msg)
-			if err != nil {
-				// also send ACK on decode error since redelivery won't fix it.
-				klog.Errorf("failed to decode pubsub message to resource statusresync event: %v", err)
-			} else {
-				fn(evt)
-			}
-			// send ACK after all receiver handlers complete.
-			msg.Ack()
-		})
-		if err != nil {
-			resyncSubscriberErrChan <- fmt.Errorf("resync subscriber is interrupted by error: %w", err)
-		}
-	}()
-
-	// wait for either subscriber to error or context cancellation
-	select {
-	case err := <-subscriberErrChan:
-		return err
-	case err := <-resyncSubscriberErrChan:
-		return err
-	case <-ctx.Done():
-		return ctx.Err()
-	}
-}
-
-func (o *pubsubAgentTransport) Close(ctx context.Context) error {
-	return o.client.Close()
-}
-
-func (o *pubsubAgentTransport) ErrorChan() <-chan error {
-	return o.errorChan
-}

pkg/cloudevents/generic/options/pubsub/agentoptions_test.go

@@ -70,9 +70,9 @@ func TestNewAgentOptions(t *testing.T) {
 				t.Fatal("expected CloudEventsTransport to be set, got nil")
 			}
 
-			transport, ok := opts.CloudEventsTransport.(*pubsubAgentTransport)
+			transport, ok := opts.CloudEventsTransport.(*pubsubTransport)
 			if !ok {
-				t.Fatalf("expected transport to be *pubsubAgentTransport, got %T", opts.CloudEventsTransport)
+				t.Fatalf("expected transport to be *pubsubTransport, got %T", opts.CloudEventsTransport)
 			}
 
 			if transport.clusterName != c.clusterName {
@@ -112,7 +112,7 @@ func TestPubsubAgentTransport_Structure(t *testing.T) {
 	}
 
 	opts := NewAgentOptions(pubsubOpts, "test-cluster", "test-agent")
-	transport := opts.CloudEventsTransport.(*pubsubAgentTransport)
+	transport := opts.CloudEventsTransport.(*pubsubTransport)
 
 	// Verify topics are set correctly for agent
 	if transport.Topics.AgentEvents != pubsubOpts.Topics.AgentEvents {

pkg/cloudevents/generic/options/pubsub/codec.go

@@ -112,6 +112,7 @@ func Decode(msg *pubsub.Message) (cloudevents.Event, error) {
 	// Get the spec version to access attributes
 	version := specs.Version(specVersion)
 
+	var dataContentType string
 	// Set all CloudEvent context attributes from message attributes
 	for _, attr := range version.Attributes() {
 		var value string
@@ -120,6 +121,7 @@ func Decode(msg *pubsub.Message) (cloudevents.Event, error) {
 		// Special handling for datacontenttype - use "Content-Type" without "ce-" prefix
 		if attr.Kind() == spec.DataContentType {
 			value, ok = msg.Attributes[contentType]
+			dataContentType = value
 		} else {
 			value, ok = msg.Attributes[attr.PrefixedName()]
 		}
@@ -152,8 +154,12 @@ func Decode(msg *pubsub.Message) (cloudevents.Event, error) {
 		}
 	}
 
+	if dataContentType == "" {
+		// default data content type be "application/JSON"
+		dataContentType = cloudevents.ApplicationJSON
+	}
 	// Set the data from the message
-	if err := evt.SetData(cloudevents.ApplicationJSON, msg.Data); err != nil {
+	if err := evt.SetData(dataContentType, msg.Data); err != nil {
 		return cloudevents.Event{}, fmt.Errorf("failed to set event data: %v", err)
 	}
 

pkg/cloudevents/generic/options/pubsub/codec_test.go

@@ -1,6 +1,7 @@
 package pubsub
 
 import (
+	"strings"
 	"testing"
 	"time"
 
@@ -341,7 +342,7 @@ func TestDecode(t *testing.T) {
 			if c.expectedErr {
 				if err == nil {
 					t.Errorf("expected error, but got none")
-				} else if c.errorContains != "" && !contains(err.Error(), c.errorContains) {
+				} else if c.errorContains != "" && !strings.Contains(err.Error(), c.errorContains) {
 					t.Errorf("expected error to contain %q, got %q", c.errorContains, err.Error())
 				}
 				return

pkg/cloudevents/generic/options/pubsub/options.go

@@ -5,6 +5,7 @@ import (
 	"os"
 	"regexp"
 	"strings"
+	"time"
 
 	"gopkg.in/yaml.v2"
 
@@ -14,11 +15,13 @@ import (
 
 // PubSubOptions holds the options that are used to build Pub/Sub client.
 type PubSubOptions struct {
-	Endpoint        string
-	ProjectID       string
-	CredentialsFile string
-	Topics          types.Topics
-	Subscriptions   types.Subscriptions
+	Endpoint          string
+	ProjectID         string
+	CredentialsFile   string
+	Topics            types.Topics
+	Subscriptions     types.Subscriptions
+	KeepaliveSettings *KeepaliveSettings
+	ReceiveSettings   *ReceiveSettings
 }
 
 // PubSubConfig holds the information needed to connect to Google Cloud Pub/Sub.
@@ -44,7 +47,45 @@ type PubSubConfig struct {
 	// Required: must be provided to specify the subscriptions to receive events from.
 	Subscriptions *types.Subscriptions `json:"subscriptions,omitempty" yaml:"subscriptions,omitempty"`
 
-	// TODO: Add support for additional Pub/Sub settings (e.g., message ordering, receive options).
+	// (Optional) KeepaliveSettings configures the keepalive parameters for Pub/Sub client.
+	KeepaliveSettings *KeepaliveSettings `json:"keepaliveSettings,omitempty" yaml:"keepaliveSettings,omitempty"`
+
+	// (Optional) ReceiveSettings configures the pubsub subscriber's receive settings.
+	ReceiveSettings *ReceiveSettings `json:"receiveSettings,omitempty" yaml:"receiveSettings,omitempty"`
+}
+
+// KeepaliveSettings defines gRPC keepalive options for the Pub/Sub client.
+type KeepaliveSettings struct {
+	// Time between pings when there’s no activity, minimum is 10s, default: 5m.
+	Time time.Duration `json:"time,omitempty" yaml:"time,omitempty"`
+
+	// Wait time for a ping response before closing the connection, default: 20s.
+	Timeout time.Duration `json:"timeout,omitempty" yaml:"timeout,omitempty"`
+
+	// If true, send pings even when no RPCs are active, default: false.
+	PermitWithoutStream bool `json:"permitWithoutStream,omitempty" yaml:"permitWithoutStream,omitempty"`
+}
+
+// ReceiveSettings defines how the Pub/Sub subscriber receives and processes messages.
+type ReceiveSettings struct {
+	// MaxExtension is the maximum period for which the Subscriber should
+	// automatically extend the ack deadline for each message.
+	MaxExtension time.Duration `json:"maxExtension,omitempty" yaml:"maxExtension,omitempty"`
+
+	// MaxDurationPerAckExtension is the maximum duration per lease extension.
+	MaxDurationPerAckExtension time.Duration `json:"maxDurationPerAckExtension,omitempty" yaml:"maxDurationPerAckExtension,omitempty"`
+
+	// MinDurationPerAckExtension is the minimum duration per lease extension.
+	MinDurationPerAckExtension time.Duration `json:"minDurationPerAckExtension,omitempty" yaml:"minDurationPerAckExtension,omitempty"`
+
+	// MaxOutstandingMessages is the maximum number of unprocessed messages.
+	MaxOutstandingMessages int `json:"maxOutstandingMessages,omitempty" yaml:"maxOutstandingMessages,omitempty"`
+
+	// MaxOutstandingBytes is the maximum size of unprocessed messages.
+	MaxOutstandingBytes int `json:"maxOutstandingBytes,omitempty" yaml:"maxOutstandingBytes,omitempty"`
+
+	// NumGoroutines is the number of StreamingPull streams to pull messages from the subscription.
+	NumGoroutines int `json:"numGoroutines,omitempty" yaml:"numGoroutines,omitempty"`
 }
 
 // LoadConfig loads the Pub/Sub configuration from a file.
@@ -79,13 +120,29 @@ func BuildPubSubOptionsFromFlags(configPath string) (*PubSubOptions, error) {
 		return nil, err
 	}
 
-	return &PubSubOptions{
+	options := &PubSubOptions{
 		Endpoint:        config.Endpoint,
 		ProjectID:       config.ProjectID,
 		CredentialsFile: config.CredentialsFile,
 		Topics:          *config.Topics,
 		Subscriptions:   *config.Subscriptions,
-	}, nil
+		// enable keepalive by default
+		KeepaliveSettings: &KeepaliveSettings{
+			Time:                5 * time.Minute,
+			Timeout:             20 * time.Second,
+			PermitWithoutStream: false,
+		},
+	}
+
+	if config.KeepaliveSettings != nil {
+		options.KeepaliveSettings = config.KeepaliveSettings
+	}
+
+	if config.ReceiveSettings != nil {
+		options.ReceiveSettings = config.ReceiveSettings
+	}
+
+	return options, nil
 }
 
 func validateTopicsAndSubscriptions(topics *types.Topics, subscriptions *types.Subscriptions, projectID string) error {