mirror of
https://github.com/binwiederhier/ntfy.git
synced 2026-09-15 06:52:46 -04:00
91 lines
2.8 KiB
Go
91 lines
2.8 KiB
Go
package util_test
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
"heckel.io/ntfy/v2/util"
|
|
)
|
|
|
|
func TestLingerQueue_LingerFlush(t *testing.T) {
|
|
// Items enqueued within the linger window are emitted as a single batch when it expires
|
|
q := util.NewLingerQueue[int](16, 100, 0, nil, 50*time.Millisecond)
|
|
defer q.Close()
|
|
require.True(t, q.TryEnqueue(1))
|
|
require.True(t, q.TryEnqueue(2))
|
|
require.True(t, q.TryEnqueue(3))
|
|
start := time.Now()
|
|
batch := <-q.Dequeue()
|
|
require.Equal(t, []int{1, 2, 3}, batch)
|
|
require.GreaterOrEqual(t, time.Since(start), 30*time.Millisecond) // Waited out the linger
|
|
}
|
|
|
|
func TestLingerQueue_MaxBatchFlush(t *testing.T) {
|
|
// Hitting the count cap flushes early, before the linger expires
|
|
q := util.NewLingerQueue[int](16, 5, 0, nil, time.Minute)
|
|
defer q.Close()
|
|
for i := 0; i < 12; i++ {
|
|
require.True(t, q.TryEnqueue(i))
|
|
}
|
|
require.Len(t, <-q.Dequeue(), 5)
|
|
require.Len(t, <-q.Dequeue(), 5)
|
|
q.Close() // Flushes the remainder
|
|
require.Len(t, <-q.Dequeue(), 2)
|
|
}
|
|
|
|
func TestLingerQueue_SizeCapFlush(t *testing.T) {
|
|
// Hitting the byte cap flushes early, before count cap or linger
|
|
q := util.NewLingerQueue(16, 100, 10, func(s string) int { return len(s) }, time.Minute)
|
|
defer q.Close()
|
|
require.True(t, q.TryEnqueue("aaaa"))
|
|
require.True(t, q.TryEnqueue("bbbb"))
|
|
require.True(t, q.TryEnqueue("cccc")) // 12 bytes >= 10 -> flush
|
|
batch := <-q.Dequeue()
|
|
require.Equal(t, []string{"aaaa", "bbbb", "cccc"}, batch)
|
|
}
|
|
|
|
func TestLingerQueue_TryEnqueueFull(t *testing.T) {
|
|
// A full queue drops (returns false) instead of blocking the producer
|
|
q := util.NewLingerQueue[int](1, 1, 0, nil, 0)
|
|
defer q.Close()
|
|
require.True(t, q.TryEnqueue(1)) // Taken by the batcher, blocks emitting (no consumer)
|
|
waitForCond(t, func() bool { return q.TryEnqueue(2) }) // Fills the buffer once slot frees
|
|
require.False(t, q.TryEnqueue(3)) // Buffer full, batcher blocked -> drop
|
|
}
|
|
|
|
func TestLingerQueue_CloseFlushesAndCloses(t *testing.T) {
|
|
q := util.NewLingerQueue[int](16, 100, 0, nil, time.Minute)
|
|
require.True(t, q.TryEnqueue(1))
|
|
require.True(t, q.TryEnqueue(2))
|
|
q.Close()
|
|
require.Equal(t, []int{1, 2}, <-q.Dequeue()) // Remainder flushed without waiting out the linger
|
|
_, ok := <-q.Dequeue()
|
|
require.False(t, ok) // Channel closed
|
|
require.False(t, q.TryEnqueue(3))
|
|
q.Close() // Idempotent
|
|
}
|
|
|
|
func TestLingerQueue_LingerZeroImmediate(t *testing.T) {
|
|
q := util.NewLingerQueue[int](16, 100, 0, nil, 0)
|
|
defer q.Close()
|
|
require.True(t, q.TryEnqueue(1))
|
|
select {
|
|
case batch := <-q.Dequeue():
|
|
require.Contains(t, batch, 1)
|
|
case <-time.After(time.Second):
|
|
t.Fatal("expected immediate flush with zero linger")
|
|
}
|
|
}
|
|
|
|
func waitForCond(t *testing.T, f func() bool) {
|
|
t.Helper()
|
|
for i := 0; i < 100; i++ {
|
|
if f() {
|
|
return
|
|
}
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
t.Fatal("timed out waiting for condition")
|
|
}
|