processor.go

package processor

import (
	"errors"
	"log/slog"

	"go.opentelemetry.io/otel/metric"
)

var (
	ErrControlNotSupported = errors.New("control not supported")
	ErrProcessorNotRunning = errors.New("processor not running")
	ErrAlreadyRunning      = errors.New("already running")
	ErrAlreadyPaused       = errors.New("already paused")
	ErrProcessorBusy       = errors.New("processor busy")
	ErrUnableToStart       = errors.New("unable to start")
	ErrMultipleStart       = errors.New("multiple start")
	ErrMultipleStop        = errors.New("multiple stop")
)

// SkipResult is a sentinel error that signals a processor should skip output generation
// for the current input. Effective for synchronous processors; async processors will
// log a warning as it doesn't affect their separate Output() channels.
var SkipResult = errors.New("skip result")

type config struct {
	label              *string
	logger             *slog.Logger
	startPaused        bool
	blockOnOutput      bool
	useRoundRobinFanIn bool // TODO: warn if set on non-N processor
	outputChannelSize  int

	meterProvider metric.MeterProvider
}

type Option func(*config)

func WithLogger(logger *slog.Logger) Option {
	return func(c *config) {
		c.logger = logger
	}
}

func WithLabel(label string) Option {
	return func(c *config) {
		c.label = &label
	}
}

func StartPaused() Option {
	return func(c *config) {
		c.startPaused = true
	}
}

func BlockOnOutput() Option {
	return func(c *config) {
		c.blockOnOutput = true
	}
}

func UseRoundRobinFanIn() Option {
	return func(c *config) {
		c.useRoundRobinFanIn = true
	}
}

func WithOutputChannelSize(size int) Option {
	return func(c *config) {
		c.outputChannelSize = max(0, size)
	}
}

func WithMeterProvider(mp metric.MeterProvider) Option {
	return func(c *config) {
		c.meterProvider = mp
	}
}