Documentation
¶
Overview ¶
Package manager implements the types.Manager interface used for creating and sharing resources across a Benthos service.
Index ¶
- func Spec() docs.FieldSpecs
- type APIReg
- type ErrResourceNotFound
- type OptFunc
- type ResourceConfig
- type Type
- func (t *Type) AccessCache(ctx context.Context, name string, fn func(cache.V1)) error
- func (t *Type) AccessInput(ctx context.Context, name string, fn func(iinput.Streamed)) error
- func (t *Type) AccessOutput(ctx context.Context, name string, fn func(ioutput.Sync)) error
- func (t *Type) AccessProcessor(ctx context.Context, name string, fn func(iprocessor.V1)) error
- func (t *Type) AccessRateLimit(ctx context.Context, name string, fn func(ratelimit.V1)) error
- func (t *Type) BloblEnvironment() *bloblang.Environment
- func (t *Type) CloseAsync()
- func (t *Type) Environment() *bundle.Environment
- func (t *Type) ForStream(id string) interop.Manager
- func (t *Type) GetDocs(name string, ctype docs.Type) (docs.ComponentSpec, bool)
- func (t *Type) GetPipe(name string) (<-chan message.Transaction, error)
- func (t *Type) IntoPath(segments ...string) interop.Manager
- func (t *Type) Label() string
- func (t *Type) Logger() log.Modular
- func (t *Type) Metrics() metrics.Type
- func (t *Type) NewBuffer(conf buffer.Config) (buffer.Streamed, error)
- func (t *Type) NewCache(conf cache.Config) (cache.V1, error)
- func (t *Type) NewInput(conf input.Config, pipelines ...iprocessor.PipelineConstructorFunc) (iinput.Streamed, error)
- func (t *Type) NewOutput(conf output.Config, pipelines ...iprocessor.PipelineConstructorFunc) (ioutput.Streamed, error)
- func (t *Type) NewProcessor(conf processor.Config) (iprocessor.V1, error)
- func (t *Type) NewRateLimit(conf ratelimit.Config) (ratelimit.V1, error)
- func (t *Type) Path() []string
- func (t *Type) ProbeCache(name string) bool
- func (t *Type) ProbeInput(name string) bool
- func (t *Type) ProbeOutput(name string) bool
- func (t *Type) ProbeProcessor(name string) bool
- func (t *Type) ProbeRateLimit(name string) bool
- func (t *Type) RegisterEndpoint(apiPath, desc string, h http.HandlerFunc)
- func (t *Type) SetPipe(name string, tran <-chan message.Transaction)
- func (t *Type) StoreCache(ctx context.Context, name string, conf cache.Config) error
- func (t *Type) StoreInput(ctx context.Context, name string, conf input.Config) error
- func (t *Type) StoreOutput(ctx context.Context, name string, conf output.Config) error
- func (t *Type) StoreProcessor(ctx context.Context, name string, conf processor.Config) error
- func (t *Type) StoreRateLimit(ctx context.Context, name string, conf ratelimit.Config) error
- func (t *Type) UnsetPipe(name string, tran <-chan message.Transaction)
- func (t *Type) WaitForClose(timeout time.Duration) error
- func (t *Type) WithAddedMetrics(m metrics.Type) interop.Manager
- func (t *Type) WithMetricsMapping(m *metrics.Mapping) *Type
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type APIReg ¶
type APIReg interface {
RegisterEndpoint(path, desc string, h http.HandlerFunc)
}
APIReg is an interface representing an API builder.
type ErrResourceNotFound ¶
type ErrResourceNotFound string
ErrResourceNotFound represents an error where a named resource could not be accessed because it was not found by the manager.
func (ErrResourceNotFound) Error ¶
func (e ErrResourceNotFound) Error() string
Error implements the standard error interface.
type OptFunc ¶
type OptFunc func(*Type)
OptFunc is an opt setting for a manager type.
func OptSetBloblangEnvironment ¶
func OptSetBloblangEnvironment(env *bloblang.Environment) OptFunc
OptSetBloblangEnvironment determines the environment from which the manager parses bloblang functions and methods. This option is for internal use only.
func OptSetEnvironment ¶
func OptSetEnvironment(e *bundle.Environment) OptFunc
OptSetEnvironment determines the environment from which the manager initializes components and resources. This option is for internal use only.
type ResourceConfig ¶
type ResourceConfig struct {
ResourceInputs []input.Config `json:"input_resources,omitempty" yaml:"input_resources,omitempty"`
ResourceProcessors []processor.Config `json:"processor_resources,omitempty" yaml:"processor_resources,omitempty"`
ResourceOutputs []output.Config `json:"output_resources,omitempty" yaml:"output_resources,omitempty"`
ResourceCaches []cache.Config `json:"cache_resources,omitempty" yaml:"cache_resources,omitempty"`
ResourceRateLimits []ratelimit.Config `json:"rate_limit_resources,omitempty" yaml:"rate_limit_resources,omitempty"`
}
ResourceConfig contains fields for specifying resource components at the root of a Benthos config.
func NewResourceConfig ¶
func NewResourceConfig() ResourceConfig
NewResourceConfig creates a ResourceConfig with default values.
func (*ResourceConfig) AddFrom ¶
func (r *ResourceConfig) AddFrom(extra *ResourceConfig) error
AddFrom takes another Config and adds all of its resources to itself. If there are any resource name collisions an error is returned.
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 other resources.
func NewV2 ¶
func NewV2(conf ResourceConfig, apiReg APIReg, log log.Modular, stats *metrics.Namespaced, opts ...OptFunc) (*Type, error)
NewV2 returns an instance of manager.Type, which can be shared amongst components and logical threads of a Benthos service.
func (*Type) AccessCache ¶
AccessCache attempts to access a cache resource by a unique identifier and executes a closure function with the cache as an argument. Returns an error if the cache does not exist (or is otherwise inaccessible).
During the execution of the provided closure it is guaranteed that the resource will not be closed or removed. However, it is possible for the resource to be accessed by any number of components in parallel.
func (*Type) AccessInput ¶
AccessInput attempts to access an input resource by a unique identifier and executes a closure function with the input as an argument. Returns an error if the input does not exist (or is otherwise inaccessible).
During the execution of the provided closure it is guaranteed that the resource will not be closed or removed. However, it is possible for the resource to be accessed by any number of components in parallel.
func (*Type) AccessOutput ¶
AccessOutput attempts to access an output resource by a unique identifier and executes a closure function with the output as an argument. Returns an error if the output does not exist (or is otherwise inaccessible).
During the execution of the provided closure it is guaranteed that the resource will not be closed or removed. However, it is possible for the resource to be accessed by any number of components in parallel.
func (*Type) AccessProcessor ¶
AccessProcessor attempts to access a processor resource by a unique identifier and executes a closure function with the processor as an argument. Returns an error if the processor does not exist (or is otherwise inaccessible).
During the execution of the provided closure it is guaranteed that the resource will not be closed or removed. However, it is possible for the resource to be accessed by any number of components in parallel.
func (*Type) AccessRateLimit ¶
AccessRateLimit attempts to access a rate limit resource by a unique identifier and executes a closure function with the rate limit as an argument. Returns an error if the rate limit does not exist (or is otherwise inaccessible).
During the execution of the provided closure it is guaranteed that the resource will not be closed or removed. However, it is possible for the resource to be accessed by any number of components in parallel.
func (*Type) BloblEnvironment ¶
func (t *Type) BloblEnvironment() *bloblang.Environment
BloblEnvironment returns a Bloblang environment used by the manager. This is for internal use only.
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) Environment ¶
func (t *Type) Environment() *bundle.Environment
Environment returns a bundle environment used by the manager. This is for internal use only.
func (*Type) ForStream ¶
ForStream returns a variant of this manager to be used by a particular stream identifer, where APIs registered will be namespaced by that id.
func (*Type) GetPipe ¶
func (t *Type) GetPipe(name string) (<-chan message.Transaction, error)
GetPipe attempts to obtain and return a named output Pipe
func (*Type) IntoPath ¶
IntoPath returns a variant of this manager to be used by a particular component path, which is a child of the current component, where observability components will be automatically tagged with the new path.
func (*Type) NewInput ¶
func (t *Type) NewInput(conf input.Config, pipelines ...iprocessor.PipelineConstructorFunc) (iinput.Streamed, error)
NewInput attempts to create a new input component from a config.
func (*Type) NewOutput ¶
func (t *Type) NewOutput(conf output.Config, pipelines ...iprocessor.PipelineConstructorFunc) (ioutput.Streamed, error)
NewOutput attempts to create a new output component from a config.
func (*Type) NewProcessor ¶
NewProcessor attempts to create a new processor component from a config.
func (*Type) NewRateLimit ¶
NewRateLimit attempts to create a new rate limit component from a config.
func (*Type) ProbeCache ¶
ProbeCache returns true if a cache resource exists under the provided name.
func (*Type) ProbeInput ¶
ProbeInput returns true if an input resource exists under the provided name.
func (*Type) ProbeOutput ¶
ProbeOutput returns true if an output resource exists under the provided name.
func (*Type) ProbeProcessor ¶
ProbeProcessor returns true if a processor resource exists under the provided name.
func (*Type) ProbeRateLimit ¶
ProbeRateLimit returns true if a rate limit resource exists under the provided name.
func (*Type) RegisterEndpoint ¶
func (t *Type) RegisterEndpoint(apiPath, desc string, h http.HandlerFunc)
RegisterEndpoint registers a server wide HTTP endpoint.
func (*Type) SetPipe ¶
func (t *Type) SetPipe(name string, tran <-chan message.Transaction)
SetPipe registers a new transaction chan to a named pipe.
func (*Type) StoreCache ¶
StoreCache attempts to store a new cache resource. If an existing resource has the same name it is closed and removed _before_ the new one is initialized in order to avoid duplicate connections.
func (*Type) StoreInput ¶
StoreInput attempts to store a new input resource. If an existing resource has the same name it is closed and removed _before_ the new one is initialized in order to avoid duplicate connections.
func (*Type) StoreOutput ¶
StoreOutput attempts to store a new output resource. If an existing resource has the same name it is closed and removed _before_ the new one is initialized in order to avoid duplicate connections.
func (*Type) StoreProcessor ¶
StoreProcessor attempts to store a new processor resource. If an existing resource has the same name it is closed and removed _before_ the new one is initialized in order to avoid duplicate connections.
func (*Type) StoreRateLimit ¶
StoreRateLimit attempts to store a new rate limit resource. If an existing resource has the same name it is closed and removed _before_ the new one is initialized in order to avoid duplicate connections.
func (*Type) UnsetPipe ¶
func (t *Type) UnsetPipe(name string, tran <-chan message.Transaction)
UnsetPipe removes a named pipe transaction chan.
func (*Type) WaitForClose ¶
WaitForClose blocks until either all closable resource types are shut down or a timeout occurs.
func (*Type) WithAddedMetrics ¶
WithAddedMetrics returns a modified version of the manager where metrics are registered to both the current metrics target as well as the provided one.
Source Files
Directories