sqspoller

README

SQS-Poller

SQS-Poller is a simple queue polling framework, designed specifically to work with AWS SQS.

Documentation

Index

• Constants
• Variables
• type CtxTackingValue
• type DeleteMessageOutput
• type Handler
• type Message
  • func (m *Message) Delete() (*DeleteMessageOutput, error)
• type MessageOutput
• type Middleware
  • func IgnoreEmptyResponses() Middleware
  • func Tracking() Middleware
• type Poller
  • func Default(sqsSvc *sqs.SQS, config sqs.ReceiveMessageInput, options ...request.Option) *Poller
  • func New(sqsSvc *sqs.SQS, config sqs.ReceiveMessageInput, options ...request.Option) *Poller
  • func (p *Poller) Handle(handler Handler, middleware ...Middleware)
  • func (p *Poller) Run() error
  • func (p *Poller) SetHandlerTimeout(t time.Duration)
  • func (p *Poller) SetPollInterval(t time.Duration)
  • func (p *Poller) ShutdownAfter(t time.Duration) error
  • func (p *Poller) ShutdownGracefully() error
  • func (p *Poller) ShutdownNow() error
  • func (p *Poller) Use(middleware ...Middleware)

Constants

const CtxKey ctxKey = 1

CtxKey should be used to access the values on the context object of type *CtxTackingValue.

This can only be used if the Tracking middleware has been used. The Poller returned by Default() comes with this middleware installed.

Variables

var (
	ErrNoHandler = errors.New("ErrNoHandler: no handler set on Poller instance")

	ErrHandlerTimeout = errors.New("ErrHandlerTimeout: handler took to long to process message")

	ErrShutdownNow      = errors.New("ErrShutdownNow: poller was suddenly shutdown")
	ErrShutdownGraceful = errors.New("ErrShutdownGraceful: poller could not shutdown gracefully in time")

	ErrIntegrityIssue = errors.New("ErrIntegrityIssue: unknown integrity issue")
)

Types

type CtxTackingValue

type CtxTackingValue struct {
	TraceID string
	Now     time.Time
}

CtxTackingValue represents the values stored on the context object about the message response which is passed down through the handler function and middleware.

This can only be used if the Tracking middleware has been used. The Poller returned by Default() comes with this middleware installed.

type DeleteMessageOutput

type DeleteMessageOutput struct {
	*sqs.DeleteMessageOutput
}

DeleteMessageOutput wraps the sqs.DeleteMessageOutput output

type Handler

type Handler func(ctx context.Context, msgOutput *MessageOutput, err error) error

Handler is a function which handles the incoming SQS message.

When making Handlers to be used by the Poller, make sure the error value is checked first, before any business logic code, unless you have created an error checking middleware that wraps the core Handler.

If the error is non-nil, it will be of type *awserr.Error which is returned from a failed request for message from SQS.

type Message

type Message struct {
	*sqs.Message
	// contains filtered or unexported fields
}

Message is an individual message, contained within a MessageOutput, it provides methods to remove itself from the SQS queue.

func (*Message) Delete

func (m *Message) Delete() (*DeleteMessageOutput, error)

Delete removes the message from the queue, permanently.

type MessageOutput

type MessageOutput struct {
	*sqs.ReceiveMessageOutput
	Messages []*Message
	// contains filtered or unexported fields
}

MessageOutput is contains the SQS ReceiveMessageOutput and is passed down to the Handler when the Poller is running.

type Middleware

type Middleware func(Handler) Handler

Middleware is a function which that wraps a Handler to add functionality before or after the Handler code.

func IgnoreEmptyResponses

func IgnoreEmptyResponses() Middleware

IgnoreEmptyResponses stops the data from being passed down to the inner handler, if there is no message to be handled.

func Tracking

func Tracking() Middleware

Tracking adds tracking information to the context object for each message output received from the queue. The information can be accessed on the context object by using the CtxKey constant and returns a *CtxTackingValue object, containing a traceID and receive time.

type Poller

type Poller struct {

	// Time to wait for handler to process message, if handler function
	// takes longer than this to return, then the program is exited.
	HandlerTimeout time.Duration

	// Time interval between each poll request. After a poll request
	// has been made and response has been handled, the poller will
	// wait for this amount of time before making the next call.
	PollInterval time.Duration

	// Holds the time of the last poll request that was made. This can
	// be checked periodically, to confirm the Poller is running as expected.
	LastPollTime time.Time
	// contains filtered or unexported fields
}

Poller is an instance of the polling framework, it contains the SQS client and provides a simple API for polling an SQS queue.

func Default

func Default(sqsSvc *sqs.SQS, config sqs.ReceiveMessageInput, options ...request.Option) *Poller

Default creates a new instance of the SQS Poller from an instance of sqs.SQS and an sqs.ReceiveMessageInput, to configure how the SQS queue will be polled. It comes set up with the recommend middleware plugged in.

func New

func New(sqsSvc *sqs.SQS, config sqs.ReceiveMessageInput, options ...request.Option) *Poller

New creates a new instance of the SQS Poller from an instance of sqs.SQS and an sqs.ReceiveMessageInput, to configure how the SQS queue will be polled.

func (*Poller) Handle

func (p *Poller) Handle(handler Handler, middleware ...Middleware)

Handle attaches a Handler to the Poller instance, if a Handler already exists on the Poller instance, it will be replaced.

func (*Poller) Run

func (p *Poller) Run() error

Run starts the poller, the poller will continuously poll SQS until an error is returned, or explicitly told to shutdown.

func (*Poller) SetHandlerTimeout

func (p *Poller) SetHandlerTimeout(t time.Duration)

SetHandlerTimeout lets the user set the timeout for handling a message, if the handler function cannot finish execution within this time frame, then the Poller will exit and return ErrHandlerTimeout.

func (*Poller) SetPollInterval

func (p *Poller) SetPollInterval(t time.Duration)

SetPollInterval lets the user set the time interval between poll requests.

func (*Poller) ShutdownAfter

func (p *Poller) ShutdownAfter(t time.Duration) error

ShutdownAfter will attempt to shutdown gracefully, if graceful shutdown cannot be achieved within the given time frame, the Poller will exit, potentially leaking unhandled resources.

func (*Poller) ShutdownGracefully

func (p *Poller) ShutdownGracefully() error

ShutdownGracefully gracefully shuts down the poller.

func (*Poller) ShutdownNow

func (p *Poller) ShutdownNow() error

ShutdownNow shuts down the Poller instantly, potentially leaking unhandled resources.

func (*Poller) Use

func (p *Poller) Use(middleware ...Middleware)

Use attaches global middleware to the Poller instance which will wrap any Handler and Handler specific middleware.