sqs

Documentation

Overview

Package sqs provides a native consumer for AWS SQS.

Index

• type API
• type Batch
• type Component
  • func New(name, queueName string, sqsAPI API, proc ProcessorFunc, oo ...OptionFunc) (*Component, error)
  • func (c *Component) Run(ctx context.Context) error
• type Message
• type OptionFunc
  • func MaxMessages(maxMessages int32) OptionFunc
  • func PollWaitSeconds(pollWaitSeconds int32) OptionFunc
  • func QueueOwner(queueOwner string) OptionFunc
  • func QueueStatsInterval(interval time.Duration) OptionFunc
  • func Retries(count uint) OptionFunc
  • func RetryWait(interval time.Duration) OptionFunc
  • func VisibilityTimeout(visibilityTimeout int32) OptionFunc
• type ProcessorFunc

Types

type API

type API interface {
	CreateQueue(ctx context.Context, params *sqs.CreateQueueInput, optFns ...func(*sqs.Options)) (*sqs.CreateQueueOutput, error)
	GetQueueUrl(ctx context.Context, params *sqs.GetQueueUrlInput, optFns ...func(*sqs.Options)) (*sqs.GetQueueUrlOutput, error)
	GetQueueAttributes(ctx context.Context, params *sqs.GetQueueAttributesInput, optFns ...func(*sqs.Options)) (*sqs.GetQueueAttributesOutput, error)
	DeleteMessageBatch(ctx context.Context, params *sqs.DeleteMessageBatchInput, optFns ...func(*sqs.Options)) (*sqs.DeleteMessageBatchOutput, error)
	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)
}

type Batch

type Batch interface {
	// Messages of the batch.
	Messages() []Message
	// ACK deletes all messages from SQS with a single call and completes the all the message tracing spans.
	// In case the action will not manage to ACK all the messages, a slice of the failed messages will be returned.
	ACK() ([]Message, error)
	// NACK leaves all messages in the queue and completes the all the message tracing spans.
	NACK()
}

Batch interface for multiple AWS SQS messages.

type Component

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

Component implementation of an async component.

func New

func New(name, queueName string, sqsAPI API, proc ProcessorFunc, oo ...OptionFunc) (*Component, error)

New creates a new component with support for functional configuration.

func (*Component) Run

func (c *Component) Run(ctx context.Context) error

Run starts the consumer processing loop messages.

type Message

type Message interface {
	// Context will contain the context to be used for processing.
	// Each context will have a logger setup which can be used to create a logger from context.
	Context() context.Context
	// ID of the message.
	ID() string
	// Body of the message.
	Body() []byte
	// Message will contain the raw SQS message.
	Message() types.Message
	// Span contains the tracing span of this message.
	Span() opentracing.Span
	// ACK deletes the message from the queue and completes the tracing span.
	ACK() error
	// NACK leaves the message in the queue and completes the tracing span.
	NACK()
}

Message interface for AWS SQS message.

type OptionFunc

type OptionFunc func(*Component) error

OptionFunc definition for configuring the component in a functional way.

func MaxMessages

func MaxMessages(maxMessages int32) OptionFunc

MaxMessages option for setting the max number of messages fetched. Allowed values are between 1 and 10. If messages can be processed very quickly, maxing out this value is fine, otherwise having a high value is risky as it might trigger the visibility timeout. Having a value too small isn't recommended either, as it increases the number of SQS API requests, thus AWS costs.

func PollWaitSeconds

func PollWaitSeconds(pollWaitSeconds int32) OptionFunc

PollWaitSeconds sets the wait time for the long polling mechanism in seconds. Allowed values are between 0 and 20. 0 enables short polling.

func QueueOwner

func QueueOwner(queueOwner string) OptionFunc

QueueOwner sets the AWS queue owner ID for an SQS queue.

func QueueStatsInterval

func QueueStatsInterval(interval time.Duration) OptionFunc

QueueStatsInterval sets the interval at which we retrieve AWS SQS stats.

func Retries

func Retries(count uint) OptionFunc

Retries sets the error retries of the component.

func RetryWait

func RetryWait(interval time.Duration) OptionFunc

RetryWait sets the wait period for the component retry.

func VisibilityTimeout

func VisibilityTimeout(visibilityTimeout int32) OptionFunc

VisibilityTimeout sets the time a message is invisible after it has been requested. This is a built-in resiliency mechanism so that, should the consumer fail to acknowledge the message within such timeout, it will become visible again and thus available for retries. Allowed values are between 0 and 12 hours in seconds.

type ProcessorFunc

type ProcessorFunc func(context.Context, Batch)

ProcessorFunc definition of an async processor.