From 02123df8cc15896f97c1580e49d8e2a991e37ddd Mon Sep 17 00:00:00 2001 From: sangtran Date: Thu, 3 Sep 2026 23:10:33 +0700 Subject: Moving all legacy quacker source code from github to cgit --- fanin/example_test.go | 35 ++++++ fanin/fanin.go | 142 +++++++++++++++++++++++ fanin/fanin_bench_test.go | 36 ++++++ fanin/fanin_test.go | 281 ++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 494 insertions(+) create mode 100644 fanin/example_test.go create mode 100644 fanin/fanin.go create mode 100644 fanin/fanin_bench_test.go create mode 100644 fanin/fanin_test.go (limited to 'fanin') 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() +} -- cgit v1.3.1