manager

Documentation

Overview

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

Index

• Variables
• 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 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) HandleCRUD(w http.ResponseWriter, r *http.Request)
  • func (m *Type) HandleList(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 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 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) HandleCRUD

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

HandleCRUD is an http.HandleFunc for performing CRUD operations on streams.

func (*Type) HandleList

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

HandleList is an http.HandleFunc for returning maps of active benthos streams by their id, status and uptime.

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.