refactor resume command

Signed-off-by: Somtochi Onyekwere <somtochionyekwere@gmail.com>

cmd/flux/resume_source_bucket.go

@@ -17,20 +17,8 @@ limitations under the License.
 package main
 
 import (
-	"context"
-	"fmt"
-
-	"github.com/fluxcd/flux2/internal/utils"
-	"github.com/fluxcd/pkg/apis/meta"
-
-	"github.com/spf13/cobra"
-	apimeta "k8s.io/apimachinery/pkg/api/meta"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/wait"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-
 	sourcev1 "github.com/fluxcd/source-controller/api/v1beta1"
+	"github.com/spf13/cobra"
 )
 
 var resumeSourceBucketCmd = &cobra.Command{
@@ -40,76 +28,20 @@ var resumeSourceBucketCmd = &cobra.Command{
 	Example: `  # Resume reconciliation for an existing Bucket
   flux resume source bucket podinfo
 `,
-	RunE: resumeSourceBucketCmdRun,
+	RunE: resumeCommand{
+		apiType: bucketType,
+		object:  &bucketAdapter{&sourcev1.Bucket{}},
+	}.run,
 }
 
 func init() {
 	resumeSourceCmd.AddCommand(resumeSourceBucketCmd)
 }
 
-func resumeSourceBucketCmdRun(cmd *cobra.Command, args []string) error {
-	if len(args) < 1 {
-		return fmt.Errorf("source name is required")
-	}
-	name := args[0]
-
-	ctx, cancel := context.WithTimeout(context.Background(), rootArgs.timeout)
-	defer cancel()
-
-	kubeClient, err := utils.KubeClient(rootArgs.kubeconfig, rootArgs.kubecontext)
-	if err != nil {
-		return err
-	}
-
-	namespacedName := types.NamespacedName{
-		Namespace: rootArgs.namespace,
-		Name:      name,
-	}
-	var bucket sourcev1.Bucket
-	err = kubeClient.Get(ctx, namespacedName, &bucket)
-	if err != nil {
-		return err
-	}
-
-	logger.Actionf("resuming source %s in %s namespace", name, rootArgs.namespace)
-	bucket.Spec.Suspend = false
-	if err := kubeClient.Update(ctx, &bucket); err != nil {
-		return err
-	}
-	logger.Successf("source resumed")
-
-	logger.Waitingf("waiting for Bucket reconciliation")
-	if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout,
-		isBucketResumed(ctx, kubeClient, namespacedName, &bucket)); err != nil {
-		return err
-	}
-	logger.Successf("Bucket reconciliation completed")
-
-	logger.Successf("fetched revision %s", bucket.Status.Artifact.Revision)
-	return nil
+func (obj bucketAdapter) getObservedGeneration() int64 {
+	return obj.Bucket.Status.ObservedGeneration
 }
 
-func isBucketResumed(ctx context.Context, kubeClient client.Client,
-	namespacedName types.NamespacedName, bucket *sourcev1.Bucket) wait.ConditionFunc {
-	return func() (bool, error) {
-		err := kubeClient.Get(ctx, namespacedName, bucket)
-		if err != nil {
-			return false, err
-		}
-
-		// Confirm the state we are observing is for the current generation
-		if bucket.Generation != bucket.Status.ObservedGeneration {
-			return false, nil
-		}
-
-		if c := apimeta.FindStatusCondition(bucket.Status.Conditions, meta.ReadyCondition); c != nil {
-			switch c.Status {
-			case metav1.ConditionTrue:
-				return true, nil
-			case metav1.ConditionFalse:
-				return false, fmt.Errorf(c.Message)
-			}
-		}
-		return false, nil
-	}
+func (obj bucketAdapter) setUnsuspended() {
+	obj.Bucket.Spec.Suspend = false
 }