New Methods in the binary heap interface...

Add Len() method to the Binary Heaps interface with implementation.
Start consumers only for the user-defined set from the config.
Add Headers field to the proto

Signed-off-by: Valery Piashchynski <[email protected]>

1 files changed, 5 insertions, 1 deletions

diff --git a/pkg/priorityqueue/binary_heap.go b/pkg/priorityqueue/binary_heap.go
index 47fdf5e5..a8d80fc0 100644
--- a/pkg/priorityqueue/binary_heap.go
+++ b/pkg/priorityqueue/binary_heap.go
@@ -68,6 +68,10 @@ func (bh *BinHeap) fixDown(curr, end int) {
 	}
 }
 
+func (bh *BinHeap) Len() uint64 {
+	return atomic.LoadUint64(&bh.len)
+}
+
 func (bh *BinHeap) Insert(item Item) {
 	bh.cond.L.Lock()
 	bh.items = append(bh.items, item)
@@ -87,7 +91,7 @@ func (bh *BinHeap) GetMax() Item {
 	bh.cond.L.Lock()
 	defer bh.cond.L.Unlock()
 
-	if atomic.LoadUint64(&bh.len) == 0 {
+	for atomic.LoadUint64(&bh.len) == 0 {
 		bh.cond.Wait()
 	}