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 MessageProcessor
• type Processor
  • func NewProcessor(msgProcessor MessageProcessor, log log.Modular, stats metrics.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 MessageProcessor

type MessageProcessor interface {
	// ProcessMessage returns a message to be sent onwards, if the bool flag is
	// false then the message should be dropped.
	ProcessMessage(msg *types.Message) (*types.Message, bool)
}

MessageProcessor is a type that reads a message, performs a processing operation, and returns a message and a flag indicating whether that message should be propagated or not.

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(
	msgProcessor MessageProcessor,
	log log.Modular,
	stats metrics.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.