Skip to content

Commit

Permalink
replace deprecated queue creation
Browse files Browse the repository at this point in the history
Signed-off-by: Frank Jogeleit <[email protected]>
  • Loading branch information
fjogeleit committed Sep 30, 2024
1 parent 0650ac0 commit 0dfbaa0
Show file tree
Hide file tree
Showing 3 changed files with 8 additions and 9 deletions.
2 changes: 1 addition & 1 deletion pkg/config/resolver.go
Original file line number Diff line number Diff line change
Expand Up @@ -222,7 +222,7 @@ func (r *Resolver) Queue() (*kubernetes.Queue, error) {

return kubernetes.NewQueue(
kubernetes.NewDebouncer(1*time.Minute, r.EventPublisher()),
workqueue.NewRateLimitingQueueWithConfig(workqueue.DefaultControllerRateLimiter(), workqueue.RateLimitingQueueConfig{
workqueue.NewTypedRateLimitingQueueWithConfig(workqueue.DefaultTypedControllerRateLimiter[string](), workqueue.TypedRateLimitingQueueConfig[string]{
Name: "report-queue",
}),
client,
Expand Down
6 changes: 3 additions & 3 deletions pkg/kubernetes/policy_report_client_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ func Test_PolicyReportWatcher(t *testing.T) {

queue := kubernetes.NewQueue(
kubernetes.NewDebouncer(0, publisher),
workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test-queue"),
workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[string]()),
restClient.Wgpolicyk8sV1alpha2(),
report.NewSourceFilter(nil, nil, []report.SourceValidation{}),
result.NewReconditioner(nil),
Expand Down Expand Up @@ -89,7 +89,7 @@ func Test_ClusterPolicyReportWatcher(t *testing.T) {

queue := kubernetes.NewQueue(
kubernetes.NewDebouncer(0, publisher),
workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test-queue"),
workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[string]()),
restClient.Wgpolicyk8sV1alpha2(),
report.NewSourceFilter(nil, nil, []report.SourceValidation{}),
result.NewReconditioner(nil),
Expand Down Expand Up @@ -131,7 +131,7 @@ func Test_HasSynced(t *testing.T) {

queue := kubernetes.NewQueue(
kubernetes.NewDebouncer(0, report.NewEventPublisher()),
workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test-queue"),
workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[string]()),
restClient.Wgpolicyk8sV1alpha2(),
report.NewSourceFilter(nil, nil, []report.SourceValidation{}),
result.NewReconditioner(nil),
Expand Down
9 changes: 4 additions & 5 deletions pkg/kubernetes/queue.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ import (
)

type Queue struct {
queue workqueue.RateLimitingInterface
queue workqueue.TypedRateLimitingInterface[string]
client v1alpha2.Wgpolicyk8sV1alpha2Interface
reconditioner *result.Reconditioner
debouncer Debouncer
Expand Down Expand Up @@ -57,11 +57,10 @@ func (q *Queue) runWorker() {
}

func (q *Queue) processNextItem() bool {
obj, quit := q.queue.Get()
key, quit := q.queue.Get()
if quit {
return false
}
key := obj.(string)
defer q.queue.Done(key)

namespace, name, err := cache.SplitMetaNamespaceKey(key)
Expand Down Expand Up @@ -127,7 +126,7 @@ func (q *Queue) processNextItem() bool {
return true
}

func (q *Queue) handleErr(err error, key interface{}) {
func (q *Queue) handleErr(err error, key string) {
if err == nil {
q.queue.Forget(key)
return
Expand All @@ -148,7 +147,7 @@ func (q *Queue) handleErr(err error, key interface{}) {

func NewQueue(
debouncer Debouncer,
queue workqueue.RateLimitingInterface,
queue workqueue.TypedRateLimitingInterface[string],
client v1alpha2.Wgpolicyk8sV1alpha2Interface,
filter *report.SourceFilter,
reconditioner *result.Reconditioner,
Expand Down

0 comments on commit 0dfbaa0

Please sign in to comment.