manager

Documentation

Overview

Package manager creates and manages multiple streams, providing an API for performing CRUD operations.

Index

• Variables
• func LoadStreamConfigsFromDirectory(replaceEnvVars bool, dir string) (map[string]stream.Config, error)
• func OptAddInputPipelines(pipes ...StreamPipeConstructorFunc) func(*Type)
• func OptAddOutputPipelines(pipes ...StreamPipeConstructorFunc) func(*Type)
• func OptAddProcessors(procs ...StreamProcConstructorFunc) func(*Type)
• func OptSetAPITimeout(tout time.Duration) func(*Type)
• func OptSetLogger(log log.Modular) func(*Type)
• func OptSetManager(mgr types.Manager) func(*Type)
• func OptSetStats(stats metrics.Type) func(*Type)
• type ConfigSet
  • func (c ConfigSet) UnmarshalYAML(unmarshal func(interface{}) error) error
• type StreamPipeConstructorFunc
• type StreamProcConstructorFunc
• type StreamStatus
• type Type
  • func New(opts ...func(*Type)) *Type
  • func (m *Type) Create(id string, conf stream.Config) error
  • func (m *Type) Delete(id string, timeout time.Duration) error
  • func (m *Type) HandleStreamCRUD(w http.ResponseWriter, r *http.Request)
  • func (m *Type) HandleStreamsCRUD(w http.ResponseWriter, r *http.Request)
  • func (m *Type) Read(id string) (StreamStatus, error)
  • func (m *Type) Stop(timeout time.Duration) error
  • func (m *Type) Update(id string, conf stream.Config, timeout time.Duration) error

Variables

var (
	ErrStreamExists       = errors.New("stream already exists")
	ErrStreamDoesNotExist = errors.New("stream does not exist")
)

Errors specifically returned by a stream manager.

Functions

func LoadStreamConfigsFromDirectory

func LoadStreamConfigsFromDirectory(replaceEnvVars bool, dir string) (map[string]stream.Config, error)

LoadStreamConfigsFromDirectory reads a map of stream ids to configurations by walking a directory of .json and .yaml files.

func OptAddInputPipelines

func OptAddInputPipelines(pipes ...StreamPipeConstructorFunc) func(*Type)

OptAddInputPipelines adds pipeline constructors that will be called for every new stream and attached to the input component. The constructor is given the name of the stream as an argument.

func OptAddOutputPipelines

func OptAddOutputPipelines(pipes ...StreamPipeConstructorFunc) func(*Type)

OptAddOutputPipelines adds pipeline constructors that will be called for every new stream and attached to the output component. The constructor is given the name of the stream as an argument.

func OptAddProcessors

func OptAddProcessors(procs ...StreamProcConstructorFunc) func(*Type)

OptAddProcessors adds processor constructors that will be called for every new stream and attached to the processor pipelines. The constructor is given the name of the stream as an argument.

func OptSetAPITimeout

func OptSetAPITimeout(tout time.Duration) func(*Type)

OptSetAPITimeout sets the default timeout for HTTP API requests.

func OptSetLogger

func OptSetLogger(log log.Modular) func(*Type)

OptSetLogger sets the logging output to be used by the manager and all child streams.

func OptSetManager

func OptSetManager(mgr types.Manager) func(*Type)

OptSetManager sets the service manager to be used by the stream manager and all child streams.

func OptSetStats

func OptSetStats(stats metrics.Type) func(*Type)

OptSetStats sets the metrics aggregator to be used by the manager and all child streams.

Types

type ConfigSet

type ConfigSet map[string]stream.Config

ConfigSet is a map of stream configurations mapped by ID, which can be YAML parsed without losing default values inside the stream configs.

func (ConfigSet) UnmarshalYAML

func (c ConfigSet) 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 StreamPipeConstructorFunc

type StreamPipeConstructorFunc func(streamID string) (pipeline.Type, error)

StreamPipeConstructorFunc is a closure type that constructs a pipeline type for new streams, where the id of the stream is provided as an argument.

type StreamProcConstructorFunc

type StreamProcConstructorFunc func(streamID string) (processor.Type, error)

StreamProcConstructorFunc is a closure type that constructs a processor type for new streams, where the id of the stream is provided as an argument.

type StreamStatus

type StreamStatus struct {
	Active bool
	Uptime time.Duration
	Config stream.Config
}

StreamStatus contains fields used to describe the current status of a managed stream.

type Type

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

Type manages a collection of streams, providing APIs for CRUD operations on the streams.

func New

func New(opts ...func(*Type)) *Type

New creates a new stream manager.Type.

func (*Type) Create

func (m *Type) Create(id string, conf stream.Config) error

Create attempts to construct and run a new stream under a unique ID. If the ID already exists an error is returned.

func (*Type) Delete

func (m *Type) Delete(id string, timeout time.Duration) error

Delete attempts to stop and remove a stream by its ID. Returns an error if the stream was not found, or if clean shutdown fails in the specified period of time.

func (*Type) HandleStreamCRUD

func (m *Type) HandleStreamCRUD(w http.ResponseWriter, r *http.Request)

HandleStreamCRUD is an http.HandleFunc for performing CRUD operations on individual streams.

func (*Type) HandleStreamsCRUD

func (m *Type) HandleStreamsCRUD(w http.ResponseWriter, r *http.Request)

HandleStreamsCRUD is an http.HandleFunc for returning maps of active benthos streams by their id, status and uptime or overwriting the entire set of streams.

func (*Type) Read

func (m *Type) Read(id string) (StreamStatus, error)

Read attempts to obtain the status of a managed stream. Returns an error if the stream does not exist.

func (*Type) Stop

func (m *Type) Stop(timeout time.Duration) error

Stop attempts to gracefully shut down all active streams and close the stream manager.

func (*Type) Update

func (m *Type) Update(id string, conf stream.Config, timeout time.Duration) error

Update attempts to stop an existing stream and replace it with a new version of the same stream.