manager

Documentation

Overview

Package manager implements the types.Manager interface used for creating and sharing resources across a Benthos service.

Index

• func AddExamples(c *Config)
• func DocumentPlugin(typeString, description string, configSanitiser PluginConfigSanitiser)
• func PluginDescriptions() string
• func RegisterPlugin(typeString string, configConstructor PluginConfigConstructor, ...)
• func SanitiseConfig(conf Config) (interface{}, error)
• type APIReg
• type Config
  • func NewConfig() Config
• type PluginConfig
  • func (p *PluginConfig) UnmarshalJSON(bytes []byte) error
  • func (p *PluginConfig) UnmarshalYAML(unmarshal func(interface{}) error) error
• type PluginConfigConstructor
• type PluginConfigSanitiser
• type PluginConstructor
• type Type
  • func New(conf Config, apiReg APIReg, log log.Modular, stats metrics.Type) (*Type, error)
  • func (t *Type) CloseAsync()
  • func (t *Type) GetCache(name string) (types.Cache, error)
  • func (t *Type) GetCondition(name string) (types.Condition, error)
  • func (t *Type) GetInput(name string) (types.Input, error)
  • func (t *Type) GetOutput(name string) (types.OutputWriter, error)
  • func (t *Type) GetPipe(name string) (<-chan types.Transaction, error)
  • func (t *Type) GetPlugin(name string) (interface{}, error)
  • func (t *Type) GetProcessor(name string) (types.Processor, error)
  • func (t *Type) GetRateLimit(name string) (types.RateLimit, error)
  • func (t *Type) RegisterEndpoint(path, desc string, h http.HandlerFunc)
  • func (t *Type) SetPipe(name string, tran <-chan types.Transaction)
  • func (t *Type) UnsetPipe(name string, tran <-chan types.Transaction)
  • func (t *Type) WaitForClose(timeout time.Duration) error

Functions

func AddExamples

func AddExamples(c *Config)

AddExamples inserts example caches and conditions if none exist in the config.

func DocumentPlugin

func DocumentPlugin(
	typeString, description string,
	configSanitiser PluginConfigSanitiser,
)

DocumentPlugin adds a description and an optional configuration sanitiser function to the definition of a registered plugin. This improves the documentation generated by PluginDescriptions.

func PluginDescriptions

func PluginDescriptions() string

PluginDescriptions generates and returns a markdown formatted document listing each registered plugin and an example configuration for it.

func RegisterPlugin

func RegisterPlugin(
	typeString string,
	configConstructor PluginConfigConstructor,
	constructor PluginConstructor,
)

RegisterPlugin registers a plugin by a unique name so that it can be constucted similar to regular resources. A constructor for both the plugin itself as well as its configuration struct must be provided.

A constructed resource plugin can be any type and is wrapped as an interface{} type.

func SanitiseConfig

func SanitiseConfig(conf Config) (interface{}, error)

SanitiseConfig creates a sanitised version of a manager config.

Types

type APIReg

type APIReg interface {
	RegisterEndpoint(path, desc string, h http.HandlerFunc)
}

APIReg is an interface representing an API builder.

type Config

type Config struct {
	Inputs     map[string]input.Config     `json:"inputs" yaml:"inputs"`
	Conditions map[string]condition.Config `json:"conditions" yaml:"conditions"`
	Processors map[string]processor.Config `json:"processors" yaml:"processors"`
	Outputs    map[string]output.Config    `json:"outputs" yaml:"outputs"`
	Caches     map[string]cache.Config     `json:"caches" yaml:"caches"`
	RateLimits map[string]ratelimit.Config `json:"rate_limits" yaml:"rate_limits"`
	Plugins    map[string]PluginConfig     `json:"plugins,omitempty" yaml:"plugins,omitempty"`
}

Config contains all configuration fields for a Benthos service manager.

func NewConfig

func NewConfig() Config

NewConfig returns a Config with default values.

type PluginConfig

type PluginConfig struct {
	Type   string      `json:"type" yaml:"type"`
	Plugin interface{} `json:"plugin" yaml:"plugin"`
}

