blob: c8c616efbd7bbde4e4ff8b04f94c039c10749f12 (
plain)
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
|
package pipeline
// StageObserver receives callbacks for stage lifecycle events.
type StageObserver interface {
OnItem()
OnDrop()
OnDone()
}
// StageOption configures a stage's behavior.
type StageOption func(*stageConfig)
type stageConfig struct {
observer StageObserver
bufferSize int
}
// WithObserver attaches an observer to a stage.
func WithObserver(obs StageObserver) StageOption {
return func(cfg *stageConfig) { cfg.observer = obs }
}
// WithBufferSize sets the output channel buffer size for a stage.
func WithBufferSize(size int) StageOption {
return func(cfg *stageConfig) { cfg.bufferSize = size }
}
func applyOpts(opts []StageOption) stageConfig {
var cfg stageConfig
for _, opt := range opts {
opt(&cfg)
}
return cfg
}
|
