acp/opt.go

package acp

import (
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/sirupsen/logrus"
)

// source is one enumerated source file: base is the directory the relative path is resolved
// against, and path is the source-relative path that also maps onto a target directory.
type source struct {
	info os.FileInfo
	base string
	path string
}

func (s *source) src() string {
	return filepath.Join(s.base, s.path)
}

func (s *source) dst(dst string) string {
	return filepath.Join(dst, s.path)
}

func (s *source) append(next string) *source {
	return &source{base: s.base, path: filepath.Join(s.path, next)}
}

func comparePath(a, b string) int {
	a = strings.ReplaceAll(filepath.ToSlash(a), "/", "\x00")
	b = strings.ReplaceAll(filepath.ToSlash(b), "/", "\x00")
	return strings.Compare(a, b)
}

const (
	// defaultReadBuffer is how many items ACP may hold in its read buffer when the caller does
	// not set one. The read buffer is the backpressure limit of the feed.
	defaultReadBuffer = 4096
	// defaultResultBuffer is how many finished results ACP may hold for delivery before the
	// pipeline that produces them blocks.
	defaultResultBuffer = 256
	// defaultResultBatch is how many results one delivery carries.
	defaultResultBatch = 256
	// defaultResultFlushInterval is how long a successful result may wait for its batch.
	defaultResultFlushInterval = time.Second
	// minResultFlushInterval is the shortest result flush interval ACP accepts.
	minResultFlushInterval = 100 * time.Millisecond
)

type option struct {
	// Job options describe what the af05f05c compatibility shell enumerates.
	accurateJobs []*accurateJob
	wildcardJobs []*wildcardJob

	fromDevice *deviceOption
	toDevice   *deviceOption

	createFlag          int
	hashPolicy          HashPolicy
	readBuffer          int
	resultBuffer        int
	resultBatch         int
	resultFlushInterval time.Duration

	logger        *logrus.Logger
	eventHandlers []EventHandler
}

func newOption() *option {
	return &option{
		fromDevice:          new(deviceOption),
		toDevice:            new(deviceOption),
		createFlag:          os.O_WRONLY | os.O_CREATE | os.O_EXCL,
		readBuffer:          defaultReadBuffer,
		resultBuffer:        defaultResultBuffer,
		resultBatch:         defaultResultBatch,
		resultFlushInterval: defaultResultFlushInterval,
	}
}

// check validates the options and settles their defaults. Every value a caller can get wrong is
// rejected here, which is what makes New and NewStream report creation errors instead of
// failing midway through a run.
func (o *option) check() error {
	for _, job := range o.wildcardJobs {
		if err := job.check(); err != nil {
			return err
		}
	}
	if err := o.fromDevice.check(); err != nil {
		return fmt.Errorf("check source device failed, %w", err)
	}
	if err := o.toDevice.check(); err != nil {
		return fmt.Errorf("check target device failed, %w", err)
	}
	if o.toDevice.readMode != ReadBuffered {
		return fmt.Errorf("read mode is a source option, mode= %s", o.toDevice.readMode)
	}
	if o.readBuffer < 1 {
		return fmt.Errorf("read buffer must be at least one item, items=%d", o.readBuffer)
	}
	if o.resultBuffer < 1 {
		return fmt.Errorf("result buffer must be at least one item, items=%d", o.resultBuffer)
	}
	if o.resultBatch < 1 {
		return fmt.Errorf("result batch must be at least one item, items=%d", o.resultBatch)
	}
	if o.resultFlushInterval < minResultFlushInterval {
		return fmt.Errorf("result flush interval must be at least %s, interval= %s", minResultFlushInterval, o.resultFlushInterval)
	}
	if !o.hashPolicy.valid() {
		return fmt.Errorf("unknown hash policy, policy= %s", o.hashPolicy)
	}
	if o.logger == nil {
		o.logger = logrus.StandardLogger()
	}

	return nil
}

// Option configures one run. A repeated value option is last-wins, so a later value replaces an
// earlier one instead of combining with it; the job options and SetFromDevice/SetToDevice
// accumulate, so a second WildcardJob adds another walk and a second device option adjusts the
// same device description.
type Option func(*option) *option

func SetFromDevice(opts ...DeviceOption) Option {
	return func(o *option) *option {
		for _, opt := range opts {
			if opt == nil {
				continue
			}
			o.fromDevice = opt(o.fromDevice)
		}
		return o
	}
}

func SetToDevice(opts ...DeviceOption) Option {
	return func(o *option) *option {
		for _, opt := range opts {
			if opt == nil {
				continue
			}
			o.toDevice = opt(o.toDevice)
		}
		return o
	}
}

// Overwrite selects whether an existing target file is replaced or refused. It is a run-level
// option and applies to every requested target.
func Overwrite(b bool) Option {
	return func(o *option) *option {
		if b {
			o.createFlag = os.O_WRONLY | os.O_CREATE | os.O_TRUNC
			return o
		}

		o.createFlag = os.O_WRONLY | os.O_CREATE | os.O_EXCL
		return o
	}
}

// WithReadBuffer sets how many items ACP may hold in its read buffer, which is also the feed's
// backpressure limit: Submit blocks while the buffer is full. It defaults to 4096 items.
func WithReadBuffer(items int) Option {
	return func(o *option) *option {
		o.readBuffer = items
		return o
	}
}

// WithResultBuffer sets how many finished results ACP may hold for delivery, which is the
// backpressure limit of the result path: a pipeline stage that produces a result blocks while the
// result buffer is full. A result that carries an error is delivered as soon as it is read, so it
// does not wait for room in a batch. It defaults to 256 items.
func WithResultBuffer(items int) Option {
	return func(o *option) *option {
		o.resultBuffer = items
		return o
	}
}

// WithResultBatch sets how many results one delivery carries: ACP hands the caller's results
// callback at most this many results at a time. It defaults to 256 items.
func WithResultBatch(items int) Option {
	return func(o *option) *option {
		o.resultBatch = items
		return o
	}
}

// WithResultFlushInterval sets how long a successful result may wait for the rest of its batch. It
// defaults to one second and accepts no value below 100ms.
func WithResultFlushInterval(d time.Duration) Option {
	return func(o *option) *option {
		o.resultFlushInterval = d
		return o
	}
}

func WithProgressBar() Option {
	return WithEventHandler(NewProgressBar())
}

// WithHashPolicy sets the content hash policy, which also decides how the stored hash cache is
// read and refreshed. It defaults to HashOff.
func WithHashPolicy(policy HashPolicy) Option {
	return func(o *option) *option {
		o.hashPolicy = policy
		return o
	}
}

func WithLogger(logger *logrus.Logger) Option {
	return func(o *option) *option {
		o.logger = logger
		return o
	}
}

// WithEventHandler registers the run's event handler. Repeated options are last-wins, so this
// replaces every handler registered before it, and a nil handler clears the registration
// instead of leaving a handler that fails the run when it is called.
func WithEventHandler(h EventHandler) Option {
	return func(o *option) *option {
		o.eventHandlers = nil
		if h == nil {
			return o
		}

		o.eventHandlers = []EventHandler{h}
		return o
	}
}