stream

Documentation

Index

• func OptOnClose(onClose func()) func(*Type)
• func Spec() docs.FieldSpecs
• type Config
  • func NewConfig() Config
  • func (c Config) Sanitised() (interface{}, error)
• type Type
  • func New(conf Config, mgr bundle.NewManagement, opts ...func(*Type)) (*Type, error)
  • func (t *Type) IsReady() bool
  • func (t *Type) Stop(timeout time.Duration) error
  • func (t *Type) StopGracefully(timeout time.Duration) (err error)
  • func (t *Type) StopOrdered(timeout time.Duration) (err error)
  • func (t *Type) StopUnordered(timeout time.Duration) (err error)

Functions

func OptOnClose

func OptOnClose(onClose func()) func(*Type)

OptOnClose sets a closure to be called when the stream closes.

func Spec

func Spec() docs.FieldSpecs

Spec returns a docs.FieldSpec for a stream configuration.

Types

type Config

type Config struct {
	Input    input.Config    `json:"input" yaml:"input"`
	Buffer   buffer.Config   `json:"buffer" yaml:"buffer"`
	Pipeline pipeline.Config `json:"pipeline" yaml:"pipeline"`
	Output   output.Config   `json:"output" yaml:"output"`
}

Config is a configuration struct representing all four layers of a Benthos stream.

func NewConfig

func NewConfig() Config

NewConfig returns a new configuration with default values.

func (Config) Sanitised

func (c Config) Sanitised() (interface{}, error)

Sanitised returns a sanitised copy of the Benthos configuration, meaning fields of no consequence (unused inputs, outputs, processors etc) are excluded.

type Type

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

Type creates and manages the lifetime of a Benthos stream.

func New

func New(conf Config, mgr bundle.NewManagement, opts ...func(*Type)) (*Type, error)

New creates a new stream.Type.

func (*Type) IsReady

func (t *Type) IsReady() bool

IsReady returns a boolean indicating whether both the input and output layers of the stream are connected.

func (*Type) Stop

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

Stop attempts to close the stream within the specified timeout period. Initially the attempt is graceful, but as the timeout draws close the attempt becomes progressively less graceful.

func (*Type) StopGracefully

func (t *Type) StopGracefully(timeout time.Duration) (err error)

StopGracefully attempts to close the stream in the most graceful way by only closing the input layer and waiting for all other layers to terminate by proxy. This should guarantee that all in-flight and buffered data is resolved before shutting down.

func (*Type) StopOrdered

func (t *Type) StopOrdered(timeout time.Duration) (err error)

StopOrdered attempts to close all components of the stream in the order of positions within the stream, this allows data to flush all the way through the pipeline under certain circumstances but is less graceful than stopGracefully, which should be attempted first.

func (*Type) StopUnordered

func (t *Type) StopUnordered(timeout time.Duration) (err error)

StopUnordered attempts to close all components in parallel without allowing the stream to gracefully wind down in the order of component layers. This should only be attempted if both stopGracefully and stopOrdered failed.