buffer

Documentation

Overview

Package buffer links an input and an output together and buffers messages in between.

Index

• func Descriptions() string
• type Config
  • func NewConfig() Config
• type Empty
  • func (e *Empty) CloseAsync()
  • func (e *Empty) ErrorsChan() <-chan []error
  • func (e *Empty) MessageChan() <-chan types.Message
  • func (e *Empty) ResponseChan() <-chan types.Response
  • func (e *Empty) StartListening(responses <-chan types.Response) error
  • func (e *Empty) StartReceiving(msgs <-chan types.Message) error
  • func (e *Empty) WaitForClose(timeout time.Duration) error
• type OutputWrapper
  • func (m *OutputWrapper) CloseAsync()
  • func (m *OutputWrapper) ErrorsChan() <-chan []error
  • func (m *OutputWrapper) MessageChan() <-chan types.Message
  • func (m *OutputWrapper) ResponseChan() <-chan types.Response
  • func (m *OutputWrapper) StartListening(responses <-chan types.Response) error
  • func (m *OutputWrapper) StartReceiving(msgs <-chan types.Message) error
  • func (m *OutputWrapper) WaitForClose(timeout time.Duration) error
• type Type
  • func New(conf Config, log log.Modular, stats metrics.Type) (Type, error)
  • func NewEmpty(config Config, log log.Modular, stats metrics.Type) (Type, error)
  • func NewMemory(config Config, log log.Modular, stats metrics.Type) (Type, error)
  • func NewMmapFile(config Config, log log.Modular, stats metrics.Type) (Type, error)
  • func NewOutputWrapper(conf Config, buffer impl.Buffer, stats metrics.Type) Type

Functions

func Descriptions

func Descriptions() string

Descriptions returns a formatted string of collated descriptions of each type.

Types

type Config

type Config struct {
	Type            string                `json:"type" yaml:"type"`
	RetryThrottleMS int                   `json:"retry_throttle_ms" yaml:"retry_throttle_ms"`
	Mmap            impl.MmapBufferConfig `json:"mmap_file" yaml:"mmap_file"`
	Memory          impl.MemoryConfig     `json:"memory" yaml:"memory"`
	None            struct{}              `json:"none" yaml:"none"`
}

Config is the all encompassing configuration struct for all input types.

func NewConfig

func NewConfig() Config

NewConfig returns a configuration struct fully populated with default values.

type Empty

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

Empty is an empty buffer, simply forwards messages on directly.

func (*Empty) CloseAsync

func (e *Empty) CloseAsync()

CloseAsync shuts down the StackBuffer output and stops processing messages.

func (*Empty) ErrorsChan

func (e *Empty) ErrorsChan() <-chan []error

ErrorsChan returns the errors channel.

func (*Empty) MessageChan

func (e *Empty) MessageChan() <-chan types.Message

MessageChan returns the channel used for consuming messages from this input.

func (*Empty) ResponseChan

func (e *Empty) ResponseChan() <-chan types.Response

ResponseChan returns the response channel.

func (*Empty) StartListening

func (e *Empty) StartListening(responses <-chan types.Response) error

StartListening sets the channel for reading responses.

func (*Empty) StartReceiving

func (e *Empty) StartReceiving(msgs <-chan types.Message) error

StartReceiving assigns a messages channel for the output to read.

func (*Empty) WaitForClose

func (e *Empty) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the StackBuffer output has closed down.

type OutputWrapper

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

OutputWrapper wraps a buffer with a Producer/Consumer interface.

func (*OutputWrapper) CloseAsync

func (m *OutputWrapper) CloseAsync()

CloseAsync shuts down the OutputWrapper and stops processing messages.

func (*OutputWrapper) ErrorsChan

func (m *OutputWrapper) ErrorsChan() <-chan []error

ErrorsChan returns the errors channel.

func (*OutputWrapper) MessageChan

func (m *OutputWrapper) MessageChan() <-chan types.Message

MessageChan returns the channel used for consuming messages from this input.

func (*OutputWrapper) ResponseChan

func (m *OutputWrapper) ResponseChan() <-chan types.Response

ResponseChan returns the response channel.

func (*OutputWrapper) StartListening

func (m *OutputWrapper) StartListening(responses <-chan types.Response) error

StartListening sets the channel for reading responses.

func (*OutputWrapper) StartReceiving

func (m *OutputWrapper) StartReceiving(msgs <-chan types.Message) error

StartReceiving assigns a messages channel for the output to read.

func (*OutputWrapper) WaitForClose

func (m *OutputWrapper) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the OutputWrapper output has closed down.

type Type

type Type interface {
	types.Producer
	types.Consumer
	types.Closable

	// ErrorsChan returns the channel used for returning any accumulated errors.
	// This needs reading in the same select block where messages are sent as
	// the errors can occur at any time.
	ErrorsChan() <-chan []error
}

Type is the standard interface of an agent type.

func New

func New(conf Config, log log.Modular, stats metrics.Type) (Type, error)

New creates an input type based on an input configuration.

func NewEmpty

func NewEmpty(config Config, log log.Modular, stats metrics.Type) (Type, error)

NewEmpty creates a new buffer interface but doesn't buffer messages.

func NewMemory

func NewMemory(config Config, log log.Modular, stats metrics.Type) (Type, error)

NewMemory - Create a buffer held in memory.

func NewMmapFile

func NewMmapFile(config Config, log log.Modular, stats metrics.Type) (Type, error)

NewMmapFile creates a buffer held in memory and persisted to file through memory map.

func NewOutputWrapper

func NewOutputWrapper(
	conf Config,
	buffer impl.Buffer,
	stats metrics.Type,
) Type

NewOutputWrapper creates a new Producer/Consumer around a buffer.