buffer

Documentation

Overview

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

Index

• Variables
• func Descriptions() string
• func SanitiseConfig(conf Config) (interface{}, error)
• type Config
  • func NewConfig() Config
• type Empty
  • func (e *Empty) CloseAsync()
  • func (e *Empty) ErrorsChan() <-chan []error
  • func (e *Empty) StartReceiving(msgs <-chan types.Transaction) error
  • func (e *Empty) StopConsuming()
  • func (e *Empty) TransactionChan() <-chan types.Transaction
  • func (e *Empty) WaitForClose(timeout time.Duration) error
• type Parallel
• type ParallelWrapper
  • func (m *ParallelWrapper) CloseAsync()
  • func (m *ParallelWrapper) StartReceiving(msgs <-chan types.Transaction) error
  • func (m *ParallelWrapper) StopConsuming()
  • func (m *ParallelWrapper) TransactionChan() <-chan types.Transaction
  • func (m *ParallelWrapper) WaitForClose(timeout time.Duration) error
• type Single
• type SingleWrapper
  • func (m *SingleWrapper) CloseAsync()
  • func (m *SingleWrapper) StartReceiving(msgs <-chan types.Transaction) error
  • func (m *SingleWrapper) StopConsuming()
  • func (m *SingleWrapper) TransactionChan() <-chan types.Transaction
  • func (m *SingleWrapper) 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 NewParallelWrapper(conf Config, buffer Parallel, log log.Modular, stats metrics.Type) Type
  • func NewSingleWrapper(conf Config, buffer Single, log log.Modular, stats metrics.Type) Type
• type TypeSpec

Variables

var Constructors = map[string]TypeSpec{}

Constructors is a map of all buffer types with their specs.

Functions

func Descriptions

func Descriptions() string

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

func SanitiseConfig

func SanitiseConfig(conf Config) (interface{}, error)

SanitiseConfig returns a sanitised version of the Config, meaning sections that aren't relevant to behaviour are removed.

Types

type Config

type Config struct {
	Type   string                  `json:"type" yaml:"type"`
	Memory single.MemoryConfig     `json:"memory" yaml:"memory"`
	Mmap   single.MmapBufferConfig `json:"mmap_file" yaml:"mmap_file"`
	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) StartReceiving

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

StartReceiving assigns a messages channel for the output to read.

func (*Empty) StopConsuming

func (e *Empty) StopConsuming()

StopConsuming instructs the buffer to no longer consume data.

func (*Empty) TransactionChan

func (e *Empty) TransactionChan() <-chan types.Transaction

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

func (*Empty) WaitForClose

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

WaitForClose blocks until the StackBuffer output has closed down.

type Parallel

type Parallel interface {
	// NextMessage reads the next oldest message, the message is preserved until
	// the returned AckFunc is called.
	NextMessage() (types.Message, parallel.AckFunc, error)

	// PushMessage adds a new message to the stack. Returns the backlog in
	// bytes.
	PushMessage(types.Message) (int, error)

	// CloseOnceEmpty closes the Buffer once the buffer has been emptied, this
	// is a way for a writer to signal to a reader that it is finished writing
	// messages, and therefore the reader can close once it is caught up. This
	// call blocks until the close is completed.
	CloseOnceEmpty()

	// Close closes the Buffer so that blocked readers or writers become
	// unblocked.
	Close()
}

Parallel represents a method of buffering sequential messages, supporting only a single consumer.

type ParallelWrapper

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

ParallelWrapper wraps a buffer with a Producer/Consumer interface.

func (*ParallelWrapper) CloseAsync

func (m *ParallelWrapper) CloseAsync()

CloseAsync shuts down the ParallelWrapper and stops processing messages.

func (*ParallelWrapper) StartReceiving

func (m *ParallelWrapper) StartReceiving(msgs <-chan types.Transaction) error

StartReceiving assigns a messages channel for the output to read.

func (*ParallelWrapper) StopConsuming

func (m *ParallelWrapper) StopConsuming()

StopConsuming instructs the buffer to stop consuming messages and close once the buffer is empty.

func (*ParallelWrapper) TransactionChan

func (m *ParallelWrapper) TransactionChan() <-chan types.Transaction

TransactionChan returns the channel used for consuming messages from this buffer.

func (*ParallelWrapper) WaitForClose

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

WaitForClose blocks until the ParallelWrapper output has closed down.

type Single

type Single interface {
	// ShiftMessage removes the oldest message from the stack. Returns the
	// backlog in bytes.
	ShiftMessage() (int, error)

	// NextMessage reads the oldest message, the message is preserved until
	// ShiftMessage is called.
	NextMessage() (types.Message, error)

	// PushMessage adds a new message to the stack. Returns the backlog in
	// bytes.
	PushMessage(types.Message) (int, error)

	// CloseOnceEmpty closes the Buffer once the buffer has been emptied, this
	// is a way for a writer to signal to a reader that it is finished writing
	// messages, and therefore the reader can close once it is caught up. This
	// call blocks until the close is completed.
	CloseOnceEmpty()

	// Close closes the Buffer so that blocked readers or writers become
	// unblocked.
	Close()
}

Single represents a method of buffering sequential messages, supporting only a single consumer.

type SingleWrapper

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

SingleWrapper wraps a buffer with a Producer/Consumer interface.

func (*SingleWrapper) CloseAsync

func (m *SingleWrapper) CloseAsync()

CloseAsync shuts down the SingleWrapper and stops processing messages.

func (*SingleWrapper) StartReceiving

func (m *SingleWrapper) StartReceiving(msgs <-chan types.Transaction) error

StartReceiving assigns a messages channel for the output to read.

func (*SingleWrapper) StopConsuming

func (m *SingleWrapper) StopConsuming()

StopConsuming instructs the buffer to stop consuming messages and close once the buffer is empty.

func (*SingleWrapper) TransactionChan

func (m *SingleWrapper) TransactionChan() <-chan types.Transaction

TransactionChan returns the channel used for consuming messages from this buffer.

func (*SingleWrapper) WaitForClose

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

WaitForClose blocks until the SingleWrapper output has closed down.

type Type

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

	// StopConsuming instructs the buffer to cut off the input. It will then
	// enter a mode whereby messages can only be read, and when the buffer is
	// empty it will shut down.
	StopConsuming()
}

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 NewParallelWrapper

func NewParallelWrapper(
	conf Config,
	buffer Parallel,
	log log.Modular,
	stats metrics.Type,
) Type

NewParallelWrapper creates a new Producer/Consumer around a buffer.

func NewSingleWrapper

func NewSingleWrapper(
	conf Config,
	buffer Single,
	log log.Modular,
	stats metrics.Type,
) Type

NewSingleWrapper creates a new Producer/Consumer around a buffer.

type TypeSpec

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

TypeSpec is a constructor and usage description for each buffer type.