4 files changed, 725 insertions, 0 deletions

diff --git a/fanout/example_test.go b/fanout/example_test.go
new file mode 100644
index 0000000..50e7e96
--- /dev/null
+++ b/fanout/example_test.go
@@ -0,0 +1,45 @@
+package fanout_test
+
+import (
+        "context"
+        "fmt"
+        "sort"
+        "sync"
+
+        "git.sangtran.io.vn/quacker.git/fanout"
+)
+
+func ExampleNewFanOut() {
+        fo, _ := fanout.NewFanOut[int](
+                fanout.WithWorkerCount(2),
+                fanout.WithBufferSize(10),
+                fanout.WithStrategy(fanout.RoundRobin),
+        )
+
+        input := make(chan int, 4)
+        for i := 1; i <= 4; i++ {
+                input <- i
+        }
+        close(input)
+
+        fo.Run(context.Background(), input)
+
+        var mu sync.Mutex
+        var all []int
+        var wg sync.WaitGroup
+
+        for _, ch := range fo.Outputs() {
+                wg.Go(func() {
+                        for v := range ch {
+                                mu.Lock()
+                                all = append(all, v)
+                                mu.Unlock()
+                        }
+                })
+        }
+
+        wg.Wait()
+        sort.Ints(all)
+        fmt.Println(all)
+        // Output: [1 2 3 4]
+}
diff --git a/fanout/fanout.go b/fanout/fanout.go
new file mode 100644
index 0000000..1cd2043
--- /dev/null
+++ b/fanout/fanout.go
@@ -0,0 +1,203 @@
+// Package fanout distributes values from a single input channel to multiple output channels.
+package fanout
+
+import (
+        "context"
+        "errors"
+        "fmt"
+        "sync"
+        "sync/atomic"
+)
+
+// FanOutObserver defines callbacks for monitoring fan-out lifecycle events.
+type FanOutObserver interface {
+        OnDistribute()
+        OnOutputClosed()
+}
+
+// FanOutStrategy defines how values are distributed across output channels.
+type FanOutStrategy int
+
+const (
+        // RoundRobin distributes values evenly across output channels in rotation.
+        RoundRobin FanOutStrategy = iota
+        // Broadcast sends each value to all output channels.
+        Broadcast
+)
+
+// ErrInvalidStrategy is returned when an unknown FanOutStrategy is provided.
+var ErrInvalidStrategy = errors.New("fanout: unknown distribution strategy")
+
+// FanOutConfig holds configuration options for creating a FanOut instance.
+type FanOutConfig struct {
+        WorkerCount int
+        BufferSize  int
+        Strategy    FanOutStrategy
+        Observer    FanOutObserver
+}
+
+// Option is a functional option for configuring a FanOut instance.
+type Option func(*FanOutConfig)
+
+// FanOut distributes values from a single input channel to multiple output channels.
+type FanOut[T any] struct {
+        cfg           *FanOutConfig
+        outputs       []chan T
+        running       atomic.Bool
+        cachedOutputs []<-chan T
+        once          sync.Once
+}
+
+// NewFanOut creates a new FanOut instance with the provided options.
+func NewFanOut[T any](opts ...Option) (*FanOut[T], error) {
+        cfg := &FanOutConfig{
+                WorkerCount: 1,
+                BufferSize:  0,
+                Observer:    nil,
+                Strategy:    RoundRobin,
+        }
+
+        for _, opt := range opts {
+                opt(cfg)
+        }
+
+        if cfg.WorkerCount < 1 {
+                return nil, fmt.Errorf("fanout: worker count must be at least 1, got %d", cfg.WorkerCount)
+        }
+
+        if cfg.BufferSize < 0 {
+                return nil, fmt.Errorf("fanout: buffer size must not be negative, got %d", cfg.BufferSize)
+        }
+
+        if cfg.Strategy != RoundRobin && cfg.Strategy != Broadcast {
+                return nil, ErrInvalidStrategy
+        }
+
+        outputs := make([]chan T, cfg.WorkerCount)
+        for i := range outputs {
+                outputs[i] = make(chan T, cfg.BufferSize)
+        }
+
+        return &FanOut[T]{
+                cfg:     cfg,
+                outputs: outputs,
+        }, nil
+}
+
+// Run starts distributing values from the input channel to all output channels.
+func (f *FanOut[T]) Run(ctx context.Context, input <-chan T) {
+        if input == nil {
+                panic("fanout: input channel must not be nil")
+        }
+
+        if !f.running.CompareAndSwap(false, true) {
+                panic("fanout: Run() called multiple times")
+        }
+
+        go func() {
+                idx := 0
+                defer f.running.Store(false)
+                defer f.closeAllOutput()
+                for {
+                        select {
+                        case value, ok := <-input:
+                                if !ok {
+                                        return
+                                }
+                                switch f.cfg.Strategy {
+                                case RoundRobin:
+                                        f.roundRobin(ctx, value, &idx)
+                                case Broadcast:
+                                        f.broadcast(ctx, value)
+                                }
+                        case <-ctx.Done():
+                                return
+                        }
+                }
+        }()
+}
+
+// Outputs returns receive-only channel references for the distributed streams.
+func (f *FanOut[T]) Outputs() []<-chan T {
+        f.once.Do(func() {
+                f.cachedOutputs = make([]<-chan T, len(f.outputs))
+                for i, ch := range f.outputs {
+                        f.cachedOutputs[i] = ch
+                }
+        })
+
+        return f.cachedOutputs
+}
+
+func safeObserve(fn func()) {
+        defer func() { recover() }()
+        fn()
+}
+
+func (f *FanOut[T]) roundRobin(ctx context.Context, value T, index *int) {
+        target := *index % f.cfg.WorkerCount
+        select {
+        case f.outputs[target] <- value:
+                *index = (*index + 1) % f.cfg.WorkerCount
+                if f.cfg.Observer != nil {
+                        safeObserve(f.cfg.Observer.OnDistribute)
+                }
+        case <-ctx.Done():
+                return
+        }
+}
+
+func (f *FanOut[T]) broadcast(ctx context.Context, value T) {
+        var wg sync.WaitGroup
+        for _, ch := range f.outputs {
+                wg.Go(func() {
+                        select {
+                        case ch <- value:
+                                if f.cfg.Observer != nil {
+                                        safeObserve(f.cfg.Observer.OnDistribute)
+                                }
+                        case <-ctx.Done():
+                                return
+                        }
+                })
+        }
+        wg.Wait()
+}
+
+func (f *FanOut[T]) closeAllOutput() {
+        for _, ch := range f.outputs {
+                close(ch)
+        }
+
+        if f.cfg.Observer != nil {
+                safeObserve(f.cfg.Observer.OnOutputClosed)
+        }
+}
+
+// WithWorkerCount sets the number of output channels for the fan-out.
+func WithWorkerCount(n int) Option {
+        return func(cfg *FanOutConfig) {
+                cfg.WorkerCount = n
+        }
+}
+
+// WithBufferSize sets the buffer capacity for each output channel.
+func WithBufferSize(size int) Option {
+        return func(cfg *FanOutConfig) {
+                cfg.BufferSize = size
+        }
+}
+
+// WithStrategy sets the distribution strategy for the fan-out.
+func WithStrategy(strategy FanOutStrategy) Option {
+        return func(cfg *FanOutConfig) {
+                cfg.Strategy = strategy
+        }
+}
+
+// WithObserver attaches a FanOutObserver to receive lifecycle event callbacks.
+func WithObserver(observer FanOutObserver) Option {
+        return func(cfg *FanOutConfig) {
+                cfg.Observer = observer
+        }
+}
diff --git a/fanout/fanout_bench_test.go b/fanout/fanout_bench_test.go
new file mode 100644
index 0000000..0b88d87
--- /dev/null
+++ b/fanout/fanout_bench_test.go
@@ -0,0 +1,65 @@
+package fanout
+
+import (
+        "context"
+        "testing"
+)
+
+func BenchmarkFanOut_RoundRobin(b *testing.B) {
+        b.ReportAllocs()
+
+        fo, _ := NewFanOut[int](
+                WithWorkerCount(4),
+                WithBufferSize(64),
+                WithStrategy(RoundRobin),
+        )
+
+        ctx, cancel := context.WithCancel(context.Background())
+        defer cancel()
+
+        input := make(chan int, 64)
+        fo.Run(ctx, input)
+
+        // Drain outputs
+        for _, ch := range fo.Outputs() {
+                go func() {
+                        for range ch {
+                        }
+                }()
+        }
+
+        for b.Loop() {
+                input <- 1
+        }
+
+        close(input)
+}
+
+func BenchmarkFanOut_Broadcast(b *testing.B) {
+        b.ReportAllocs()
+
+        fo, _ := NewFanOut[int](
+                WithWorkerCount(4),
+                WithBufferSize(64),
+                WithStrategy(Broadcast),
+        )
+
+        ctx, cancel := context.WithCancel(context.Background())
+        defer cancel()
+
+        input := make(chan int, 64)
+        fo.Run(ctx, input)
+
+        for _, ch := range fo.Outputs() {
+                go func() {
+                        for range ch {
+                        }
+                }()
+        }
+
+        for b.Loop() {
+                input <- 1
+        }
+
+        close(input)
+}
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 @@
+package fanout
+
+import (
+        "context"
+        "sync"
+        "testing"
+        "time"
+)
+
+type mockObserver struct {
+        name string
+        t    *testing.T
+}
+
+func (m *mockObserver) OnDistribute() {
+        m.t.Logf("fanout: %s OnDistribute", m.name)
+}
+
+func (m *mockObserver) OnOutputClosed() {
+        m.t.Logf("fanout: %s OnOutputClosed", m.name)
+}
+
+func TestFanOut_NewFanOut(t *testing.T) {
+        t.Parallel()
+
+        tests := []struct {
+                name      string
+                opts      []Option
+                shouldErr bool
+        }{
+                {
+                        name:      "worker size is invalid",
+                        opts:      []Option{WithWorkerCount(-10)},
+                        shouldErr: true,
+                },
+                {
+                        name:      "buffer size is invalid",
+                        opts:      []Option{WithBufferSize(-10), WithWorkerCount(1)},
+                        shouldErr: true,
+                },
+                {
+                        name:      "valid options",
+                        opts:      []Option{WithBufferSize(1), WithWorkerCount(1)},
+                        shouldErr: false,
+                },
+                {
+                        name:      "worker count is zero",
+                        opts:      []Option{WithWorkerCount(0)},
+                        shouldErr: true,
+                },
+                {
+                        name:      "valid broadcast strategy",
+                        opts:      []Option{WithWorkerCount(5), WithBufferSize(10), WithStrategy(Broadcast)},
+                        shouldErr: false,
+                },
+                {
+                        name:      "invalid strategy",
+                        opts:      []Option{WithWorkerCount(1), WithStrategy(FanOutStrategy(99))},
+                        shouldErr: true,
+                },
+        }
+
+        for _, test := range tests {
+                t.Run(test.name, func(t *testing.T) {
+                        t.Parallel()
+                        _, err := NewFanOut[int](test.opts...)
+                        if (err != nil) != test.shouldErr {
+                                t.Errorf("validation error = %v, want err %v", err, test.shouldErr)
+                        }
+                })
+        }
+}
+
+func TestFanOut_ConcurrentStress(t *testing.T) {
+        t.Parallel()
+        ctx, cancel := context.WithCancel(t.Context())
+        defer cancel()
+
+        workerCount := 5
+        fo, err := NewFanOut[int](
+                WithWorkerCount(workerCount),
+                WithBufferSize(100),
+        )
+        if err != nil {
+                t.Fatalf("new fanout error: %v", err)
+        }
+
+        input := make(chan int, 1000)
+        fo.Run(ctx, input)
+
+        var wg sync.WaitGroup
+
+        for i := 0; i < 10; i++ {
+                wg.Go(func() {
+                        outs := fo.Outputs()
+                        if len(outs) != workerCount {
+                                t.Errorf("expected %d outputs, got %d", workerCount, len(outs))
+                        }
+                })
+        }
+
+        for _, ch := range fo.Outputs() {
+                wg.Go(func() {
+                        for range ch {
+                        }
+                })
+        }
+
+        producers := 10
+        messagesPerProducer := 100
+
+        var producerWg sync.WaitGroup
+        producerWg.Add(producers)
+
+        for i := 0; i < producers; i++ {
+                go func() {
+                        defer producerWg.Done()
+                        for j := 0; j < messagesPerProducer; j++ {
+                                select {
+                                case input <- j:
+                                case <-ctx.Done():
+                                        return
+                                }
+                        }
+                }()
+        }
+
+        go func() {
+                producerWg.Wait()
+                close(input)
+        }()
+
+        wg.Wait()
+}
+
+func TestFanOut_RunBroadcast(t *testing.T) {
+        t.Parallel()
+        var wg sync.WaitGroup
+        workerCount := 2
+        size := 10
+        obs := &mockObserver{
+                name: "fo broadcast",
+                t:    t,
+        }
+        fo, _ := NewFanOut[int](
+                WithWorkerCount(workerCount),
+                WithStrategy(Broadcast),
+                WithObserver(obs),
+        )
+
+        ch := make(chan int, size)
+        wg.Go(func() {
+                for i := range size {
+                        ch <- i
+                }
+                close(ch)
+        })
+
+        fo.Run(t.Context(), ch)
+
+        res := make([]int, workerCount)
+        for i, cha := range fo.Outputs() {
+                wg.Go(func() {
+                        count := 0
+                        for v := range cha {
+                                t.Logf("%v", v)
+                                count++
+                                res[i] = count
+                        }
+                })
+        }
+
+        wg.Wait()
+
+        expected := size
+        for i, count := range res {
+                if count != expected {
+                        t.Errorf("worker %d received %d items, expected %d", i, count, expected)
+                }
+        }
+}
+
+func TestFanOut_RunRoundRobin(t *testing.T) {
+        t.Parallel()
+        var wg sync.WaitGroup
+        size := 10
+        workerCount := 2
+        obs := &mockObserver{
+                name: "fo round robin",
+                t:    t,
+        }
+        fo, _ := NewFanOut[int](
+                WithWorkerCount(workerCount),
+                WithStrategy(RoundRobin),
+                WithObserver(obs),
+        )
+
+        ch := make(chan int, size)
+        wg.Go(func() {
+                for i := range size {
+                        ch <- i
+                }
+                close(ch)
+        })
+
+        fo.Run(t.Context(), ch)
+        res := make([]int, workerCount)
+        for i, cha := range fo.Outputs() {
+                wg.Go(func() {
+                        count := 0
+                        for v := range cha {
+                                t.Logf("%v", v)
+                                count++
+                                res[i] = count
+                        }
+                })
+        }
+
+        wg.Wait()
+        expected := size / workerCount
+
+        for i, count := range res {
+                if count != expected {
+                        t.Errorf("worker %d received %d items, expected %d", i, count, expected)
+                }
+        }
+}
+
+func TestFanOut_ContextTimeOut(t *testing.T) {
+        t.Parallel()
+        fo, _ := NewFanOut[int](WithWorkerCount(2))
+        ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
+        defer cancel()
+
+        input := make(chan int, 10)
+        fo.Run(ctx, input)
+
+        var wg sync.WaitGroup
+        done := make(chan struct{})
+        for _, ch := range fo.Outputs() {
+                wg.Go(func() {
+                        for v := range ch {
+                                t.Logf("received %d", v)
+                        }
+                })
+        }
+
+        go func() {
+                wg.Wait()
+                close(done)
+        }()
+
+        go func() {
+                i := 0
+                for {
+                        select {
+                        case input <- i:
+                                i++
+                                time.Sleep(time.Millisecond)
+                        case <-ctx.Done():
+                                return
+                        }
+                }
+        }()
+
+        select {
+        case <-done:
+        case <-time.After(200 * time.Millisecond):
+                t.Fatal("timeout waiting for cleanup")
+        }
+}
+
+func TestFanOut_ContextCancel(t *testing.T) {
+        t.Parallel()
+        ctx, cancel := context.WithCancel(t.Context())
+
+        fo, err := NewFanOut[int](WithWorkerCount(3))
+        if err != nil {
+                t.Fatalf("failed to create fanout: %v", err)
+        }
+
+        input := make(chan int, 10)
+        fo.Run(ctx, input)
+
+        var wg sync.WaitGroup
+        for _, ch := range fo.Outputs() {
+                wg.Go(func() {
+                        for range ch {
+                        }
+                })
+        }
+
+        cancel()
+        close(input)
+
+        done := make(chan struct{})
+        go func() {
+                wg.Wait()
+                close(done)
+        }()
+
+        select {
+        case <-done:
+        case <-time.After(time.Second):
+                t.Fatal("timeout waiting for shutdown")
+        }
+}
+
+func TestFanOut_ContextDoneRoundRobin(t *testing.T) {
+        t.Parallel()
+        ctx, cancel := context.WithCancel(t.Context())
+
+        fo, err := NewFanOut[int](
+                WithWorkerCount(3),
+                WithBufferSize(0),
+                WithStrategy(RoundRobin),
+        )
+        if err != nil {
+                t.Fatalf("new fanout error: %v", err)
+        }
+
+        input := make(chan int, 1)
+        input <- 1
+        fo.Run(ctx, input)
+
+        cancel()
+
+        for i, ch := range fo.Outputs() {
+                select {
+                case _, ok := <-ch:
+                        if ok {
+                                for range ch {
+                                }
+                        }
+                case <-time.After(time.Second):
+                        t.Fatalf("worker %d: timeout waiting for close", i)
+                }
+        }
+}
+
+func TestFanOut_ContextDoneBroadcast(t *testing.T) {
+        t.Parallel()
+        ctx, cancel := context.WithCancel(t.Context())
+
+        fo, err := NewFanOut[int](
+                WithWorkerCount(3),
+                WithStrategy(Broadcast),
+                WithBufferSize(0),
+        )
+        if err != nil {
+                t.Fatalf("new fanout error: %v", err)
+        }
+
+        input := make(chan int, 1)
+        input <- 1
+        fo.Run(ctx, input)
+
+        cancel()
+
+        for i, ch := range fo.Outputs() {
+                select {
+                case _, ok := <-ch:
+                        if ok {
+                                for range ch {
+                                }
+                        }
+                case <-time.After(time.Second):
+                        t.Fatalf("worker %d: timeout waiting for close", i)
+                }
+        }
+}
+
+func TestFanOut_RunMultipleTime(t *testing.T) {
+        t.Parallel()
+
+        ctx := t.Context()
+        fo, err := NewFanOut[int](
+                WithWorkerCount(3),
+                WithBufferSize(0),
+        )
+        if err != nil {
+                t.Fatalf("new fanout error: %v", err)
+        }
+
+        input := make(chan int, 10)
+        fo.Run(ctx, input)
+
+        defer func() {
+                if r := recover(); r == nil {
+                        t.Error("expected panic when calling Run() multiple times")
+                }
+        }()
+
+        fo.Run(ctx, input)
+}
+
+func TestFanOut_NilInput(t *testing.T) {
+        t.Parallel()
+
+        fo, err := NewFanOut[int](WithWorkerCount(1))
+        if err != nil {
+                t.Fatalf("failed to create fanout: %v", err)
+        }
+
+        defer func() {
+                if r := recover(); r == nil {
+                        t.Error("expected panic when passing nil input channel")
+                }
+        }()
+
+        fo.Run(t.Context(), nil)
+}