func (f *FIFO) Add(obj interface{}) error {
	id, err := f.keyFunc(obj)
	if err != nil {
		return KeyError{obj, err}
	}
	f.lock.Lock()
	defer f.lock.Unlock()
	f.populated = true
	if _, exists := f.items[id]; !exists {
		f.queue = append(f.queue, id)
	}
	f.items[id] = obj
	f.cond.Broadcast()
	return nil
}
Add inserts an item, and puts it in the queue. The item is only enqueued
if it doesn't already exist in the set.
package cache
import (
	"errors"
	"sync"
	"k8s.io/apimachinery/pkg/util/sets"
)
type PopProcessFunc func(interface{}) error
type ErrRequeue struct {
	Err error
}
var FIFOClosedError error = errors.New("DeltaFIFO: manipulating with closed queue")
func (e ErrRequeue) Error() string {
	if e.Err == nil {
		return "the popped item should be requeued without returning an error"
	}
	return e.Err.Error()
}
type Queue interface {
	Store
	Pop(PopProcessFunc) (interface{}, error)
	AddIfNotPresent(interface{}) error
	HasSynced() bool
	Close()
}
func Pop(queue Queue) interface{} {
	var result interface{}
	queue.Pop(func(obj interface{}) error {
		result = obj
		return nil
	})
	return result
}
type FIFO struct {
	lock sync.RWMutex
	cond sync.Cond
	items map[string]interface{}
	queue []string
	populated bool
	initialPopulationCount int
	keyFunc KeyFunc
	closed     bool
	closedLock sync.Mutex
}
var (
	_ = Queue(&FIFO{})  
)
func (f *FIFO) Close() {
	f.closedLock.Lock()
	defer f.closedLock.Unlock()
	f.closed = true
	f.cond.Broadcast()
}
func (f *FIFO) HasSynced() bool {
	f.lock.Lock()
	defer f.lock.Unlock()
	return f.populated && f.initialPopulationCount == 0
}