PluginConfig is a config struct representing a resource plugin.

func (*PluginConfig) UnmarshalJSON

func (p *PluginConfig) UnmarshalJSON(bytes []byte) error

UnmarshalJSON ensures that when parsing configs that are in a map or slice the default values are still applied.

func (*PluginConfig) UnmarshalYAML

func (p *PluginConfig) UnmarshalYAML(unmarshal func(interface{}) error) error

UnmarshalYAML ensures that when parsing configs that are in a map or slice the default values are still applied.

type PluginConfigConstructor

type PluginConfigConstructor func() interface{}

PluginConfigConstructor is a func that returns a pointer to a new and fully populated configuration struct for a plugin type. It is valid to return a pointer to an empty struct (&struct{}{}) if no configuration fields are needed.

type PluginConfigSanitiser

type PluginConfigSanitiser func(conf interface{}) interface{}

PluginConfigSanitiser is a function that takes a configuration object for a plugin and returns a sanitised (minimal) version of it for printing in examples and plugin documentation.

This function is useful for when a plugins configuration struct is very large and complex, but can sometimes be expressed in a more concise way without losing the original intent.

type PluginConstructor

type PluginConstructor func(
	config interface{},
	manager types.Manager,
	logger log.Modular,
	metrics metrics.Type,
) (interface{}, error)

PluginConstructor is a func that constructs a Benthos resource plugin. These are shareable resources that can be accessed by any other type within Benthos.

The configuration object will be the result of the PluginConfigConstructor after overlaying the user configuration.

type Type

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

Type is an implementation of types.Manager, which is expected by Benthos components that need to register service wide behaviours such as HTTP endpoints and event listeners, and obtain service wide shared resources such as caches and labelled conditions.

func New

func New(
	conf Config,
	apiReg APIReg,
	log log.Modular,
	stats metrics.Type,
) (*Type, error)

New returns an instance of manager.Type, which can be shared amongst components and logical threads of a Benthos service.

func (*Type) CloseAsync

func (t *Type) CloseAsync()

CloseAsync triggers the shut down of all resource types that implement the lifetime interface types.Closable.

func (*Type) GetCache

func (t *Type) GetCache(name string) (types.Cache, error)

GetCache attempts to find a service wide cache by its name.

func (*Type) GetCondition

func (t *Type) GetCondition(name string) (types.Condition, error)

GetCondition attempts to find a service wide condition by its name.

func (*Type) GetInput

func (t *Type) GetInput(name string) (types.Input, error)

GetInput attempts to find a service wide input by its name.

func (*Type) GetOutput

func (t *Type) GetOutput(name string) (types.OutputWriter, error)

GetOutput attempts to find a service wide output by its name.

func (*Type) GetPipe

func (t *Type) GetPipe(name string) (<-chan types.Transaction, error)

GetPipe attempts to obtain and return a named output Pipe

func (*Type) GetPlugin

func (t *Type) GetPlugin(name string) (interface{}, error)

GetPlugin attempts to find a service wide resource plugin by its name.

func (*Type) GetProcessor

func (t *Type) GetProcessor(name string) (types.Processor, error)

GetProcessor attempts to find a service wide processor by its name.

func (*Type) GetRateLimit

func (t *Type) GetRateLimit(name string) (types.RateLimit, error)

GetRateLimit attempts to find a service wide rate limit by its name.

func (*Type) RegisterEndpoint

func (t *Type) RegisterEndpoint(path, desc string, h http.HandlerFunc)

RegisterEndpoint registers a server wide HTTP endpoint.

func (*Type) SetPipe

func (t *Type) SetPipe(name string, tran <-chan types.Transaction)

SetPipe registers a new transaction chan to a named pipe.

func (*Type) UnsetPipe

func (t *Type) UnsetPipe(name string, tran <-chan types.Transaction)

UnsetPipe removes a named pipe transaction chan.

func (*Type) WaitForClose

func (t *Type) WaitForClose(timeout time.Duration) error

WaitForClose blocks until either all closable resource types are shut down or a timeout occurs.