Add work queue

A work queue is used to start workloads asynchronously. The workload itself always run once at a time.

Use cases:

* parsing ingress: the ingress parser and config builder is a workload which runs when the k8s state changes
* acme signer: a v0.9 feature, sign a new certificate is a workload which runs when a new certificate need to be issued

pkg/utils/queue.go

@@ -0,0 +1,70 @@
+/*
+Copyright 2019 The HAProxy Ingress Controller Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package utils
+
+import (
+	"k8s.io/client-go/util/workqueue"
+)
+
+// Queue ...
+type Queue interface {
+	Run()
+	Add(item interface{})
+	ShutDown()
+}
+
+type queue struct {
+	workqueue *workqueue.Type
+	running   chan struct{}
+	sync      func(item interface{})
+}
+
+// NewQueue ...
+func NewQueue(sync func(item interface{})) Queue {
+	return &queue{
+		workqueue: workqueue.New(),
+		sync:      sync,
+	}
+}
+
+func (q *queue) Run() {
+	if q.running != nil {
+		// queue already running
+		return
+	}
+	q.running = make(chan struct{})
+	for {
+		item, shutdown := q.workqueue.Get()
+		if shutdown {
+			close(q.running)
+			return
+		}
+		q.sync(item)
+		q.workqueue.Done(item)
+	}
+}
+
+func (q *queue) Add(item interface{}) {
+	q.workqueue.Add(item)
+}
+
+func (q *queue) ShutDown() {
+	q.workqueue.ShutDown()
+	if q.running != nil {
+		<-q.running
+	}
+}

pkg/utils/queue_test.go

@@ -0,0 +1,83 @@
+/*
+Copyright 2019 The HAProxy Ingress Controller Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package utils
+
+import (
+	"reflect"
+	"testing"
+	"time"
+)
+
+type task struct {
+	steps []string
+}
+
+func TestQueueNotRunning(t *testing.T) {
+	q := NewQueue(nil)
+	q.ShutDown()
+}
+
+func TestQueueAlreadyRunning(t *testing.T) {
+	q := NewQueue(nil)
+	go q.Run()
+	time.Sleep(100 * time.Millisecond)
+	q.Run() // test fail if this call blocks, the test will timeout
+	q.ShutDown()
+}
+
+func TestQueueShutdown(t *testing.T) {
+	q := NewQueue(func(item interface{}) { time.Sleep(200 * time.Millisecond) })
+	stopped := false
+	go func() {
+		q.Run()
+		stopped = true
+	}()
+	q.Add(nil)
+	time.Sleep(100 * time.Millisecond)
+	q.ShutDown()
+	if !stopped {
+		t.Error("queue is still running")
+	}
+}
+
+func TestQueueRun(t *testing.T) {
+	task := &task{}
+	q := NewQueue(task.sync)
+	go q.Run()
+	q.Add("a1")
+	time.Sleep(150 * time.Millisecond)
+	task.step("s1")
+	q.Add("a2")
+	time.Sleep(150 * time.Millisecond)
+	task.step("s2")
+	q.ShutDown()
+	task.step("s3")
+	expected := []string{"a1-1", "s1", "a1-2", "a2-1", "s2", "a2-2", "s3"}
+	if !reflect.DeepEqual(task.steps, expected) {
+		t.Errorf("steps differ, expected: %+v; actual: %+v", expected, task.steps)
+	}
+}
+
+func (t *task) sync(item interface{}) {
+	t.step(item.(string) + "-1")
+	time.Sleep(250 * time.Millisecond)
+	t.step(item.(string) + "-2")
+}
+
+func (t *task) step(id string) {
+	t.steps = append(t.steps, id)
+}