consumer

Documentation

Index

• Variables
• type ConsumerBatchFn
• type ConsumerFn
• type DataSource
• type DeleteStrategy
• type SQS
  • func NewSQSConsumer(conf *SQSConf, svc sqsiface.SQSAPI) (*SQS, error)
  • func (s *SQS) Start(ctx context.Context, consumeFn ConsumerFn) error
  • func (s *SQS) StartBatched(ctx context.Context, batcher *batcher.Batcher, consumeFn ConsumerBatchFn) error
• type SQSConf

Variables

var (
	DefaultMaxNumberOfMessages = int64(10)
	DefaultWaitTimeSeconds     = int64(5)
	DefaultConcurrency         = 1
	DefaultDeleteStrategy      = DeleteStrategyOnSuccess
)

var (
	DeleteStrategyImmediate = DeleteStrategy("IMMEDIATE")
	DeleteStrategyOnSuccess = DeleteStrategy("ON_SUCCESS")
)

Types

type ConsumerBatchFn

type ConsumerBatchFn func(data [][]byte) error

type ConsumerFn

type ConsumerFn func(data []byte) error

type DataSource

type DataSource interface {
	Start(ctx context.Context, consumeFn ConsumerFn) error
}

type DeleteStrategy

type DeleteStrategy string

type SQS

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

func NewSQSConsumer

func NewSQSConsumer(conf *SQSConf, svc sqsiface.SQSAPI) (*SQS, error)

func (*SQS) Start

func (s *SQS) Start(ctx context.Context, consumeFn ConsumerFn) error

func (*SQS) StartBatched

func (s *SQS) StartBatched(ctx context.Context, batcher *batcher.Batcher, consumeFn ConsumerBatchFn) error

type SQSConf

type SQSConf struct {
	Queue               string
	Concurrency         int
	MaxNumberOfMessages int64
	VisibilityTimeout   int64
	WaitTimeSeconds     int64
	DeleteStrategy      DeleteStrategy
}