remove ce sdk-go deps (v2)

pkg/cloudevents/clients/options/generic.go

@@ -21,6 +21,7 @@ type GenericClientOptions[T generic.ResourceObject] struct {
 	clusterName  string
 	subscription bool
 	resync       bool
+	v2           bool
 }
 
 // NewGenericClientOptions create a GenericClientOptions
@@ -86,6 +87,11 @@ func (o *GenericClientOptions[T]) WithResyncEnabled(resync bool) *GenericClientO
 	return o
 }
 
+func (o *GenericClientOptions[T]) V2(enabled bool) *GenericClientOptions[T] {
+	o.v2 = enabled
+	return o
+}
+
 func (o *GenericClientOptions[T]) ClusterName() string {
 	return o.clusterName
 }
@@ -111,20 +117,39 @@ func (o *GenericClientOptions[T]) AgentClient(ctx context.Context) (*generic.Clo
 		o.watcherStore = store.NewAgentInformerWatcherStore[T]()
 	}
 
-	options, err := generic.BuildCloudEventsAgentOptions(o.config, o.clusterName, o.clientID)
-	if err != nil {
-		return nil, err
-	}
+	var cloudEventsClient *generic.CloudEventAgentClient[T]
+	if o.v2 {
+		options, err := generic.BuildCloudEventsAgentOptionsV2(o.config, o.clusterName, o.clientID)
+		if err != nil {
+			return nil, err
+		}
 
-	cloudEventsClient, err := generic.NewCloudEventAgentClient(
-		ctx,
-		options,
-		store.NewAgentWatcherStoreLister(o.watcherStore),
-		statushash.StatusHash,
-		o.codec,
-	)
-	if err != nil {
-		return nil, err
+		cloudEventsClient, err = generic.NewCloudEventAgentClientV2(
+			ctx,
+			options,
+			store.NewAgentWatcherStoreLister(o.watcherStore),
+			statushash.StatusHash,
+			o.codec,
+		)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		options, err := generic.BuildCloudEventsAgentOptions(o.config, o.clusterName, o.clientID)
+		if err != nil {
+			return nil, err
+		}
+
+		cloudEventsClient, err = generic.NewCloudEventAgentClient(
+			ctx,
+			options,
+			store.NewAgentWatcherStoreLister(o.watcherStore),
+			statushash.StatusHash,
+			o.codec,
+		)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	if o.subscription {

pkg/cloudevents/generic/agentclient.go

@@ -14,14 +14,15 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/payload"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
+	optionsv2 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/v2/options"
 )
 
 // CloudEventAgentClient is a client for an agent to resync/send/receive its resources with cloud events.
 //
 // An agent is a component that handles the deployment of requested resources on the managed cluster and status report
 // to the source.
 type CloudEventAgentClient[T ResourceObject] struct {
-	*baseClient
+	baseClientInterface
 	lister           Lister[T]
 	codec            Codec[T]
 	statusHashGetter StatusHashGetter[T]
@@ -56,19 +57,48 @@ func NewCloudEventAgentClient[T ResourceObject](
 	}
 
 	return &CloudEventAgentClient[T]{
-		baseClient:       baseClient,
-		lister:           lister,
-		codec:            codec,
-		statusHashGetter: statusHashGetter,
-		agentID:          agentOptions.AgentID,
-		clusterName:      agentOptions.ClusterName,
+		baseClientInterface: baseClient,
+		lister:              lister,
+		codec:               codec,
+		statusHashGetter:    statusHashGetter,
+		agentID:             agentOptions.AgentID,
+		clusterName:         agentOptions.ClusterName,
+	}, nil
+}
+
+func NewCloudEventAgentClientV2[T ResourceObject](
+	ctx context.Context,
+	agentOptions *optionsv2.CloudEventsAgentOptions,
+	lister Lister[T],
+	statusHashGetter StatusHashGetter[T],
+	codec Codec[T],
+) (*CloudEventAgentClient[T], error) {
+	baseClient := &baseClientV2{
+		clientID:  agentOptions.AgentID,
+		transport: agentOptions.EventTransport,
+		// TODO move the ratelimiter to a package
+		cloudEventsRateLimiter: NewRateLimiter(options.EventRateLimit{}),
+		reconnectedChan:        make(chan struct{}),
+	}
+
+	if err := baseClient.connect(ctx); err != nil {
+		return nil, err
+	}
+
+	return &CloudEventAgentClient[T]{
+		baseClientInterface: baseClient,
+		lister:              lister,
+		codec:               codec,
+		statusHashGetter:    statusHashGetter,
+		agentID:             agentOptions.AgentID,
+		clusterName:         agentOptions.ClusterName,
 	}, nil
 }
 
 // ReconnectedChan returns a chan which indicates the source/agent client is reconnected.
 // The source/agent client callers should consider sending a resync request when receiving this signal.
 func (c *CloudEventAgentClient[T]) ReconnectedChan() <-chan struct{} {
-	return c.reconnectedChan
+	return c.getReconnectedChan()
 }
 
 // Resync the resources spec by sending a spec resync request from the current to the given source.

pkg/cloudevents/generic/agentclient_test.go

@@ -68,7 +68,8 @@ func TestAgentResync(t *testing.T) {
 			stop := make(chan bool)
 
 			go func() {
-				err = agent.cloudEventsClient.StartReceiver(ctx, func(event cloudevents.Event) {
+				baseClient := agent.baseClientInterface.(*baseClient)
+				err = baseClient.cloudEventsClient.StartReceiver(ctx, func(event cloudevents.Event) {
 					eventChan <- receiveEvent{event: event}
 				})
 				if err != nil {
@@ -136,7 +137,8 @@ func TestAgentPublish(t *testing.T) {
 			eventChan := make(chan receiveEvent)
 			stop := make(chan bool)
 			go func() {
-				err = agent.cloudEventsClient.StartReceiver(ctx, func(event cloudevents.Event) {
+				baseClient := agent.baseClientInterface.(*baseClient)
+				err = baseClient.cloudEventsClient.StartReceiver(ctx, func(event cloudevents.Event) {
 					eventChan <- receiveEvent{event: event}
 				})
 				if err != nil {
@@ -292,7 +294,8 @@ func TestStatusResyncResponse(t *testing.T) {
 			mutex := &sync.Mutex{}
 
 			go func() {
-				_ = agent.cloudEventsClient.StartReceiver(ctx, func(event cloudevents.Event) {
+				baseClient := agent.baseClientInterface.(*baseClient)
+				_ = baseClient.cloudEventsClient.StartReceiver(ctx, func(event cloudevents.Event) {
 					mutex.Lock()
 					defer mutex.Unlock()
 					receivedEvents = append(receivedEvents, event)

pkg/cloudevents/generic/baseclient.go

@@ -47,6 +47,10 @@ type baseClient struct {
 	dataType               types.CloudEventsDataType
 }
 
+func (c *baseClient) getReconnectedChan() <-chan struct{} {
+	return c.reconnectedChan
+}
+
 func (c *baseClient) connect(ctx context.Context) error {
 	logger := klog.FromContext(ctx)