async

Documentation

Index

• func DetermineDecoder(contentType string) (encoding.DecodeRawFunc, error)
• type Builder
  • func New(name string, cf ConsumerFactory, proc ProcessorFunc) *Builder
  • func (cb *Builder) Create() (*Component, error)
  • func (cb *Builder) WithFailureStrategy(fs FailStrategy) *Builder
  • func (cb *Builder) WithRetries(retries uint) *Builder
  • func (cb *Builder) WithRetryWait(retryWait time.Duration) *Builder
• type Component
  • func (c *Component) Run(ctx context.Context) error
• type Consumer
• type ConsumerFactory
• type FailStrategy
• type Message
• type ProcessorFunc

Functions

func DetermineDecoder

func DetermineDecoder(contentType string) (encoding.DecodeRawFunc, error)

DetermineDecoder determines the decoder based on the content type.

Types

type Builder

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

Builder gathers all required properties in order to construct a component

func New

func New(name string, cf ConsumerFactory, proc ProcessorFunc) *Builder

New initializes a new builder for a component with the given name by default the failStrategy will be NackExitStrategy.

func (*Builder) Create

func (cb *Builder) Create() (*Component, error)

Create constructs the Component applying

func (*Builder) WithFailureStrategy

func (cb *Builder) WithFailureStrategy(fs FailStrategy) *Builder

WithFailureStrategy defines the failure strategy to be used default value is NackExitStrategy it will append an error to the builder if the strategy is not one of the pre-defined ones.

func (*Builder) WithRetries

func (cb *Builder) WithRetries(retries uint) *Builder

WithRetries specifies the retry events number for the component default value is '0'.

func (*Builder) WithRetryWait

func (cb *Builder) WithRetryWait(retryWait time.Duration) *Builder

WithRetryWait specifies the duration for the component to wait between retries default value is '0' it will append an error to the builder if the value is smaller than '0'.

type Component

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

Component implementation of a async component.

func (*Component) Run

func (c *Component) Run(ctx context.Context) error

Run starts the consumer processing loop messages.

type Consumer

type Consumer interface {
	Consume(context.Context) (<-chan Message, <-chan error, error)
	Close() error
}

Consumer interface which every specific consumer has to implement.

type ConsumerFactory

type ConsumerFactory interface {
	Create() (Consumer, error)
}

ConsumerFactory interface for creating consumers.

type FailStrategy

type FailStrategy int

FailStrategy type definition.

const (
	// NackExitStrategy does not acknowledge the message and exits the application on error.
	NackExitStrategy FailStrategy = iota
	// NackStrategy does not acknowledge the message, leaving it for reprocessing, and continues.
	NackStrategy
	// AckStrategy acknowledges message and continues.
	AckStrategy
)

type Message

type Message interface {
	Context() context.Context
	Decode(v interface{}) error
	Ack() error
	Nack() error
	Source() string
	Payload() []byte
}

Message interface for defining messages that are handled by the async component.

type ProcessorFunc

type ProcessorFunc func(Message) error

ProcessorFunc definition of a async processor.