1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
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
}
|
