octave

type ChannelFunc func(Emitter, Sink) error

type Decoder interface {
	// Decode decodes the next message into an interface.
	// It returns io.EOF once the end of the stream is reached.
	Decode(v interface{}) error

	io.Closer
}

type Emitter interface {
	// Name returns the name of the file the data is emitted from.
	Name() string

	// Decode decodes the next message into an interface.
	// It returns io.EOF once the end of the stream is reached.
	Decode(interface{}) error
}

type Encoder interface {
	// Encode encodes the value.
	Encode(v interface{}) error

	io.Closer
}

type Options struct {
	// Number of concurrent worker threads.
	// Default: number of CPUs
	Concurrency int

	// A custom temporary directory.
	// Default: os.TempDir()
	TempDir string

	// File glob pattern.
	// Default: "**"
	Glob string

	// NewDecoder wraps the reader and returns a decoder for the given file name.
	// Default: json.NewDecoder(reader)
	NewDecoder func(name string, reader io.Reader) (Decoder, error)

	// NewEncoder wraps the writer and returns an encoder for the given file name.
	// Default: json.NewEncoder(writer)
	NewEncoder func(name string, writer io.Writer) (Encoder, error)

	// NewCompressionReader wraps the reader and returns an io.ReadCloser.
	// It may return nil to disable decompression and read the plain input.
	// Default: gzip.NewReader(reader) (if name's extension is .gz)
	NewCompressionReader func(name string, reader io.Reader) (io.ReadCloser, error)

	// NewCompressionWriter wraps the writer and returns an io.WriteCloser.
	// It may return nil to disable compression and write the plain output.
	// Default: gzip.NewWriter(writer) (if name's extension is .gz)
	NewCompressionWriter func(name string, writer io.Writer) (io.WriteCloser, error)

	// Pause between cycles. This is to prevernt the Pipeline
	// from spinning and wasting resources on empty or processed
	// buckets.
	// Default: 5s.
	Pause time.Duration

	// BeforeCycle is a callback which is triggered before each cycles.
	BeforeCycle func() error

	// ProcessFile callback is triggered before processing to determine
	// if a file should be processed or skipped. Must return true to proceed.
	ProcessFile func(name string) (bool, error)
}

type Pipeline struct {
	// contains filtered or unexported fields
}

type Sink interface {
	// Encode encodes the value to the given file name.
	Encode(name string, v interface{}) error
}