stream

Documentation

Index

• func OptOnClose(onClose func()) func(*Type)
• func Spec() docs.FieldSpecs
• type Config
  • func NewConfig() Config
  • func (c Config) Sanitised() (any, error)
• type Type
  • func New(conf Config, mgr bundle.NewManagement, opts ...func(*Type)) (*Type, error)
  • func (t *Type) IsReady() bool
  • func (t *Type) Stop(ctx context.Context) error
  • func (t *Type) StopGracefully(ctx context.Context) (err error)
  • func (t *Type) StopUnordered(ctx context.Context) (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() (any, 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(ctx context.Context) error

Stop attempts to close the stream within the specified timeout period. Initially the attempt is graceful, but if the context contains a deadline and it draws near the attempt becomes progressively less graceful.

If the context is cancelled an error is returned _after_ asynchronously instructing the remaining stream components to terminate ungracefully.

func (*Type) StopGracefully

func (t *Type) StopGracefully(ctx context.Context) (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) StopUnordered

func (t *Type) StopUnordered(ctx context.Context) (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.