processor

Documentation

Index

• func GetFail(part *message.Part) string
• func MarkErr(part *message.Part, span *tracing.Span, err error)
• type Pipeline
• type PipelineConstructorFunc
• type V1
  • func NewV2BatchedToV1Processor(typeStr string, p V2Batched, stats metrics.Type) V1
  • func NewV2ToV1Processor(typeStr string, p V2, stats metrics.Type) V1
• type V2
• type V2Batched

Functions

func GetFail

func GetFail(part *message.Part) string

GetFail returns an error string for a message part if it has failed, or an empty string if not.

func MarkErr

func MarkErr(part *message.Part, span *tracing.Span, err error)

MarkErr marks a message part as having failed. This includes modifying metadata to contain this error as well as adding the error to a tracing span if the message has one.

Types

type Pipeline

type Pipeline interface {
	// TransactionChan returns a channel used for consuming transactions from
	// this type. Every transaction received must be resolved before another
	// transaction will be sent.
	TransactionChan() <-chan message.Transaction

	// Consume starts the type receiving transactions from a Transactor.
	Consume(<-chan message.Transaction) error

	// CloseAsync triggers the shut down of this component but should not block
	// the calling goroutine.
	CloseAsync()

	// WaitForClose is a blocking call to wait until the component has finished
	// shutting down and cleaning up resources.
	WaitForClose(timeout time.Duration) error
}

Pipeline is an interface that implements channel based based consumer and producer methods for streaming data through a processing pipeline.

type PipelineConstructorFunc

type PipelineConstructorFunc func() (Pipeline, error)

PipelineConstructorFunc is a constructor to be called for each parallel stream pipeline thread in order to construct a custom pipeline implementation.

type V1

type V1 interface {
	// ProcessMessage attempts to process a message. This method returns both a
	// slice of messages or a response indicating whether messages were dropped
	// due to an intermittent error or were intentionally filtered.
	//
	// If an error occurs due to the contents of a message being invalid and you
	// wish to expose this as a recoverable fault you can use FlagErr to flag a
	// message as having failed without dropping it.
	//
	// More information about this form of error handling can be found at:
	// https://www.benthos.dev/docs/configuration/error_handling
	ProcessMessage(*message.Batch) ([]*message.Batch, error)

	// CloseAsync triggers the shut down of this component but should not block
	// the calling goroutine.
	CloseAsync()

	// WaitForClose is a blocking call to wait until the component has finished
	// shutting down and cleaning up resources.
	WaitForClose(timeout time.Duration) error
}

V1 is a common interface implemented by processors.

func NewV2BatchedToV1Processor

func NewV2BatchedToV1Processor(typeStr string, p V2Batched, stats metrics.Type) V1

NewV2BatchedToV1Processor wraps a processor.V2Batched with a struct that implements types.Processor.

func NewV2ToV1Processor

func NewV2ToV1Processor(typeStr string, p V2, stats metrics.Type) V1

NewV2ToV1Processor wraps a processor.V2 with a struct that implements V1.

type V2

type V2 interface {
	// Process a message into one or more resulting messages, or return an error
	// if the message could not be processed. If zero messages are returned and
	// the error is nil then the message is filtered.
	Process(ctx context.Context, p *message.Part) ([]*message.Part, error)

	// Close the component, blocks until either the underlying resources are
	// cleaned up or the context is cancelled. Returns an error if the context
	// is cancelled.
	Close(ctx context.Context) error
}

V2 is a simpler interface to implement than V1.

type V2Batched

type V2Batched interface {
	// Process a batch of messages into one or more resulting batches, or return
	// an error if the entire batch could not be processed. If zero messages are
	// returned and the error is nil then all messages are filtered.
	ProcessBatch(ctx context.Context, spans []*tracing.Span, b *message.Batch) ([]*message.Batch, error)

	// Close the component, blocks until either the underlying resources are
	// cleaned up or the context is cancelled. Returns an error if the context
	// is cancelled.
	Close(ctx context.Context) error
}

V2Batched is a simpler interface to implement than V1 and allows batch-wide processing.