aboutsummaryrefslogtreecommitdiff
path: root/fanout/fanout_test.go
diff options
context:
space:
mode:
Diffstat (limited to 'fanout/fanout_test.go')
-rw-r--r--fanout/fanout_test.go412
1 files changed, 412 insertions, 0 deletions
diff --git a/fanout/fanout_test.go b/fanout/fanout_test.go
new file mode 100644
index 0000000..3f15217
--- /dev/null
+++ b/fanout/fanout_test.go
@@ -0,0 +1,412 @@
1package fanout
2
3import (
4 "context"
5 "sync"
6 "testing"
7 "time"
8)
9
10type mockObserver struct {
11 name string
12 t *testing.T
13}
14
15func (m *mockObserver) OnDistribute() {
16 m.t.Logf("fanout: %s OnDistribute", m.name)
17}
18
19func (m *mockObserver) OnOutputClosed() {
20 m.t.Logf("fanout: %s OnOutputClosed", m.name)
21}
22
23func TestFanOut_NewFanOut(t *testing.T) {
24 t.Parallel()
25
26 tests := []struct {
27 name string
28 opts []Option
29 shouldErr bool
30 }{
31 {
32 name: "worker size is invalid",
33 opts: []Option{WithWorkerCount(-10)},
34 shouldErr: true,
35 },
36 {
37 name: "buffer size is invalid",
38 opts: []Option{WithBufferSize(-10), WithWorkerCount(1)},
39 shouldErr: true,
40 },
41 {
42 name: "valid options",
43 opts: []Option{WithBufferSize(1), WithWorkerCount(1)},
44 shouldErr: false,
45 },
46 {
47 name: "worker count is zero",
48 opts: []Option{WithWorkerCount(0)},
49 shouldErr: true,
50 },
51 {
52 name: "valid broadcast strategy",
53 opts: []Option{WithWorkerCount(5), WithBufferSize(10), WithStrategy(Broadcast)},
54 shouldErr: false,
55 },
56 {
57 name: "invalid strategy",
58 opts: []Option{WithWorkerCount(1), WithStrategy(FanOutStrategy(99))},
59 shouldErr: true,
60 },
61 }
62
63 for _, test := range tests {
64 t.Run(test.name, func(t *testing.T) {
65 t.Parallel()
66 _, err := NewFanOut[int](test.opts...)
67 if (err != nil) != test.shouldErr {
68 t.Errorf("validation error = %v, want err %v", err, test.shouldErr)
69 }
70 })
71 }
72}
73
74func TestFanOut_ConcurrentStress(t *testing.T) {
75 t.Parallel()
76 ctx, cancel := context.WithCancel(t.Context())
77 defer cancel()
78
79 workerCount := 5
80 fo, err := NewFanOut[int](
81 WithWorkerCount(workerCount),
82 WithBufferSize(100),
83 )
84 if err != nil {
85 t.Fatalf("new fanout error: %v", err)
86 }
87
88 input := make(chan int, 1000)
89 fo.Run(ctx, input)
90
91 var wg sync.WaitGroup
92
93 for i := 0; i < 10; i++ {
94 wg.Go(func() {
95 outs := fo.Outputs()
96 if len(outs) != workerCount {
97 t.Errorf("expected %d outputs, got %d", workerCount, len(outs))
98 }
99 })
100 }
101
102 for _, ch := range fo.Outputs() {
103 wg.Go(func() {
104 for range ch {
105 }
106 })
107 }
108
109 producers := 10
110 messagesPerProducer := 100
111
112 var producerWg sync.WaitGroup
113 producerWg.Add(producers)
114
115 for i := 0; i < producers; i++ {
116 go func() {
117 defer producerWg.Done()
118 for j := 0; j < messagesPerProducer; j++ {
119 select {
120 case input <- j:
121 case <-ctx.Done():
122 return
123 }
124 }
125 }()
126 }
127
128 go func() {
129 producerWg.Wait()
130 close(input)
131 }()
132
133 wg.Wait()
134}
135
136func TestFanOut_RunBroadcast(t *testing.T) {
137 t.Parallel()
138 var wg sync.WaitGroup
139 workerCount := 2
140 size := 10
141 obs := &mockObserver{
142 name: "fo broadcast",
143 t: t,
144 }
145 fo, _ := NewFanOut[int](
146 WithWorkerCount(workerCount),
147 WithStrategy(Broadcast),
148 WithObserver(obs),
149 )
150
151 ch := make(chan int, size)
152 wg.Go(func() {
153 for i := range size {
154 ch <- i
155 }
156 close(ch)
157 })
158
159 fo.Run(t.Context(), ch)
160
161 res := make([]int, workerCount)
162 for i, cha := range fo.Outputs() {
163 wg.Go(func() {
164 count := 0
165 for v := range cha {
166 t.Logf("%v", v)
167 count++
168 res[i] = count
169 }
170 })
171 }
172
173 wg.Wait()
174
175 expected := size
176 for i, count := range res {
177 if count != expected {
178 t.Errorf("worker %d received %d items, expected %d", i, count, expected)
179 }
180 }
181}
182
183func TestFanOut_RunRoundRobin(t *testing.T) {
184 t.Parallel()
185 var wg sync.WaitGroup
186 size := 10
187 workerCount := 2
188 obs := &mockObserver{
189 name: "fo round robin",
190 t: t,
191 }
192 fo, _ := NewFanOut[int](
193 WithWorkerCount(workerCount),
194 WithStrategy(RoundRobin),
195 WithObserver(obs),
196 )
197
198 ch := make(chan int, size)
199 wg.Go(func() {
200 for i := range size {
201 ch <- i
202 }
203 close(ch)
204 })
205
206 fo.Run(t.Context(), ch)
207 res := make([]int, workerCount)
208 for i, cha := range fo.Outputs() {
209 wg.Go(func() {
210 count := 0
211 for v := range cha {
212 t.Logf("%v", v)
213 count++
214 res[i] = count
215 }
216 })
217 }
218
219 wg.Wait()
220 expected := size / workerCount
221
222 for i, count := range res {
223 if count != expected {
224 t.Errorf("worker %d received %d items, expected %d", i, count, expected)
225 }
226 }
227}
228
229func TestFanOut_ContextTimeOut(t *testing.T) {
230 t.Parallel()
231 fo, _ := NewFanOut[int](WithWorkerCount(2))
232 ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
233 defer cancel()
234
235 input := make(chan int, 10)
236 fo.Run(ctx, input)
237
238 var wg sync.WaitGroup
239 done := make(chan struct{})
240 for _, ch := range fo.Outputs() {
241 wg.Go(func() {
242 for v := range ch {
243 t.Logf("received %d", v)
244 }
245 })
246 }
247
248 go func() {
249 wg.Wait()
250 close(done)
251 }()
252
253 go func() {
254 i := 0
255 for {
256 select {
257 case input <- i:
258 i++
259 time.Sleep(time.Millisecond)
260 case <-ctx.Done():
261 return
262 }
263 }
264 }()
265
266 select {
267 case <-done:
268 case <-time.After(200 * time.Millisecond):
269 t.Fatal("timeout waiting for cleanup")
270 }
271}
272
273func TestFanOut_ContextCancel(t *testing.T) {
274 t.Parallel()
275 ctx, cancel := context.WithCancel(t.Context())
276
277 fo, err := NewFanOut[int](WithWorkerCount(3))
278 if err != nil {
279 t.Fatalf("failed to create fanout: %v", err)
280 }
281
282 input := make(chan int, 10)
283 fo.Run(ctx, input)
284
285 var wg sync.WaitGroup
286 for _, ch := range fo.Outputs() {
287 wg.Go(func() {
288 for range ch {
289 }
290 })
291 }
292
293 cancel()
294 close(input)
295
296 done := make(chan struct{})
297 go func() {
298 wg.Wait()
299 close(done)
300 }()
301
302 select {
303 case <-done:
304 case <-time.After(time.Second):
305 t.Fatal("timeout waiting for shutdown")
306 }
307}
308
309func TestFanOut_ContextDoneRoundRobin(t *testing.T) {
310 t.Parallel()
311 ctx, cancel := context.WithCancel(t.Context())
312
313 fo, err := NewFanOut[int](
314 WithWorkerCount(3),
315 WithBufferSize(0),
316 WithStrategy(RoundRobin),
317 )
318 if err != nil {
319 t.Fatalf("new fanout error: %v", err)
320 }
321
322 input := make(chan int, 1)
323 input <- 1
324 fo.Run(ctx, input)
325
326 cancel()
327
328 for i, ch := range fo.Outputs() {
329 select {
330 case _, ok := <-ch:
331 if ok {
332 for range ch {
333 }
334 }
335 case <-time.After(time.Second):
336 t.Fatalf("worker %d: timeout waiting for close", i)
337 }
338 }
339}
340
341func TestFanOut_ContextDoneBroadcast(t *testing.T) {
342 t.Parallel()
343 ctx, cancel := context.WithCancel(t.Context())
344
345 fo, err := NewFanOut[int](
346 WithWorkerCount(3),
347 WithStrategy(Broadcast),
348 WithBufferSize(0),
349 )
350 if err != nil {
351 t.Fatalf("new fanout error: %v", err)
352 }
353
354 input := make(chan int, 1)
355 input <- 1
356 fo.Run(ctx, input)
357
358 cancel()
359
360 for i, ch := range fo.Outputs() {
361 select {
362 case _, ok := <-ch:
363 if ok {
364 for range ch {
365 }
366 }
367 case <-time.After(time.Second):
368 t.Fatalf("worker %d: timeout waiting for close", i)
369 }
370 }
371}
372
373func TestFanOut_RunMultipleTime(t *testing.T) {
374 t.Parallel()
375
376 ctx := t.Context()
377 fo, err := NewFanOut[int](
378 WithWorkerCount(3),
379 WithBufferSize(0),
380 )
381 if err != nil {
382 t.Fatalf("new fanout error: %v", err)
383 }
384
385 input := make(chan int, 10)
386 fo.Run(ctx, input)
387
388 defer func() {
389 if r := recover(); r == nil {
390 t.Error("expected panic when calling Run() multiple times")
391 }
392 }()
393
394 fo.Run(ctx, input)
395}
396
397func TestFanOut_NilInput(t *testing.T) {
398 t.Parallel()
399
400 fo, err := NewFanOut[int](WithWorkerCount(1))
401 if err != nil {
402 t.Fatalf("failed to create fanout: %v", err)
403 }
404
405 defer func() {
406 if r := recover(); r == nil {
407 t.Error("expected panic when passing nil input channel")
408 }
409 }()
410
411 fo.Run(t.Context(), nil)
412}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-17 19:33:18 +0000