func (q *WrappedQueue) IsConsistent() bool {
	return q.consistent
}
IsConsistent returns whether the queue and mapping of MarshallableQueueItems have ever fallen out of sync.
package state
import (
	"sync"
	"github.com/golang-collections/go-datastructures/queue"
	"github.com/pkg/errors"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/google/simhospital/pkg/state/persist"
)
var ErrSyncerNotSet = errors.New("ItemSyncer not set; cannot load from the syncer")
type WrappedQueue struct {
	q        *queue.PriorityQueue
	m        map[string]MarshallableQueueItem
	mutex    sync.Mutex
	itemType string
	syncer persist.ItemSyncer
	consistent bool
}
func NewWrappedQueue(itemType string, syncer persist.ItemSyncer) (*WrappedQueue, error) {
	w := &WrappedQueue{
		q:          queue.NewPriorityQueue(100),
		m:          map[string]MarshallableQueueItem{},
		mutex:      sync.Mutex{},
		consistent: true,
		itemType:   itemType,
		syncer:     syncer,
	}
	if w.syncer != nil {
		if err := w.LoadFromSyncer(); err != nil {
			return w, errors.Wrap(err, "cannot load from syncer")
		}
	}
	return w, nil
}
type MarshallableQueueItem interface {
	persist.MarshallableItem
	Compare(queue.Item) int
}
func (q *WrappedQueue) LoadFromSyncer() error {
	if q.syncer == nil {
		return ErrSyncerNotSet
	}
	items, err := q.syncer.LoadAll()
	if err != nil {
		return err
	}
	queueItems := make([]MarshallableQueueItem, len(items))
	for i, item := range items {
		queueItems[i] = item.(MarshallableQueueItem)
	}
	return q.put(nil, queueItems...)
}
func (q *WrappedQueue) Len() int {
	q.mutex.Lock()
	defer q.mutex.Unlock()
	return len(q.m)
}
func (q *WrappedQueue) Put(items ...MarshallableQueueItem) error {
	q.mutex.Lock()
	defer q.mutex.Unlock()
	return q.put(q.syncer, items...)
}
func (q *WrappedQueue) put(syncer persist.ItemSyncer, items ...MarshallableQueueItem) error {
	for _, item := range items {
		if syncer != nil {
			syncer.Write(item)
		}
		err := q.q.Put(item)
		if err != nil {
			return err
		}
		q.addToInternalMap(item)
		if q.q.Len() != len(q.m) && q.consistent {
			log.Warningf("Queues out of sync after Put method: #priority queue: %d, #wrapped map: %d", q.q.Len(), len(q.m))
			q.consistent = false
		} else {
			counters.SimulatedHospital.PendingItem.With(prometheus.Labels{
				"item_type": q.itemType,
			}).Inc()
		}
	}
	return nil
}
func (q *WrappedQueue) addToInternalMap(i MarshallableQueueItem) error {
	s, err := i.ID()
	if err != nil {
		return errors.Wrap(err, "cannot get item ID")
	}
	logLocal := log.WithField("item_id", s)
	logLocal.Debugf("Adding item: %v", i)
	_, ok := q.m[s]
	if ok {
		logLocal.Warning("Key collision, elements can be lost")
	}
	q.m[s] = i
	return nil
}
func (q *WrappedQueue) Get() (*MarshallableQueueItem, error) {
	q.mutex.Lock()
	defer q.mutex.Unlock()
	i, err := q.q.Get(1)
	if err != nil {
		return nil, err
	}
	if len(i) != 1 {
		return nil, errors.Wrapf(err, "Failed to consume item in queue, expected 1, got %d", len(i))
	}
	item := i[0].(MarshallableQueueItem)
	if q.syncer != nil {
		q.syncer.Delete(item)
	}
	id, err := item.ID()
	if err != nil {
		return nil, errors.Wrap(err, "cannot get item ID")
	}
	_, ok := q.m[id]
	if !ok {
		log.WithField("item_id", id).Warning("Elements out of sync: asked to remove an item that wasn't present")
	} else {
		delete(q.m, id)
		counters.SimulatedHospital.PendingItem.With(prometheus.Labels{
			"item_type": q.itemType,
		}).Dec()
	}
	if q.q.Len() != len(q.m) && q.consistent {
		log.Warningf("Elements out of sync after GET method: #priority queue: %d, #wrapped map: %d", q.q.Len(), len(q.m))
		q.consistent = false
	}
	return &item, nil
}
func (q *WrappedQueue) Peek() queue.Item {
	q.mutex.Lock()
	defer q.mutex.Unlock()
	return q.q.Peek()
}
func (q *WrappedQueue) Empty() bool {
	q.mutex.Lock()
	defer q.mutex.Unlock()
	return q.q.Empty()
}