gosqs

README

go-sqs

A simplistic library to listen for and produce messages to Amazons Simple Queuing Service (SQS).

Quick start

Get started with go-sqs via

go get github.com/engelmi/go-sqs

Features

• Provides an easy to use producer interface to put messages into an SQS queue
  • Can be configured via a configuration struct
• Provides an easy to use consumer interface to listen for messages in an SQS queue
  • Can be configured via a configuration struct, e.g. with poll timeout
  • A handler function is used to process messages
  • Errors returned from the handler function - as well as any occuring panic - result in the message not being acknowledged
  • Messages that are not acknowledged are not deleted from the queue and, therefore, consumed again
  • The listening for messages can be started as a goroutine
  • A wait group can be used to have the currently executed handler finish before terminating the program

Usage

Examples on how to use go-sqs are provided in github.com/engelmi/go-sqs/examples.

The credentials file ~/.aws/credentials containing dummy values for the access key and secret key is required to run the examples:

[default]
aws_access_key_id = foo
aws_secret_access_key = bar

Alternatively, provide both values via environment variables:

AWS_ACCESS_KEY_ID='foo' AWS_SECRET_ACCESS_KEY='bar' go run <example-file>.go

Documentation

Index

• func MarshalToJson(payload interface{}) ([]byte, error)
• type Consumer
  • func NewConsumer(config ConsumerConfig, handlerFunc MessageHandler) (Consumer, error)
• type ConsumerConfig
• type IncomingMessage
• type IncomingMessageAttributeValue
• type MessageHandler
• type OutgoingMessage
• type Producer
  • func NewProducer(config ProducerConfig) (Producer, error)
• type ProducerConfig
• type QueueConfig

Functions

func MarshalToJson

func MarshalToJson(payload interface{}) ([]byte, error)

Types

type Consumer

type Consumer interface {
	StartListening(ctx context.Context, wg *sync.WaitGroup)
	StopListening() error
}

func NewConsumer

func NewConsumer(config ConsumerConfig, handlerFunc MessageHandler) (Consumer, error)

type ConsumerConfig

type ConsumerConfig struct {
	QueueConfig
	PollTimeout         time.Duration
	AckTimeout          time.Duration
	MaxNumberOfMessages int64
	Logger              logrus.Logger
}

type IncomingMessage

type IncomingMessage struct {
	MessageId              *string
	ReceiptHandle          *string
	Body                   *string
	MD5OfBody              *string
	MessageAttributes      map[string]*IncomingMessageAttributeValue
	MD5OfMessageAttributes *string
	Attributes             map[string]*string
}

type IncomingMessageAttributeValue

type IncomingMessageAttributeValue struct {
	BinaryValue []byte
	DataType    *string
	StringValue *string
}

type MessageHandler

type MessageHandler func(ctx context.Context, msg IncomingMessage) error

type OutgoingMessage

type OutgoingMessage struct {
	DeduplicationId *string
	GroupId         *string
	Payload         []byte
	Attributes      map[string]string
}

type Producer

type Producer interface {
	Send(ctx context.Context, msg OutgoingMessage) (*string, error)
}

func NewProducer

func NewProducer(config ProducerConfig) (Producer, error)

type ProducerConfig

type ProducerConfig struct {
	QueueConfig
	Timeout time.Duration
}

type QueueConfig

type QueueConfig struct {
	Region   string
	Endpoint string
	Queue    string
}