Various performance and correctness improvements.

queue/queue.go

@@ -0,0 +1,125 @@
+package queue
+
+import (
+	"context"
+	"git.wisehodl.dev/jay/go-honeybee/types"
+)
+
+func RunQueue(
+	id string,
+	ctx context.Context,
+	in <-chan types.ReceivedMessage,
+	out chan<- types.ReceivedMessage,
+	maxQueueSize int,
+) {
+	var next types.ReceivedMessage
+	var queue messageQueue
+	if maxQueueSize > 0 {
+		queue = newBoundedRing(maxQueueSize)
+	} else {
+		queue = newUnboundedRing(64)
+	}
+
+	for {
+		var outOrNil chan<- types.ReceivedMessage
+
+		// enable out channel if queue is populated
+		if queue.len() > 0 {
+			outOrNil = out
+			next = queue.peek()
+		}
+
+		select {
+		case <-ctx.Done():
+			return
+		case msg := <-in:
+			// limit queue size if maximum is configured
+			if maxQueueSize > 0 && queue.len() >= maxQueueSize {
+				// drop oldest message
+				_ = queue.pop()
+			}
+			// add new message
+			queue.push(msg)
+		// send next message to out channel
+		case outOrNil <- next:
+			_ = queue.pop()
+		}
+	}
+}
+
+// Ring Buffer Queue
+
+type messageQueue interface {
+	push(types.ReceivedMessage)
+	pop() types.ReceivedMessage
+	peek() types.ReceivedMessage
+	len() int
+}
+
+type ring struct {
+	buf  []types.ReceivedMessage
+	head int
+	size int
+}
+
+func (r *ring) len() int { return r.size }
+
+func (r *ring) pop() types.ReceivedMessage {
+	m := r.buf[r.head]
+	var zero types.ReceivedMessage
+	r.buf[r.head] = zero // release reference for GC
+	r.head = (r.head + 1) % len(r.buf)
+	r.size--
+	return m
+}
+
+func (r *ring) peek() types.ReceivedMessage {
+	m := r.buf[r.head]
+	return m
+}
+
+// shared write at logical tail; caller guarantees space exists
+func (r *ring) writeTail(m types.ReceivedMessage) {
+	r.buf[(r.head+r.size)%len(r.buf)] = m
+	r.size++
+}
+
+// Bounded ring
+
+type boundedRing struct{ ring }
+
+func newBoundedRing(cap int) *boundedRing {
+	return &boundedRing{ring{buf: make([]types.ReceivedMessage, cap)}}
+}
+
+func (b *boundedRing) push(m types.ReceivedMessage) {
+	if b.size == len(b.buf) {
+		b.buf[b.head] = m
+		b.head = (b.head + 1) % len(b.buf)
+		return
+	}
+	b.writeTail(m)
+}
+
+// Unbounded Ring
+
+type unboundedRing struct{ ring }
+
+func newUnboundedRing(initialCap int) *unboundedRing {
+	if initialCap < 1 {
+		initialCap = 1
+	}
+	return &unboundedRing{ring{buf: make([]types.ReceivedMessage, initialCap)}}
+}
+
+func (u *unboundedRing) push(m types.ReceivedMessage) {
+	if u.size == len(u.buf) {
+		bigger := make([]types.ReceivedMessage, len(u.buf)*2)
+		for i := 0; i < u.size; i++ {
+			bigger[i] = u.buf[(u.head+i)%len(u.buf)]
+		}
+		u.buf = bigger
+		u.head = 0
+	}
+	u.writeTail(m)
+}

queue/queue_test.go

@@ -0,0 +1,104 @@
+package queue
+
+import (
+	"context"
+	"git.wisehodl.dev/jay/go-honeybee/honeybeetest"
+	"git.wisehodl.dev/jay/go-honeybee/types"
+	"github.com/stretchr/testify/assert"
+	"testing"
+	"time"
+)
+
+func TestRunQueue(t *testing.T) {
+	t.Run("message passes through to inbox", func(t *testing.T) {
+		id := "wss://test"
+		inChan := make(chan types.ReceivedMessage, 1)
+		outChan := make(chan types.ReceivedMessage, 1)
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		go RunQueue(id, ctx, inChan, outChan, 0)
+
+		inChan <- types.ReceivedMessage{Data: []byte("hello"), ReceivedAt: time.Now()}
+
+		honeybeetest.Eventually(t, func() bool {
+			select {
+			case msg := <-outChan:
+				return string(msg.Data) == "hello"
+			default:
+				return false
+			}
+		}, "expected message")
+	})
+
+	t.Run("oldest message dropped when queue is full", func(t *testing.T) {
+		id := "wss://test"
+		inChan := make(chan types.ReceivedMessage, 1)
+		outChan := make(chan types.ReceivedMessage, 1)
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		gate := make(chan struct{})
+		gatedInbox := make(chan types.ReceivedMessage)
+
+		// gate the inbox from receiving messages until the gate is opened
+		go func() {
+			<-gate
+			for msg := range gatedInbox {
+				outChan <- msg
+			}
+		}()
+
+		go RunQueue(id, ctx, inChan, gatedInbox, 2)
+
+		// send three messages while the gated inbox is blocked
+		inChan <- types.ReceivedMessage{Data: []byte("first"), ReceivedAt: time.Now()}
+		inChan <- types.ReceivedMessage{Data: []byte("second"), ReceivedAt: time.Now()}
+		inChan <- types.ReceivedMessage{Data: []byte("third"), ReceivedAt: time.Now()}
+
+		// allow time for the first message to be dropped
+		time.Sleep(20 * time.Millisecond)
+
+		// close the gate, draining messages into the inbox
+		close(gate)
+
+		// receive messages from the inbox
+		var received []string
+		honeybeetest.Eventually(t, func() bool {
+			select {
+			case msg := <-outChan:
+				received = append(received, string(msg.Data))
+			default:
+			}
+			return len(received) == 2
+		}, "expected messages")
+
+		// first message was dropped
+		assert.Equal(t, []string{"second", "third"}, received)
+
+	})
+
+	t.Run("exits on context cancellation", func(t *testing.T) {
+		id := "wss://test"
+		inChan := make(chan types.ReceivedMessage, 1)
+		outChan := make(chan types.ReceivedMessage, 1)
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		done := make(chan struct{})
+		go func() {
+			RunQueue(id, ctx, inChan, outChan, 0)
+			close(done)
+		}()
+
+		cancel()
+		honeybeetest.Eventually(t, func() bool {
+			select {
+			case <-done:
+				return true
+			default:
+				return false
+			}
+		}, "expected done signal")
+	})
+}