1 files changed, 281 insertions, 0 deletions

diff --git a/fanin/fanin_test.go b/fanin/fanin_test.go
new file mode 100644
index 0000000..bf2a413
--- /dev/null
+++ b/fanin/fanin_test.go
@@ -0,0 +1,281 @@
+package fanin
+
+import (
+        "context"
+        "sync"
+        "testing"
+        "time"
+)
+
+type mockObserver struct {
+        name string
+        t    *testing.T
+}
+
+func (o *mockObserver) OnInputAdded() {
+        o.t.Logf("%s: OnInputAdded", o.name)
+}
+
+func (o *mockObserver) OnInputClosed() {
+        o.t.Logf("%s: OnInputClosed", o.name)
+}
+
+func TestNewFanIn(t *testing.T) {
+        t.Parallel()
+
+        obs := new(mockObserver)
+
+        tests := []struct {
+                name      string
+                cfg       *FanInConfig
+                shouldErr bool
+        }{
+                {
+                        name: "valid config",
+                        cfg: &FanInConfig{
+                                BufferSize: 1,
+                                Observer:   obs,
+                        },
+                        shouldErr: false,
+                },
+                {
+                        name: "invalid buffer size",
+                        cfg: &FanInConfig{
+                                BufferSize: -1,
+                                Observer:   obs,
+                        },
+                        shouldErr: true,
+                },
+        }
+
+        for _, test := range tests {
+                t.Run(test.name, func(t *testing.T) {
+                        t.Parallel()
+                        _, err := NewFanIn[int](WithBufferSize(test.cfg.BufferSize), WithObserver(obs))
+
+                        if (err != nil) != test.shouldErr {
+                                t.Errorf("validation error = %v, want err %v", err, test.shouldErr)
+                        }
+                })
+        }
+}
+
+func TestFanIn_Add(t *testing.T) {
+        t.Parallel()
+        fi, _ := NewFanIn[int]()
+
+        inputs := make([]chan int, 10)
+        for i := range inputs {
+                inputs[i] = make(chan int, 1)
+                inputs[i] <- i
+                close(inputs[i])
+        }
+
+        for _, ch := range inputs {
+                fi.Add(ch)
+        }
+
+        count := 0
+        for range fi.Run(t.Context()) {
+                count++
+        }
+
+        if count != 10 {
+                t.Fatalf("expected 10 items, got %d", count)
+        }
+}
+
+func TestFanIn_Run(t *testing.T) {
+        t.Parallel()
+        fi, _ := NewFanIn[int]()
+
+        ch := make(chan int)
+        go func() {
+                ch <- 1
+                ch <- 2
+                ch <- 3
+                close(ch)
+        }()
+
+        fi.Add(ch)
+
+        for v := range fi.Run(t.Context()) {
+                t.Logf("value received %d", v)
+        }
+
+        <-fi.Done()
+}
+
+func TestFanIn_ContextCancellation(t *testing.T) {
+        t.Parallel()
+        fi, _ := NewFanIn[int]()
+        ctx, cancel := context.WithTimeout(t.Context(), time.Millisecond)
+        defer cancel()
+
+        inputs := make([]chan int, 10)
+        for i := range inputs {
+                inputs[i] = make(chan int, 1)
+                inputs[i] <- i
+                close(inputs[i])
+        }
+
+        for _, ch := range inputs {
+                fi.Add(ch)
+        }
+
+        select {
+        case <-fi.Run(ctx):
+                cancel()
+        case <-fi.Done():
+                cancel()
+        }
+}
+
+func TestFanIn_ShouldPanicRunMultiple(t *testing.T) {
+        t.Parallel()
+
+        defer func() {
+                if r := recover(); r == nil {
+                        t.Errorf("expected panic for multiple Run calls")
+                }
+        }()
+
+        ctx := t.Context()
+        fi, _ := NewFanIn[int]()
+
+        inputs := make([]chan int, 10)
+        for i := range inputs {
+                inputs[i] = make(chan int, 1)
+                inputs[i] <- i
+                close(inputs[i])
+        }
+
+        for _, ch := range inputs {
+                fi.Add(ch)
+        }
+
+        for v := range fi.Run(ctx) {
+                t.Logf("value received %d", v)
+        }
+
+        fi.Run(ctx)
+}
+
+func TestFanIn_ShouldPanicAddAfterRun(t *testing.T) {
+        t.Parallel()
+
+        defer func() {
+                if r := recover(); r == nil {
+                        t.Errorf("expected panic for Add after Run")
+                }
+        }()
+
+        fi, _ := NewFanIn[int]()
+
+        inputs := make([]chan int, 10)
+        for i := range inputs {
+                inputs[i] = make(chan int, 1)
+                inputs[i] <- i
+                close(inputs[i])
+        }
+
+        for _, ch := range inputs {
+                fi.Add(ch)
+        }
+
+        for v := range fi.Run(t.Context()) {
+                t.Logf("value received %d", v)
+        }
+
+        fi.Add(make(chan int))
+}
+
+func TestFanIn_Observer(t *testing.T) {
+        t.Parallel()
+        obs := &mockObserver{t: t}
+
+        fi, _ := NewFanIn[int](WithObserver(obs))
+
+        inputs := make([]chan int, 10)
+        for i := range inputs {
+                inputs[i] = make(chan int, 1)
+                inputs[i] <- i
+                close(inputs[i])
+        }
+
+        for _, ch := range inputs {
+                fi.Add(ch)
+        }
+
+        for v := range fi.Run(t.Context()) {
+                t.Logf("value received %d", v)
+        }
+
+        <-fi.Done()
+}
+
+func TestFanIn_AddConcurrency(t *testing.T) {
+        t.Parallel()
+
+        fi, _ := NewFanIn[int]()
+        var wg sync.WaitGroup
+        numGoroutines := 10
+        channelPerGoroutine := 10
+
+        for range numGoroutines {
+                wg.Go(func() {
+                        inputs := make([]chan int, channelPerGoroutine)
+                        for i := range inputs {
+                                inputs[i] = make(chan int, 1)
+                                inputs[i] <- i
+                                close(inputs[i])
+                                fi.Add(inputs[i])
+                        }
+                })
+        }
+        wg.Wait()
+
+        count := 0
+        for range fi.Run(t.Context()) {
+                count++
+        }
+
+        <-fi.Done()
+
+        expectedCount := numGoroutines * channelPerGoroutine
+        if count != expectedCount {
+                t.Errorf("expected %d items, got %d", expectedCount, count)
+        }
+}
+
+func TestFanIn_NilChannel(t *testing.T) {
+        t.Parallel()
+
+        fi, _ := NewFanIn[int]()
+
+        defer func() {
+                if r := recover(); r == nil {
+                        t.Error("expected panic when adding nil channel")
+                }
+        }()
+
+        fi.Add(nil)
+}
+
+func TestFanIn_RunWithNoInputs(t *testing.T) {
+        t.Parallel()
+
+        fi, _ := NewFanIn[int]()
+        out := fi.Run(t.Context())
+
+        count := 0
+        for range out {
+                count++
+        }
+
+        if count != 0 {
+                t.Errorf("expected 0 values from empty fan-in, got %d", count)
+        }
+
+        <-fi.Done()
+}