pipeline

Documentation

Overview

Package pipeline contains structures that implement both the Producer and Consumer interfaces. They can be used as extra pipeline components for various utilities.

Index

• func SanitiseConfig(conf Config) (interface{}, error)
• type Config
  • func NewConfig() Config
• type Pool
  • func NewPool(constructor types.PipelineConstructorFunc, threads int, log log.Modular, ...) (*Pool, error)
  • func (p *Pool) CloseAsync()
  • func (p *Pool) Consume(msgs <-chan types.Transaction) error
  • func (p *Pool) TransactionChan() <-chan types.Transaction
  • func (p *Pool) WaitForClose(timeout time.Duration) error
• type Processor
  • func NewProcessor(log log.Modular, stats metrics.Type, msgProcessors ...types.Processor) *Processor
  • func (p *Processor) CloseAsync()
  • func (p *Processor) Consume(msgs <-chan types.Transaction) error
  • func (p *Processor) TransactionChan() <-chan types.Transaction
  • func (p *Processor) WaitForClose(timeout time.Duration) error
• type Type
  • func New(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type, ...) (Type, error)

Functions

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 {
	Threads    int                `json:"threads" yaml:"threads"`
	Processors []processor.Config `json:"processors" yaml:"processors"`
}

Config is a configuration struct for a pipeline.

func NewConfig

func NewConfig() Config

NewConfig returns a configuration struct fully populated with default values.

type Pool

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

Pool is a pool of pipelines. Each pipeline reads from a shared transaction channel. Inputs remain coupled to their outputs as they propagate the response channel in the transaction.

func NewPool

func NewPool(
	constructor types.PipelineConstructorFunc,
	threads int,
	log log.Modular,
	stats metrics.Type,
) (*Pool, error)

NewPool returns a new pipeline pool that utilises multiple processor threads.

func (*Pool) CloseAsync

func (p *Pool) CloseAsync()

CloseAsync shuts down the pipeline and stops processing messages.

func (*Pool) Consume

func (p *Pool) Consume(msgs <-chan types.Transaction) error

Consume assigns a messages channel for the pipeline to read.

func (*Pool) TransactionChan

func (p *Pool) TransactionChan() <-chan types.Transaction

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

func (*Pool) WaitForClose

func (p *Pool) WaitForClose(timeout time.Duration) error

WaitForClose - Blocks until the StackBuffer output has closed down.

type Processor

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

Processor is a pipeline that supports both Consumer and Producer interfaces. The processor will read from a source, perform some processing, and then either propagate a new message or drop it.

func NewProcessor

func NewProcessor(
	log log.Modular,
	stats metrics.Type,
	msgProcessors ...types.Processor,
) *Processor

NewProcessor returns a new message processing pipeline.

func (*Processor) CloseAsync

func (p *Processor) CloseAsync()

CloseAsync shuts down the pipeline and stops processing messages.

func (*Processor) Consume

func (p *Processor) Consume(msgs <-chan types.Transaction) error

Consume assigns a messages channel for the pipeline to read.

func (*Processor) TransactionChan

func (p *Processor) TransactionChan() <-chan types.Transaction

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

func (*Processor) WaitForClose

func (p *Processor) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the StackBuffer output has closed down.

type Type

type Type interface {
	types.Pipeline
}

Type is implemented by all pipeline implementations.

func New

func New(
	conf Config,
	mgr types.Manager,
	log log.Modular,
	stats metrics.Type,
	processorCtors ...types.ProcessorConstructorFunc,
) (Type, error)

New creates an input type based on an input configuration.