Add support for duplicated items

* Implement 'allowDuplicates' flag, while maintaining backwards compatibility.
* Add unit test.

Author: ichinaski

queue/priority_queue.go

@@ -105,12 +105,13 @@ func (items *priorityItems) push(item Item) {
 // items that implement the Item interface and adds them
 // to the queue in priority order.
 type PriorityQueue struct {
-	waiters     waiters
-	items       priorityItems
-	itemMap     map[Item]struct{}
-	lock        sync.Mutex
-	disposeLock sync.Mutex
-	disposed    bool
+	waiters         waiters
+	items           priorityItems
+	itemMap         map[Item]struct{}
+	lock            sync.Mutex
+	disposeLock     sync.Mutex
+	disposed        bool
+	allowDuplicates bool
 }
 
 // Put adds items to the queue.
@@ -126,7 +127,9 @@ func (pq *PriorityQueue) Put(items ...Item) error {
 	}
 
 	for _, item := range items {
-		if _, ok := pq.itemMap[item]; !ok {
+		if pq.allowDuplicates {
+			pq.items.push(item)
+		} else if _, ok := pq.itemMap[item]; !ok {
 			pq.itemMap[item] = struct{}{}
 			pq.items.push(item)
 		}
@@ -188,7 +191,9 @@ func (pq *PriorityQueue) Get(number int) ([]Item, error) {
 		pq.disposeLock.Unlock()
 
 		items = pq.items.get(number)
-		deleteItems(items)
+		if !pq.allowDuplicates {
+			deleteItems(items)
+		}
 		sema.response.Done()
 		return items, nil
 	}
@@ -253,6 +258,12 @@ func (pq *PriorityQueue) Dispose() {
 	pq.waiters = nil
 }
 
+// AllowDuplicates determines whether the queue supports
+// duplicated items. This setting is false by default.
+func (pq *PriorityQueue) AllowDuplicates(allow bool) {
+	pq.allowDuplicates = allow
+}
+
 // NewPriorityQueue is the constructor for a priority queue.
 func NewPriorityQueue(hint int) *PriorityQueue {
 	return &PriorityQueue{

queue/priority_queue_test.go

@@ -245,3 +245,12 @@ func TestInsertDuplicate(t *testing.T) {
 
 	assert.Equal(t, 1, q.Len())
 }
+
+func TestAllowDuplicates(t *testing.T) {
+	q := NewPriorityQueue(2)
+	q.AllowDuplicates(true)
+	q.Put(mockItem(1))
+	q.Put(mockItem(1))
+
+	assert.Equal(t, 2, q.Len())
+}