Verified Moving all legacy quacker source code from github to cgitHEADmain

4 files changed, 494 insertions, 0 deletions

diff --git a/fanin/example_test.go b/fanin/example_test.go
new file mode 100644
index 0000000..0edae08
--- /dev/null
+++ b/fanin/example_test.go
@@ -0,0 +1,35 @@
+package fanin_test
+
+import (
+        "context"
+        "fmt"
+        "sort"
+
+        "git.sangtran.io.vn/quacker.git/fanin"
+)
+
+func ExampleNewFanIn() {
+        fi, _ := fanin.NewFanIn[int]()
+
+        ch1 := make(chan int, 2)
+        ch1 <- 1
+        ch1 <- 2
+        close(ch1)
+
+        ch2 := make(chan int, 2)
+        ch2 <- 3
+        ch2 <- 4
+        close(ch2)
+
+        fi.Add(ch1)
+        fi.Add(ch2)
+
+        var results []int
+        for v := range fi.Run(context.Background()) {
+                results = append(results, v)
+        }
+
+        sort.Ints(results)
+        fmt.Println(results)
+        // Output: [1 2 3 4]
+}
diff --git a/fanin/fanin.go b/fanin/fanin.go
new file mode 100644
index 0000000..913927f
--- /dev/null
+++ b/fanin/fanin.go
@@ -0,0 +1,142 @@
+// Package fanin merges multiple input channels into a single output channel.
+package fanin
+
+import (
+        "context"
+        "fmt"
+        "sync"
+        "sync/atomic"
+)
+
+// FanInObserver defines callbacks for monitoring fan-in lifecycle events.
+type FanInObserver interface {
+        OnInputAdded()
+        OnInputClosed()
+}
+
+// FanInConfig holds configuration options for creating a FanIn instance.
+type FanInConfig struct {
+        BufferSize int
+        Observer   FanInObserver
+}
+
+// Option is a functional option for configuring a FanIn instance.
+type Option func(*FanInConfig)
+
+// FanIn merges multiple input channels of type T into a single output channel.
+type FanIn[T any] struct {
+        cfg     *FanInConfig
+        mu      sync.Mutex
+        inputs  []<-chan T
+        running atomic.Bool
+        done    chan struct{}
+}
+
+// NewFanIn creates a new FanIn instance with the provided options.
+func NewFanIn[T any](opts ...Option) (*FanIn[T], error) {
+        cfg := &FanInConfig{}
+
+        for _, opt := range opts {
+                opt(cfg)
+        }
+
+        if cfg.BufferSize < 0 {
+                return nil, fmt.Errorf("fanin: buffer size must not be negative, got %d", cfg.BufferSize)
+        }
+
+        return &FanIn[T]{
+                cfg: cfg,
+        }, nil
+}
+
+// WithBufferSize sets the buffer capacity for the merged output channel.
+func WithBufferSize(size int) Option {
+        return func(f *FanInConfig) {
+                f.BufferSize = size
+        }
+}
+
+// WithObserver attaches a FanInObserver to receive lifecycle event callbacks.
+func WithObserver(observer FanInObserver) Option {
+        return func(f *FanInConfig) {
+                f.Observer = observer
+        }
+}
+
+// Add registers an input channel to be merged into the output.
+// It must be called before Run, and is safe for concurrent use.
+func (f *FanIn[T]) Add(ch <-chan T) {
+        if ch == nil {
+                panic("fanin: cannot add nil channel")
+        }
+
+        if f.running.Load() {
+                panic("fanin: Add() called after Run()")
+        }
+
+        f.mu.Lock()
+        defer f.mu.Unlock()
+
+        f.inputs = append(f.inputs, ch)
+
+        if f.cfg.Observer != nil {
+                safeObserve(f.cfg.Observer.OnInputAdded)
+        }
+}
+
+// Run starts merging the registered input channels into a single output channel.
+func (f *FanIn[T]) Run(ctx context.Context) <-chan T {
+        if !f.running.CompareAndSwap(false, true) {
+                panic("fanin: Run() called multiple times")
+        }
+
+        f.mu.Lock()
+        inputs := append([]<-chan T(nil), f.inputs...)
+        out := make(chan T, f.cfg.BufferSize)
+        f.done = make(chan struct{})
+        f.mu.Unlock()
+
+        var wg sync.WaitGroup
+        for _, ch := range inputs {
+                wg.Go(func() {
+                        for {
+                                select {
+                                case v, ok := <-ch:
+                                        if !ok {
+                                                if f.cfg.Observer != nil {
+                                                        safeObserve(f.cfg.Observer.OnInputClosed)
+                                                }
+                                                return
+                                        }
+                                        select {
+                                        case out <- v:
+                                        case <-ctx.Done():
+                                                return
+                                        }
+                                case <-ctx.Done():
+                                        return
+                                }
+                        }
+                })
+        }
+
+        go func() {
+                wg.Wait()
+                close(out)
+                close(f.done)
+        }()
+
+        return out
+}
+
+func safeObserve(fn func()) {
+        defer func() { recover() }()
+        fn()
+}
+
+// Done returns a channel that is closed when all inputs are processed or the context is cancelled.
+func (f *FanIn[T]) Done() <-chan struct{} {
+        f.mu.Lock()
+        defer f.mu.Unlock()
+        return f.done
+}
diff --git a/fanin/fanin_bench_test.go b/fanin/fanin_bench_test.go
new file mode 100644
index 0000000..1141224
--- /dev/null
+++ b/fanin/fanin_bench_test.go
@@ -0,0 +1,36 @@
+package fanin
+
+import (
+        "testing"
+)
+
+func BenchmarkFanIn(b *testing.B) {
+        b.ReportAllocs()
+
+        fi, _ := NewFanIn[int](WithBufferSize(64))
+
+        producers := 4
+        chs := make([]chan int, producers)
+        for i := range chs {
+                chs[i] = make(chan int, 64)
+                fi.Add(chs[i])
+        }
+
+        out := fi.Run(b.Context())
+
+        // Drain output
+        go func() {
+                for range out {
+                }
+        }()
+
+        for b.Loop() {
+                chs[b.N%producers] <- 1
+        }
+
+        for _, ch := range chs {
+                close(ch)
+        }
+
+        <-fi.Done()
+}
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()
+}