implement priority queue

Signed-off-by: Troy Connor <[email protected]>

Author: troy0820

pkg/handler/enqueue_mapped.go

@@ -82,23 +82,8 @@ func (e *enqueueRequestsFromMapFunc[object, request]) Create(
 	evt event.TypedCreateEvent[object],
 	q workqueue.TypedRateLimitingInterface[request],
 ) {
-	wq, ok := q.(workqueue.TypedRateLimitingInterface[reconcile.Request])
-	if ok {
-		evnt, ok := any(e).(event.TypedCreateEvent[client.Object])
-		if ok {
-			priorityQueue, isPriorityQueue := wq.(priorityqueue.PriorityQueue[reconcile.Request])
-			if isPriorityQueue {
-				item := reconcile.Request{NamespacedName: types.NamespacedName{
-					Name:      evnt.Object.GetName(),
-					Namespace: evnt.Object.GetNamespace(),
-				}}
-				addToQueueCreate(priorityQueue, evnt, item)
-				return
-			}
-			reqs := map[request]empty{}
-			e.mapAndEnqueue(ctx, q, evt.Object, reqs)
-		}
-	}
+	reqs := map[request]empty{}
+	e.mapAndEnqueueCreate(ctx, q, evt.Object, reqs)
 }
 
 // Update implements EventHandler.
@@ -107,43 +92,9 @@ func (e *enqueueRequestsFromMapFunc[object, request]) Update(
 	evt event.TypedUpdateEvent[object],
 	q workqueue.TypedRateLimitingInterface[request],
 ) {
-	wq, ok := q.(workqueue.TypedRateLimitingInterface[reconcile.Request])
-	if ok {
-		evnt, ok := any(e).(event.TypedUpdateEvent[client.Object])
-		if ok {
-			switch {
-			case !isNil(evt.ObjectNew):
-				priorityQueue, isPriorityQueue := wq.(priorityqueue.PriorityQueue[reconcile.Request])
-				if isPriorityQueue {
-					item := reconcile.Request{NamespacedName: types.NamespacedName{
-						Name:      evnt.ObjectNew.GetName(),
-						Namespace: evnt.ObjectNew.GetNamespace(),
-					}}
-					addToQueueUpdate(priorityQueue, evnt, item)
-					return
-				}
-				reqs := map[request]empty{}
-				e.mapAndEnqueue(ctx, q, evt.ObjectNew, reqs)
-			case !isNil(evt.ObjectOld):
-				priorityQueue, isPriorityQueue := wq.(priorityqueue.PriorityQueue[reconcile.Request])
-				if isPriorityQueue {
-					item := reconcile.Request{NamespacedName: types.NamespacedName{
-						Name:      evnt.ObjectOld.GetName(),
-						Namespace: evnt.ObjectOld.GetNamespace(),
-					}}
-					addToQueueUpdate(priorityQueue, evnt, item)
-					return
-				}
-				reqs := map[request]empty{}
-				e.mapAndEnqueue(ctx, q, evt.ObjectOld, reqs)
-			default:
-				enqueueLog.Error(nil, "UpdateEvent received with no metadata", "event", evt)
-			}
-		}
-	}
 	reqs := map[request]empty{}
-	e.mapAndEnqueue(ctx, q, evt.ObjectOld, reqs)
-	e.mapAndEnqueue(ctx, q, evt.ObjectNew, reqs)
+	e.mapAndEnqueueUpdate(ctx, q, evt.ObjectOld, reqs)
+	e.mapAndEnqueueUpdate(ctx, q, evt.ObjectNew, reqs)
 }
 
 // Delete implements EventHandler.
@@ -170,8 +121,74 @@ func (e *enqueueRequestsFromMapFunc[object, request]) mapAndEnqueue(ctx context.
 	for _, req := range e.toRequests(ctx, o) {
 		_, ok := reqs[req]
 		if !ok {
+			reqs := map[request]empty{}
 			q.Add(req)
 			reqs[req] = empty{}
 		}
 	}
 }
+
+func (e *enqueueRequestsFromMapFunc[object, request]) mapAndEnqueueCreate(ctx context.Context, q workqueue.TypedRateLimitingInterface[request], o object, reqs map[request]empty) {
+	for _, req := range e.toRequests(ctx, o) {
+		_, ok := reqs[req]
+		if !ok {
+			wq, ok := q.(workqueue.TypedRateLimitingInterface[reconcile.Request])
+			if ok {
+				evnt, ok := any(e).(event.TypedCreateEvent[client.Object])
+				if ok {
+					priorityQueue, isPriorityQueue := wq.(priorityqueue.PriorityQueue[reconcile.Request])
+					if isPriorityQueue {
+						item := reconcile.Request{NamespacedName: types.NamespacedName{
+							Name:      evnt.Object.GetName(),
+							Namespace: evnt.Object.GetNamespace(),
+						}}
+						addToQueueCreate(priorityQueue, evnt, item)
+						continue
+					}
+				}
+				reqs := map[request]empty{}
+				q.Add(req)
+				reqs[req] = empty{}
+			}
+		}
+	}
+}
+
+func (e *enqueueRequestsFromMapFunc[object, request]) mapAndEnqueueUpdate(ctx context.Context, q workqueue.TypedRateLimitingInterface[request], o object, reqs map[request]empty) {
+	for _, req := range e.toRequests(ctx, o) {
+		_, ok := reqs[req]
+		if !ok {
+			wq, ok := q.(workqueue.TypedRateLimitingInterface[reconcile.Request])
+			if ok {
+				evnt, ok := any(e).(event.TypedUpdateEvent[client.Object])
+				if ok {
+					priorityQueue, isPriorityQueue := wq.(priorityqueue.PriorityQueue[reconcile.Request])
+					if !isPriorityQueue {
+						reqs := map[request]empty{}
+						q.Add(req)
+						reqs[req] = empty{}
+						continue
+					}
+					switch {
+					case !isNil(evnt.ObjectNew):
+						item := reconcile.Request{NamespacedName: types.NamespacedName{
+							Name:      evnt.ObjectNew.GetName(),
+							Namespace: evnt.ObjectNew.GetNamespace(),
+						}}
+						addToQueueUpdate(priorityQueue, evnt, item)
+						continue
+					case !isNil(evnt.ObjectOld):
+						item := reconcile.Request{NamespacedName: types.NamespacedName{
+							Name:      evnt.ObjectOld.GetName(),
+							Namespace: evnt.ObjectOld.GetNamespace(),
+						}}
+						addToQueueUpdate(priorityQueue, evnt, item)
+						continue
+					default:
+						enqueueLog.Error(nil, "UpdateEvent received with no metadata", "event", evnt)
+					}
+				}
+			}
+		}
+	}
+}