worker

Documentation

Index

• type Config
• type Handler
• type HandlerFunc
  • func (f HandlerFunc) HandleMessage(msg *types.Message) error
• type InvalidEventError
  • func NewInvalidEventError(event, msg string) InvalidEventError
  • func (e InvalidEventError) Error() string
• type QueueAPI
• type QueueDeleteReceiverAPI
  • func CreateSqsClient(ctx context.Context, region, sqsEndpoint string) (QueueDeleteReceiverAPI, error)
• type Worker
  • func New(ctx context.Context, client QueueAPI, config *Config) *Worker
  • func (worker *Worker) Start(ctx context.Context, h Handler)

Types

type Config

type Config struct {
	MaxNumberOfMessage int32
	QueueName          string
	QueueURL           string
	WaitTimeSecond     int32
}

Config struct

type Handler

type Handler interface {
	HandleMessage(msg *types.Message) error
}

Handler interface

type HandlerFunc

type HandlerFunc func(msg *types.Message) error

HandlerFunc is used to define the Handler that is run on for each message

func (HandlerFunc) HandleMessage

func (f HandlerFunc) HandleMessage(msg *types.Message) error

HandleMessage wraps a function for handling sqs messages

type InvalidEventError

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

InvalidEventError struct

func NewInvalidEventError

func NewInvalidEventError(event, msg string) InvalidEventError

NewInvalidEventError creates InvalidEventError struct

func (InvalidEventError) Error

func (e InvalidEventError) Error() string

type QueueAPI

type QueueAPI interface {
	GetQueueUrl(ctx context.Context, params *sqs.GetQueueUrlInput, optFns ...func(*sqs.Options)) (*sqs.GetQueueUrlOutput, error)
	QueueDeleteReceiverAPI
}

QueueAPI interface is the minimum interface required from a queue implementation to invoke New worker. Invoking worker.New() takes in a queue name which is why GetQueueUrl is needed.

type QueueDeleteReceiverAPI

type QueueDeleteReceiverAPI interface {
	DeleteMessage(ctx context.Context, params *sqs.DeleteMessageInput, optFns ...func(*sqs.Options)) (*sqs.DeleteMessageOutput, error)
	ReceiveMessage(ctx context.Context, params *sqs.ReceiveMessageInput, optFns ...func(*sqs.Options)) (*sqs.ReceiveMessageOutput, error)
}

QueueDeleteReceiverAPI interface is the minimum interface required to run a worker. When a worker is in its Receive loop, it requires this interface.

func CreateSqsClient

func CreateSqsClient(ctx context.Context, region, sqsEndpoint string) (QueueDeleteReceiverAPI, error)

type Worker

type Worker struct {
	Config    *Config
	Log       logging.Logger
	SqsClient QueueDeleteReceiverAPI
}

Worker struct

func New

func New(ctx context.Context, client QueueAPI, config *Config) *Worker

New sets up a new Worker

func (*Worker) Start

func (worker *Worker) Start(ctx context.Context, h Handler)

Start starts the polling and will continue polling till the application is forcibly stopped