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

• type ConstructorFunc
• type Pool
  • func NewPool(constructor ConstructorFunc, workers int, log log.Modular, stats metrics.Type) (*Pool, error)
  • func (p *Pool) CloseAsync()
  • func (p *Pool) MessageChan() <-chan types.Message
  • func (p *Pool) ResponseChan() <-chan types.Response
  • func (p *Pool) StartListening(responses <-chan types.Response) error
  • func (p *Pool) StartReceiving(msgs <-chan types.Message) error
  • func (p *Pool) WaitForClose(timeout time.Duration) error
• type Processor
  • func NewProcessor(log log.Modular, stats metrics.Type, msgProcessors ...processor.Type) *Processor
  • func (p *Processor) CloseAsync()
  • func (p *Processor) MessageChan() <-chan types.Message
  • func (p *Processor) ResponseChan() <-chan types.Response
  • func (p *Processor) StartListening(responses <-chan types.Response) error
  • func (p *Processor) StartReceiving(msgs <-chan types.Message) error
  • func (p *Processor) WaitForClose(timeout time.Duration) error
• type Type

Types

type ConstructorFunc

type ConstructorFunc func() (Type, error)

ConstructorFunc is a common type for constructing a pipeline type.

type Pool

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

Pool is a pool of pipelines. It reads from a single source and writes to a single source. The input is decoupled which means failed delivery notification cannot be propagated back up to the original input.

If delivery acknowledgements to the input is required this pool should not be used. Instead, you should configure multiple inputs each with their own pipeline e.g. configure 8 kafka_balanced inputs each with a single processor rather than a single kafka_balanced with a pool of 8 workers.

func NewPool

func NewPool(
	constructor ConstructorFunc,
	workers int,
	log log.Modular,
	stats metrics.Type,
) (*Pool, error)

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

func (*Pool) CloseAsync

func (p *Pool) CloseAsync()

CloseAsync shuts down the pipeline and stops processing messages.

func (*Pool) MessageChan

func (p *Pool) MessageChan() <-chan types.Message

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

func (*Pool) ResponseChan

func (p *Pool) ResponseChan() <-chan types.Response

ResponseChan returns the response channel from this pipeline.

func (*Pool) StartListening

func (p *Pool) StartListening(responses <-chan types.Response) error

StartListening sets the channel that this pipeline will read responses from.

func (*Pool) StartReceiving

func (p *Pool) StartReceiving(msgs <-chan types.Message) error

StartReceiving assigns a messages channel for the pipeline to read.

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 ...processor.Type,
) *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) MessageChan

func (p *Processor) MessageChan() <-chan types.Message

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

func (*Processor) ResponseChan

func (p *Processor) ResponseChan() <-chan types.Response

ResponseChan returns the response channel from this pipeline.

func (*Processor) StartListening

func (p *Processor) StartListening(responses <-chan types.Response) error

StartListening sets the channel that this pipeline will read responses from.

func (*Processor) StartReceiving

func (p *Processor) StartReceiving(msgs <-chan types.Message) error

StartReceiving assigns a messages channel for the pipeline to read.

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.Producer
	types.Consumer
	types.Closable
}

Type is an interface that all pipeline types should implement.