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 Handler
• type Message
  • func (m *Message) Delete() (*sqs.DeleteMessageOutput, error)
• type MessageOutput
• type Middleware
  • func HandlerTimeout(t time.Duration) Middleware
  • func IgnoreEmptyResponses() Middleware
  • func Tracking() Middleware
• type Poller
  • func Default(sqsSvc *sqs.SQS) *Poller
  • func New(sqsSvc *sqs.SQS) *Poller
  • func (p *Poller) Handle(handler Handler, middleware ...Middleware)
  • func (p *Poller) ReceiveMessageParams(input *sqs.ReceiveMessageInput, opts ...request.Option)
  • 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 outerMiddleware has been used. The Poller returned by Default() comes with this outerMiddleware installed.

Variables

var (
	ErrNoHandler              = errors.New("ErrNoHandler: no handler set on poller instance")
	ErrNoReceiveMessageParams = errors.New("ErrNoReceiveMessageParams: no ReceiveMessage parameters have been set")
	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")
	ErrAlreadyShuttingDown    = errors.New("ErrAlreadyShuttingDown: poller is already in the process of shutting down")
	ErrAlreadyRunning         = errors.New("ErrAlreadyShuttingDown: poller is already running")
	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 outerMiddleware.

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

type Handler

type Handler func(ctx context.Context, client *sqs.SQS, 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 outerMiddleware that wraps the core Handler.

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

The sqs Client used to instantiate the poller will also be made available to allow the user to perform standard sqs operations.

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() (*sqs.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 HandlerTimeout

func HandlerTimeout(t time.Duration) Middleware

HandlerTimeout takes a timeout duration and returns ErrHandlerTimeout if the handler cannot process the message within that time. The user can then use other outerMiddleware to check for ErrHandlerTimeout and decide whether to exit or move onto the next poll request.

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) *Poller

Default creates a new instance of the SQS Poller from an instance of sqs.SQS. It also comes set up with the recommend outerMiddleware plugged in.

func New

func New(sqsSvc *sqs.SQS) *Poller

New creates a new instance of the SQS Poller from an instance of sqs.SQS.

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. The Middleware supplied to Handle will be applied first before any global middleware which are set by Use().

func (*Poller) ReceiveMessageParams

func (p *Poller) ReceiveMessageParams(input *sqs.ReceiveMessageInput, opts ...request.Option)

ReceiveMessageParams accepts the same parameters as the SQS ReceiveMessage method. It configures how the poller receives new messages, the parameters must be set before the Poller is run.

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, the Handler will return ErrHandlerTimeout. The error can be caught and handled by custom middleware, so the user can choose to move onto the next poll request if they so wish.

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 outerMiddleware to the Poller instance which will wrap any Handler and Handler specific outerMiddleware.