1 files changed, 34 insertions, 0 deletions

diff --git a/pipeline/options.go b/pipeline/options.go
new file mode 100644
index 0000000..c8c616e
--- /dev/null
+++ b/pipeline/options.go
@@ -0,0 +1,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
+}