async

Documentation

Index

• func DetermineDecoder(contentType string) (encoding.DecodeRawFunc, error)
• type Component
  • func New(name string, p ProcessorFunc, cf ConsumerFactory, oo ...OptionFunc) (*Component, error)
  • func (c *Component) Info() map[string]interface{}
  • func (c *Component) Run(ctx context.Context) error
• type Consumer
• type ConsumerFactory
• type FailStrategy
  • func (fs FailStrategy) String() string
• type Message
• type OptionFunc
  • func ConsumerRetry(retries int, retryWait time.Duration) OptionFunc
  • func FailureStrategy(fs FailStrategy) OptionFunc
• type ProcessorFunc

Functions

func DetermineDecoder

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

DetermineDecoder determines the decoder based on the content type.

Types

type Component

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

Component implementation of a async component.

func New

func New(name string, p ProcessorFunc, cf ConsumerFactory, oo ...OptionFunc) (*Component, error)

New returns a new async component. The default behavior is to return a error of failure. Use options to change the default behavior.

func (*Component) Info

func (c *Component) Info() map[string]interface{}

Info return information of the 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
	Info() map[string]interface{}
}

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 = 0
	// NackStrategy does not acknowledge the message, leaving it for reprocessing, and continues.
	NackStrategy FailStrategy = 1
	// AckStrategy acknowledges message and continues.
	AckStrategy FailStrategy = 2
)

func (FailStrategy) String

func (fs FailStrategy) String() string

type Message

type Message interface {
	Context() context.Context
	Decode(v interface{}) error
	Ack() error
	Nack() error
}

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

type OptionFunc

type OptionFunc func(*Component) error

OptionFunc definition for configuring the component in a functional way.

func ConsumerRetry

func ConsumerRetry(retries int, retryWait time.Duration) OptionFunc

ConsumerRetry set's the parameters for the retry policy of the consumer.

func FailureStrategy

func FailureStrategy(fs FailStrategy) OptionFunc

FailureStrategy option for setting the strategy of handling failures in the async component.

type ProcessorFunc

type ProcessorFunc func(Message) error

ProcessorFunc definition of a async processor.