sqspoller

README

SQS-Poller

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

Documentation

Index

• Constants
• Variables
• type CtxValue
• type DeleteMessageOutput
• type Handler
• type Message
  • func (m *Message) Delete() (*DeleteMessageOutput, error)
• type MessageOutput
• type Middleware
  • func IgnoreEmptyResponses() 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) SetInterval(t time.Duration)
  • func (p *Poller) SetTimeoutHandling(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 is the package's context key used to store values on context.Context object to avoid clashing with other packages.

Variables

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

	ErrTimeoutNoMessages = errors.New("ErrTimeoutNoMessages: no new messages in given time frame")
	ErrTimeoutHandling   = errors.New("ErrTimeoutHandling: handler took to long to process message")
	ErrTimeoutShutdown   = errors.New("ErrTimeoutShutdown: could not shut down gracefully")

	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 CtxValue

type CtxValue struct {
	TraceID string
	Now     time.Time
}

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

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 function which handles the incoming SQS message.

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.

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.
	TimeoutHandling 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.
	Interval time.Duration
	// 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) SetInterval

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

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

func (*Poller) SetTimeoutHandling

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

SetTimeoutHandling 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.