pipeline

package
Version: v1.20.4 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: May 13, 2019 License: MIT Imports: 11 Imported by: 6

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

Constants

This section is empty.

Variables

This section is empty.

Functions

func SanitiseConfig added in v0.10.4

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 added in v0.10.4

type Config struct {
	Threads    int                `json:"threads" yaml:"threads"`
	Processors []processor.Config `json:"processors" yaml:"processors"`
}

Config is a configuration struct for creating parallel processing pipelines. The number of resuling parallel processing pipelines will match the number of threads specified. Processors are executed on each message in the order that they are written.

In order to fully utilise each processing thread you must either have a number of parallel inputs that matches or surpasses the number of pipeline threads, or use a memory buffer.

func NewConfig added in v0.10.4

func NewConfig() Config

NewConfig returns a configuration struct fully populated with default values.

type Pool added in v0.4.6

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 added in v0.4.6

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 added in v0.4.6

func (p *Pool) CloseAsync()

CloseAsync shuts down the pipeline and stops processing messages.

func (*Pool) Consume added in v0.19.0

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

Consume assigns a messages channel for the pipeline to read.

func (*Pool) TransactionChan added in v0.9.0

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

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

func (*Pool) WaitForClose added in v0.4.6

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 added in v0.19.0

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

Consume assigns a messages channel for the pipeline to read.

func (*Processor) TransactionChan added in v0.9.0

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 added in v0.10.4

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.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL