Documentation
¶
Index ¶
- func InferStreamID(dir, path string) (string, error)
- func LintBytes(ctx docs.LintContext, rawBytes []byte) ([]string, error)
- func MarshalYAML(v yaml.Node) ([]byte, error)
- func ReadFileEnvSwap(path string) (configBytes []byte, lints []string, err error)
- func ReadFileLinted(path string, rejectDeprecated bool, config *Type) ([]string, error)
- func ReadStreamFile(path string) (conf stream.Config, lints []string, err error)
- func ReplaceEnvVariables(inBytes []byte) []byte
- func Spec() docs.FieldSpecs
- func SpecWithoutStream() docs.FieldSpecs
- type MainUpdateFunc
- type OptFunc
- type Reader
- func (r *Reader) BeginFileWatching(mgr bundle.NewManagement, strict bool) error
- func (r *Reader) Close(ctx context.Context) error
- func (r *Reader) Read(conf *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
- type Type
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func InferStreamID ¶
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 LintBytes ¶
func LintBytes(ctx docs.LintContext, rawBytes []byte) ([]string, error)
LintBytes attempts to report errors within a user config. Returns a slice of lint results.
func MarshalYAML ¶
MarshalYAML marshals a structure into YAML with consistent formatting across all Benthos components.
func ReadFileEnvSwap ¶
ReadFileEnvSwap reads a file and replaces any environment variable interpolations before returning the contents. Linting errors are returned if the file has an unexpected higher level format, such as invalid utf-8 encoding.
func ReadFileLinted ¶
ReadFileLinted will attempt to read a configuration file path into a structure. Returns an array of lint messages or an error.
func ReadStreamFile ¶
ReadStreamFile attempts to read a stream config and returns the result
func ReplaceEnvVariables ¶
ReplaceEnvVariables will search a blob of data for the pattern `${FOO:bar}`, where `FOO` is an environment variable name and `bar` is a default value. The `bar` section (including the colon) can be left out if there is no appropriate default value for the field.
For each aforementioned pattern found in the blob the contents of the respective environment variable will be read and will replace the pattern. If the environment variable is empty or does not exist then either the default value is used or the field will be left empty.
func Spec ¶
func Spec() docs.FieldSpecs
Spec returns a docs.FieldSpec for an entire Benthos configuration.
func SpecWithoutStream ¶
func SpecWithoutStream() docs.FieldSpecs
SpecWithoutStream describes a stream config without the core stream fields.
Types ¶
type MainUpdateFunc ¶
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 ¶
OptAddOverrides adds one or more override expressions to the config reader, each of the form `path=value`.
func OptSetStreamPaths ¶
OptSetStreamPaths marks this config reader as operating in streams mode, and adds a list of paths to obtain individual stream configs from.
func OptTestSuffix ¶
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 (*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) ReadStreams ¶
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 ¶
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.
type Type ¶
type Type struct {
HTTP api.Config `json:"http" yaml:"http"`
stream.Config `json:",inline" yaml:",inline"`
manager.ResourceConfig `json:",inline" yaml:",inline"`
Logger log.Config `json:"logger" yaml:"logger"`
Metrics metrics.Config `json:"metrics" yaml:"metrics"`
Tracer tracer.Config `json:"tracer" yaml:"tracer"`
SystemCloseTimeout string `json:"shutdown_timeout" yaml:"shutdown_timeout"`
Tests []interface{} `json:"tests,omitempty" yaml:"tests,omitempty"`
}
Type is the Benthos service configuration struct.
Source Files
Directories