1 files changed, 113 insertions, 0 deletions
diff --git a/pkg/priorityqueue/binary_heap_test.go b/pkg/priorityqueue/binary_heap_test.go
new file mode 100644
index 00000000..149ec764
--- /dev/null
+++ b/pkg/priorityqueue/binary_heap_test.go
@@ -0,0 +1,113 @@
+package priorityqueue
+
+import (
+	"fmt"
+	"math/rand"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/spiral/roadrunner/v2/utils"
+	"github.com/stretchr/testify/require"
+)
+
+type Test int
+
+func (t Test) Ack() {
+}
+
+func (t Test) Nack() {
+}
+
+func (t Test) Body() []byte {
+	return nil
+}
+
+func (t Test) Context() []byte {
+	return nil
+}
+
+func (t Test) ID() *string {
+	return utils.AsStringPtr("none")
+}
+
+func (t Test) Priority() *uint64 {
+	return utils.AsUint64Ptr(uint64(t))
+}
+
+func TestBinHeap_Init(t *testing.T) {
+	a := []Item{Test(2), Test(23), Test(33), Test(44), Test(1), Test(2), Test(2), Test(2), Test(4), Test(6), Test(99)}
+
+	bh := NewBinHeap()
+
+	for i := 0; i < len(a); i++ {
+		bh.Insert(a[i])
+	}
+
+	expected := []Item{Test(1), Test(2), Test(2), Test(2), Test(2), Test(4), Test(6), Test(23), Test(33), Test(44), Test(99)}
+
+	res := make([]Item, 0, 12)
+
+	for i := 0; i < 11; i++ {
+		item := bh.GetMax()
+		res = append(res, item)
+	}
+
+	require.Equal(t, expected, res)
+}
+
+func TestNewPriorityQueue(t *testing.T) {
+	insertsPerSec := uint64(0)
+	getPerSec := uint64(0)
+	stopCh := make(chan struct{}, 1)
+	pq := NewBinHeap()
+
+	go func() {
+		tt := time.NewTicker(time.Second)
+
+		for {
+			select {
+			case <-tt.C:
+				fmt.Println(fmt.Sprintf("GetMax per second: %d", atomic.LoadUint64(&getPerSec)))
+				fmt.Println(fmt.Sprintf("Insert per second: %d", atomic.LoadUint64(&insertsPerSec)))
+				atomic.StoreUint64(&getPerSec, 0)
+				atomic.StoreUint64(&insertsPerSec, 0)
+			case <-stopCh:
+				tt.Stop()
+				return
+			}
+		}
+	}()
+
+	go func() {
+		for {
+			select {
+			case <-stopCh:
+				return
+			default:
+				it := pq.GetMax()
+				if it == nil {
+					continue
+				}
+				atomic.AddUint64(&getPerSec, 1)
+			}
+		}
+	}()
+
+	go func() {
+		for {
+			select {
+			case <-stopCh:
+				return
+			default:
+				pq.Insert(Test(rand.Int())) //nolint:gosec
+				atomic.AddUint64(&insertsPerSec, 1)
+			}
+		}
+	}()
+
+	time.Sleep(time.Second * 5)
+	stopCh <- struct{}{}
+	stopCh <- struct{}{}
+	stopCh <- struct{}{}
+}