address comments.

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

Author: morvencao

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

@@ -108,7 +108,13 @@ 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,
+				},
 			},
+			expectedTransportType: "*pubsub.pubsubTransport",
 		},
 	}
 
@@ -138,6 +144,7 @@ func TestBuildCloudEventsAgentOptions(t *testing.T) {
 					Timeout:    60 * time.Second,
 				},
 			},
+			expectedTransportType: "*mqtt.mqttAgentTransport",
 		},
 		{
 			name:       "grpc config",
@@ -154,6 +161,7 @@ func TestBuildCloudEventsAgentOptions(t *testing.T) {
 					},
 				},
 			},
+			expectedTransportType: "*grpc.grpcTransport",
 		},
 		{
 			name:       "pubsub config",
@@ -169,7 +177,13 @@ 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,
+				},
 			},
+			expectedTransportType: "*pubsub.pubsubTransport",
 		},
 	}
 
@@ -200,16 +214,15 @@ func assertSourceOptions(t *testing.T, c buildingCloudEventsOptionTestCase) {
 		t.Errorf("unexpected config %v, %v", config, c.expectedOptions)
 	}
 
-	options, err := BuildCloudEventsSourceOptions(config, "agent1", sourceId, types.CloudEventsDataType{})
+	options, err := BuildCloudEventsSourceOptions(config, "client", sourceId, types.CloudEventsDataType{})
 	if err != nil {
 		t.Errorf("unexpected error %v", err)
 	}
 
-	optionsRaw, _ := json.Marshal(options)
-	expectedRaw, _ := json.Marshal(c.expectedOptions)
+	tt := reflect.TypeOf(options.CloudEventsTransport)
 
-	if !strings.Contains(string(optionsRaw), string(expectedRaw)) {
-		t.Errorf("the results %v\n does not contain the original options %v\n", string(optionsRaw), string(expectedRaw))
+	if tt.String() != c.expectedTransportType {
+		t.Errorf("expected %s, but got %s", c.expectedTransportType, tt)
 	}
 }
 
@@ -224,7 +237,7 @@ func assertAgentOptions(t *testing.T, c buildingCloudEventsOptionTestCase) {
 		t.Errorf("unexpected config %v, %v", config, c.expectedOptions)
 	}
 
-	options, err := BuildCloudEventsAgentOptions(config, "cluster1", "agent1", types.CloudEventsDataType{})
+	options, err := BuildCloudEventsAgentOptions(config, "cluster1", "client", types.CloudEventsDataType{})
 	if err != nil {
 		t.Errorf("unexpected error %v", err)
 	}

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