1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
|
package buffered_queue
import (
"sync"
"testing"
)
func TestJobQueue(t *testing.T) {
type Job[T any] struct {
ID int
Action string
Data T
}
queue := NewBufferedQueue[Job[string]](2) // Chunk size of 5
queue.Enqueue(Job[string]{ID: 1, Action: "task1", Data: "hello"})
queue.Enqueue(Job[string]{ID: 2, Action: "task2", Data: "world"})
if queue.Size() != 2 {
t.Errorf("Expected queue size of 2, got %d", queue.Size())
}
queue.Enqueue(Job[string]{ID: 3, Action: "task3", Data: "3!"})
queue.Enqueue(Job[string]{ID: 4, Action: "task4", Data: "4!"})
queue.Enqueue(Job[string]{ID: 5, Action: "task5", Data: "5!"})
if queue.Size() != 5 {
t.Errorf("Expected queue size of 5, got %d", queue.Size())
}
println("enqueued 5 items")
println("dequeue", 1)
job, ok := queue.Dequeue()
if !ok {
t.Errorf("Expected dequeue to return true")
}
if job.ID != 1 {
t.Errorf("Expected job ID of 1, got %d", job.ID)
}
println("dequeue", 2)
job, ok = queue.Dequeue()
if !ok {
t.Errorf("Expected dequeue to return true")
}
println("enqueue", 6)
queue.Enqueue(Job[string]{ID: 6, Action: "task6", Data: "6!"})
println("enqueue", 7)
queue.Enqueue(Job[string]{ID: 7, Action: "task7", Data: "7!"})
for i := 0; i < 5; i++ {
println("dequeue ...")
job, ok = queue.Dequeue()
if !ok {
t.Errorf("Expected dequeue to return true")
}
println("dequeued", job.ID)
}
if queue.Size() != 0 {
t.Errorf("Expected queue size of 0, got %d", queue.Size())
}
for i := 0; i < 5; i++ {
println("enqueue", i+8)
queue.Enqueue(Job[string]{ID: i + 8, Action: "task", Data: "data"})
}
for i := 0; i < 5; i++ {
job, ok = queue.Dequeue()
if !ok {
t.Errorf("Expected dequeue to return true")
}
if job.ID != i+8 {
t.Errorf("Expected job ID of %d, got %d", i, job.ID)
}
println("dequeued", job.ID)
}
}
func TestJobQueueClose(t *testing.T) {
type Job[T any] struct {
ID int
Action string
Data T
}
queue := NewBufferedQueue[Job[string]](2)
queue.Enqueue(Job[string]{ID: 1, Action: "task1", Data: "hello"})
queue.Enqueue(Job[string]{ID: 2, Action: "task2", Data: "world"})
wg := sync.WaitGroup{}
wg.Add(1)
go func() {
defer wg.Done()
for data, ok := queue.Dequeue(); ok; data, ok = queue.Dequeue() {
println("dequeued", data.ID)
}
}()
for i := 0; i < 5; i++ {
queue.Enqueue(Job[string]{ID: i + 3, Action: "task", Data: "data"})
}
queue.CloseInput()
wg.Wait()
}
func BenchmarkBufferedQueue(b *testing.B) {
type Job[T any] struct {
ID int
Action string
Data T
}
queue := NewBufferedQueue[Job[string]](1024)
for i := 0; i < b.N; i++ {
queue.Enqueue(Job[string]{ID: i, Action: "task", Data: "data"})
}
for i := 0; i < b.N; i++ {
_, _ = queue.Dequeue()
}
}
|
