expose the list options for resync (#304)

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

cloudevents/generic/agentclient.go

@@ -65,10 +65,10 @@ func NewCloudEventAgentClient[T ResourceObject](
 	}, nil
 }
 
-// Resync the resources spec by sending a spec resync request from an agent to all sources.
-func (c *CloudEventAgentClient[T]) Resync(ctx context.Context) error {
-	// list the resource objects that are maintained by the current agent from all sources
-	objs, err := c.lister.List(types.ListOptions{ClusterName: c.clusterName, Source: types.SourceAll})
+// Resync the resources spec by sending a spec resync request from an agent to sources with list options.
+func (c *CloudEventAgentClient[T]) Resync(ctx context.Context, listOptions types.ListOptions) error {
+	// list the resource objects that are maintained by the current agent with list options
+	objs, err := c.lister.List(listOptions)
 	if err != nil {
 		return err
 	}

cloudevents/generic/agentclient_test.go

@@ -62,7 +62,7 @@ func TestAgentResync(t *testing.T) {
 				t.Errorf("unexpected error %v", err)
 			}
 
-			if err := agent.Resync(context.TODO()); err != nil {
+			if err := agent.Resync(context.TODO(), types.ListOptions{}); err != nil {
 				t.Errorf("unexpected error %v", err)
 			}
 

cloudevents/generic/interface.go

@@ -54,7 +54,7 @@ type Codec[T ResourceObject] interface {
 
 type CloudEventsClient[T ResourceObject] interface {
 	// Resync the resources of one source/agent by sending resync request.
-	Resync(ctx context.Context) error
+	Resync(ctx context.Context, listOptions types.ListOptions) error
 
 	// Publish the resources spec/status event to the broker.
 	Publish(ctx context.Context, eventType types.CloudEventsType, obj T) error

cloudevents/generic/sourceclient.go

@@ -64,10 +64,10 @@ func NewCloudEventSourceClient[T ResourceObject](
 	}, nil
 }
 
-// Resync the resources status by sending a status resync request from a source to all clusters.
-func (c *CloudEventSourceClient[T]) Resync(ctx context.Context) error {
-	// list the resource objects that are maintained by the current source from all clusters
-	objs, err := c.lister.List(types.ListOptions{ClusterName: types.ClusterAll, Source: c.sourceID})
+// Resync the resources status by sending a status resync request from a source to clusters with list options.
+func (c *CloudEventSourceClient[T]) Resync(ctx context.Context, listOptions types.ListOptions) error {
+	// list the resource objects that are maintained by the current source with list options
+	objs, err := c.lister.List(listOptions)
 	if err != nil {
 		return err
 	}

cloudevents/generic/sourceclient_test.go

@@ -49,7 +49,7 @@ func TestSourceResync(t *testing.T) {
 				t.Errorf("unexpected error %v", err)
 			}
 
-			if err := source.Resync(context.TODO()); err != nil {
+			if err := source.Resync(context.TODO(), types.ListOptions{}); err != nil {
 				t.Errorf("unexpected error %v", err)
 			}
 

cloudevents/work/agent/client/manifestwork.go

@@ -80,7 +80,7 @@ func (c *ManifestWorkAgentClient) Get(ctx context.Context, name string, opts met
 func (c *ManifestWorkAgentClient) List(ctx context.Context, opts metav1.ListOptions) (*workv1.ManifestWorkList, error) {
 	klog.V(4).Infof("sync manifestworks")
 	// send resync request to fetch manifestworks from source when the ManifestWorkInformer starts
-	if err := c.cloudEventsClient.Resync(ctx); err != nil {
+	if err := c.cloudEventsClient.Resync(ctx, types.ListOptions{}); err != nil {
 		return nil, err
 	}
 

test/integration/cloudevents/cloudevents_test.go

@@ -78,7 +78,7 @@ var _ = ginkgo.Describe("Cloudevents clients test", func() {
 			}, 10*time.Second, 1*time.Second).Should(gomega.Succeed())
 
 			ginkgo.By("resync the status from source")
-			err = mqttSourceCloudEventsClient.Resync(context.TODO())
+			err = mqttSourceCloudEventsClient.Resync(context.TODO(), types.ListOptions{})
 			gomega.Expect(err).ToNot(gomega.HaveOccurred())
 
 			gomega.Eventually(func() error {