Add requeue At and requeue after to new controller framework

pkg/controller/v2/controller_test.go

@@ -55,6 +55,35 @@ func TestRequeue(t *testing.T) {
 	assert.NoError(t, controller.Reconcile("bar"))
 	<-done
 }
+func TestRqueueAfter(t *testing.T) {
+	var requeued atomic.Bool
+	done := make(chan struct{})
+	controller := NewController[testID](func(ctx context.Context, request Request[testID]) Directive[testID] {
+		if requeued.CompareAndSwap(false, true) {
+			return request.Requeue().After(time.Second)
+		}
+		close(done)
+		return request.Ack()
+	}, WithParallelism(10))
+	defer controller.Stop()
+	assert.NoError(t, controller.Reconcile("bar"))
+	<-done
+}
+
+func TestRqueueAt(t *testing.T) {
+	var requeued atomic.Bool
+	done := make(chan struct{})
+	controller := NewController[testID](func(ctx context.Context, request Request[testID]) Directive[testID] {
+		if requeued.CompareAndSwap(false, true) {
+			return request.Requeue().At(time.Now().Add(time.Second))
+		}
+		close(done)
+		return request.Ack()
+	}, WithParallelism(10))
+	defer controller.Stop()
+	assert.NoError(t, controller.Reconcile("bar"))
+	<-done
+}
 
 func TestRetry(t *testing.T) {
 	var retried atomic.Bool

pkg/controller/v2/reconciler.go

@@ -101,6 +101,50 @@ func (r *Requeue[I]) Do(controller *Controller[I]) {
 	go controller.enqueue(r.request)
 }
 
+// After retries the request after the given delay
+func (r *Requeue[I]) After(delay time.Duration) *RequeueAfter[I] {
+	return &RequeueAfter[I]{
+		Requeue: r,
+		delay:   delay,
+	}
+}
+
+// At retries the request at the given time
+func (r *Requeue[I]) At(t time.Time) *RequeueAt[I] {
+	return &RequeueAt[I]{
+		Requeue: r,
+		t:       t,
+	}
+}
+
+// RequeueAfter requeues a reconciliation request after the given delay
+type RequeueAfter[I ID] struct {
+	*Requeue[I]
+	delay time.Duration
+}
+
+// Do executes the controller directive
+func (r *RequeueAfter[I]) Do(controller *Controller[I]) {
+	controller.Log.Debugw("Requeueing request", "Request.ID", r.request.ID, "delay", r.delay)
+	time.AfterFunc(r.delay, func() {
+		controller.enqueue(r.request)
+	})
+}
+
+// RequeueAt requeue a reconciliation request at a specific time
+type RequeueAt[I ID] struct {
+	*Requeue[I]
+	t time.Time
+}
+
+// Do executes the controller directive
+func (r *RequeueAt[I]) Do(controller *Controller[I]) {
+	controller.Log.Debugw("Requeueing request", "Request.ID", r.request.ID, "Time", r.t)
+	time.AfterFunc(time.Until(r.t), func() {
+		controller.enqueue(r.request)
+	})
+}
+
 // Fail fails a reconciliation request
 type Fail[I ID] struct {
 	request Request[I]