func (sq *Queue) processNextWorkItem() bool {
	obj, shutdown := sq.Workqueue.Get()
	if shutdown {
		return false
	}
	defer sq.Workqueue.Done(obj)
	action, ok := obj.(Action)
	if !ok {
		sq.Workqueue.Forget(obj)
		sq.Metrics.QueueSizeGaugeMetric(sq.Workqueue.Len())
		sq.Logger.Errorf("got an unknown item of type %T in the queue", obj)
		return true
	}
	err := action.Sync(sq.KubeClient, sq.Logger)
	if err == nil {
		sq.Workqueue.Forget(obj)
		action.SetMetrics(sq.KubeClient, sq.Metrics, sq.Logger)
		sq.Metrics.QueueSizeGaugeMetric(sq.Workqueue.Len())
		sq.Logger.Infof("Successfully handled: %s", action)
		return true
	}
	action.SetMetricError(sq.Metrics)
	numRequeues := sq.Workqueue.NumRequeues(obj)
	if numRequeues < queueMaxRetries {
		sq.Logger.Errorf("Error handling %s: %v. Number of requeues: %d. Requeuing.", action, err, numRequeues)
		sq.Workqueue.AddRateLimited(obj)
		sq.Metrics.QueueSizeGaugeMetric(sq.Workqueue.Len())
		return true
	}
	sq.Workqueue.Forget(obj)
	sq.Logger.Errorf("Dropping %s out of the queue because we failed to handle the item %d times: %v", action, queueMaxRetries, err)
	sq.Metrics.QueueSizeGaugeMetric(sq.Workqueue.Len())
	return true
}
processNextWorkItem will read a single work item off the workqueue and
attempt to process it, by calling the syncHandler.
package sync
import (
	"time"
	localmetrics "github.com/projectcontour/gimbal/pkg/metrics"
	"github.com/sirupsen/logrus"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/util/workqueue"
)
const (
	actionAdd       = "add"
	actionUpdate    = "update"
	actionDelete    = "delete"
	queueMaxRetries = 3
)
type Queue struct {
	Logger      *logrus.Logger
	KubeClient  kubernetes.Interface
	Workqueue   workqueue.RateLimitingInterface
	Threadiness int
	Metrics     localmetrics.DiscovererMetrics
}
func NewQueue(logger *logrus.Logger, kubeClient kubernetes.Interface,
	threadiness int, metrics localmetrics.DiscovererMetrics) Queue {
	return Queue{
		KubeClient:  kubeClient,
		Logger:      logger,
		Workqueue:   workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "syncqueue"),
		Threadiness: threadiness,
		Metrics:     metrics,
	}
}
type Action interface {
	Sync(kube kubernetes.Interface, logger *logrus.Logger) error
	ObjectMeta() *metav1.ObjectMeta
	SetMetrics(gimbalKubeClient kubernetes.Interface, lm localmetrics.DiscovererMetrics, logger *logrus.Logger)
	SetMetricError(metrics localmetrics.DiscovererMetrics)
	GetActionType() string
}
func (sq *Queue) Enqueue(action Action) {
	sq.Workqueue.AddRateLimited(action)
	sq.Metrics.QueueSizeGaugeMetric(sq.Workqueue.Len())
}
func (sq *Queue) Run(stopCh <-chan struct{}) {
	defer runtime.HandleCrash()
	defer sq.Workqueue.ShutDown()
	sq.Logger.Infof("Starting queue workers")
	for i := 0; i < sq.Threadiness; i++ {
		go wait.Until(sq.runWorker, time.Second, stopCh)
	}
	sq.Logger.Infof("Started workers")
	<-stopCh
	sq.Logger.Infof("Shutting down workers")
}
func (sq *Queue) runWorker() {
	for sq.processNextWorkItem() {
	}
}