config

Documentation

Index

• func InferStreamID(dir, path string) (string, error)
• func ReadStreamFile(path string) (conf stream.Config, lints []string, err error)
• type MainUpdateFunc
• type OptFunc
  • func OptAddOverrides(overrides ...string) OptFunc
  • func OptSetStreamPaths(streamsPaths ...string) OptFunc
  • func OptTestSuffix(suffix string) OptFunc
• type Reader
  • func NewReader(mainPath string, resourcePaths []string, opts ...OptFunc) *Reader
  • func (r *Reader) BeginFileWatching(mgr bundle.NewManagement, strict bool) error
  • func (r *Reader) Close(ctx context.Context) error
  • func (r *Reader) Read(conf *config.Type) (lints []string, err error)
  • func (r *Reader) ReadStreams(confs map[string]stream.Config) (lints []string, err error)
  • func (r *Reader) SubscribeConfigChanges(fn MainUpdateFunc) error
  • func (r *Reader) SubscribeStreamChanges(fn StreamUpdateFunc) error
• type StreamUpdateFunc

Functions

func InferStreamID

func InferStreamID(dir, path string) (string, error)

InferStreamID attempts to infer a stream identifier from a file path and containing directory. If the dir field is non-empty then the identifier will include all sub-directories in the path as an id prefix, this means loading streams with the same file name from different branches are still given unique names.

func ReadStreamFile

func ReadStreamFile(path string) (conf stream.Config, lints []string, err error)

ReadStreamFile attempts to read a stream config and returns the result

Types

type MainUpdateFunc

type MainUpdateFunc func(conf stream.Config) bool

MainUpdateFunc is a closure function called whenever a main config has been updated. A boolean should be returned indicating whether the stream was successfully updated, if false then the attempt will be made again after a grace period.

type OptFunc

type OptFunc func(*Reader)

OptFunc is an opt function that changes the behaviour of a config reader.

func OptAddOverrides

func OptAddOverrides(overrides ...string) OptFunc

OptAddOverrides adds one or more override expressions to the config reader, each of the form `path=value`.

func OptSetStreamPaths

func OptSetStreamPaths(streamsPaths ...string) OptFunc

OptSetStreamPaths marks this config reader as operating in streams mode, and adds a list of paths to obtain individual stream configs from.

func OptTestSuffix

func OptTestSuffix(suffix string) OptFunc

OptTestSuffix configures the suffix given to unit test definition files, this is used in order to exclude unit tests from being run in streams mode with arbitrary directory walking.

type Reader

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

Reader provides utilities for parsing a Benthos config as a main file with a collection of resource files, and options such as overrides.

func NewReader

func NewReader(mainPath string, resourcePaths []string, opts ...OptFunc) *Reader

NewReader creates a new config reader.

func (*Reader) BeginFileWatching

func (r *Reader) BeginFileWatching(mgr bundle.NewManagement, strict bool) error

BeginFileWatching creates a goroutine that watches all active configuration files for changes. If a resource is changed then it is swapped out automatically through the provided manager. If a main config or stream config changes then the closures registered with either SubscribeConfigChanges or SubscribeStreamChanges will be called.

WARNING: Either SubscribeConfigChanges or SubscribeStreamChanges must be called before this, as otherwise it is unsafe to register them during watching.

func (*Reader) Close

func (r *Reader) Close(ctx context.Context) error

Close the reader, when this method exits all reloading will be stopped.

func (*Reader) Read

func (r *Reader) Read(conf *config.Type) (lints []string, err error)

Read a Benthos config from the files and options specified.

func (*Reader) ReadStreams

func (r *Reader) ReadStreams(confs map[string]stream.Config) (lints []string, err error)

ReadStreams attempts to read Benthos stream configs from one or more paths. Stream configs are extracted and added to a provided map, where the id is derived from the path of the stream config file.

func (*Reader) SubscribeConfigChanges

func (r *Reader) SubscribeConfigChanges(fn MainUpdateFunc) error

SubscribeConfigChanges registers a closure function to be called whenever the main configuration file is updated.

The provided closure should return true if the stream was successfully replaced.

func (*Reader) SubscribeStreamChanges

func (r *Reader) SubscribeStreamChanges(fn StreamUpdateFunc) error

SubscribeStreamChanges registers a closure to be called whenever the configuration of a stream is updated.

The provided closure should return true if the stream was successfully replaced.

type StreamUpdateFunc

type StreamUpdateFunc func(id string, conf stream.Config) bool

StreamUpdateFunc is a closure function called whenever a stream config has been updated. A boolean should be returned indicating whether the stream was successfully updated, if false then the attempt will be made again after a grace period.