loafergo

README

Loafer Go

Loafer Go is a lightweight Go library designed for high-throughput and concurrent processing of messages from AWS SQS queues and sending to AWS SNS topics.

---

✨ Features

• ✅ Concurrent Message Consumers with fixed worker pool size
• ✅ FIFO Grouped Processing
  • Based MessageGroupId and custom fields (loafergo.PerGroupID)
  • Parallel (loafergo.Parallel)
• ✅ SNS Producer with support for both standard and FIFO topics
• ✅ SQS Batch Receive and Parallel Handling
• ✅ Simple API with clean abstractions and interfaces
• ✅ Test Coverage & Benchmarks
• ✅ Fully Configurable via functional options

---

📦 Installation

go get -u github.com/justcodes/loafer-go/v2

Import into your project:

import "github.com/justcodes/loafer-go/v2"

---

🚀 Quickstart Example

Start by writing a main application that produces messages to SNS and consumes from SQS.

example

---

🐳 Local Development (with LocalStack)

Make sure you have Docker installed.

docker compose up -d

The init script in ./aws/init-aws.sh will:

• Create topics (standard and fifo)
• Create queues
• Subscribe queues to the topics

---

🧪 Testing

Run tests:

make test

Run benchmarks:

make test-bench

Install formatters and linters:

make configure

---

📁 Project Structure

• loafergo/ – Main package code
• aws/ – AWS configuration, SQS/SNS clients, and route handlers
• example/ – Sample producer and consumer demonstrating loafergo usage
• fake/ – Fakes for tests

---

🧪 Benchmark (SNS & SQS)

BenchmarkParserJSONToAnotherJSON_Small-12       12374347               474.9 ns/op           272 B/op          8 allocs/op
BenchmarkParserJSONToAnotherJSON_Medium-12       3689594              1573 ns/op             552 B/op         10 allocs/op
BenchmarkParserJSONToAnotherJSON_Large-12        1672772              3476 ns/op            1144 B/op          7 allocs/op
BenchmarkStructToMap-12                          4388439              1468 ns/op            1160 B/op          8 allocs/op
BenchmarkStructToMapConcurrent-12                4082488              1512 ns/op            1160 B/op          8 allocs/op
BenchmarkStructToMapWithCache-12                 6046621              1025 ns/op            1160 B/op          8 allocs/op

---

📌 Makefile Tasks

make update-dependencies     # Update Go dependencies
make format                  # Run goimports
make lint                    # Run golangci-lint
make install-golang-ci       # Install GolangCI-Lint
make install-goimports       # Install GoImports
make clean                   # Clean test cache
make test                    # Run tests with coverage
make test-bench              # Run benchmarks

---

🔚 Acknowledgments

Inspired by:

• loafer
• gosqs

Documentation

Index

• Variables
• type Config
• type Error
  • func (e Error) Context(err error) Error
  • func (e Error) Error() string
• type Handler
• type Logger
• type LoggerFunc
  • func (f LoggerFunc) Log(args ...interface{})
• type Manager
  • func NewManager(config *Config) *Manager
  • func (m *Manager) GetRoutes() []Router
  • func (m *Manager) RegisterRoute(route Router)
  • func (m *Manager) RegisterRoutes(routes []Router)
  • func (m *Manager) Run(ctx context.Context) error
• type Message
• type Mode
• type NoOpLogger
  • func (NoOpLogger) Log(args ...any)
• type Router
• type SNSClient
• type SQSClient

Variables

var (
	ErrNoRoute            = Error{/* contains filtered or unexported fields */}
	ErrGetMessage         = Error{/* contains filtered or unexported fields */}
	ErrInvalidCreds       = Error{/* contains filtered or unexported fields */}
	ErrMarshal            = Error{/* contains filtered or unexported fields */}
	ErrNoSQSClient        = Error{/* contains filtered or unexported fields */}
	ErrNoHandler          = Error{/* contains filtered or unexported fields */}
	ErrEmptyParam         = Error{/* contains filtered or unexported fields */}
	ErrEmptyRequiredField = Error{/* contains filtered or unexported fields */}
	ErrEmptyInput         = Error{/* contains filtered or unexported fields */}
)

Predefined errors.

Types

type Config

type Config struct {
	Logger       Logger
	RetryTimeout time.Duration
}

Config defines settings shared across the manager and routes.

type Error

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

Error represents a typed application error with context.

func (Error) Context

func (e Error) Context(err error) Error

Context wraps a base error with context.

func (Error) Error

func (e Error) Error() string

Error returns the composed error message.

type Handler

type Handler func(context.Context, Message) error

Handler represents the handler function

type Logger

type Logger interface {
	Log(args ...any)
}

A Logger is a minimalistic interface for the loafer to log messages to. Should be used to provide custom logging writers for the loafer to use.

type LoggerFunc

type LoggerFunc func(...interface{})

A LoggerFunc is a convenience type to convert a function taking a variadic list of arguments and wrap it so the Logger interface can be used.

Example:

loafergo.NewManager(context.Background(), loafergo.Config{Logger: loafergo.LoggerFunc(func(args ...interface{}) {
    fmt.Fprintln(os.Stdout, args...)
})})

func (LoggerFunc) Log

func (f LoggerFunc) Log(args ...interface{})

Log calls the wrapped function with the arguments provided

type Manager

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

Manager coordinates multiple routes and startWorker pools.

func NewManager

func NewManager(config *Config) *Manager

NewManager creates a new Manager with the provided configuration.

func (*Manager) GetRoutes

func (m *Manager) GetRoutes() []Router

GetRoutes returns all registered routes.

func (*Manager) RegisterRoute

func (m *Manager) RegisterRoute(route Router)

RegisterRoute adds a single route to the manager.

func (*Manager) RegisterRoutes

func (m *Manager) RegisterRoutes(routes []Router)

RegisterRoutes adds multiple routes to the manager.

func (*Manager) Run

func (m *Manager) Run(ctx context.Context) error

Run the Manager distributing the startWorker pool by the number of routes. Returns an error if no routes are registered.

type Message

type Message interface {
	// Decode will unmarshal the body message into a supplied output using JSON
	Decode(out interface{}) error
	// Attribute will return the custom attribute sent throughout the request.
	Attribute(key string) string
	// Attributes will return the custom attributes sent with the request.
	Attributes() map[string]string
	// SystemAttributeByKey will return the system attributes by key.
	SystemAttributeByKey(key string) string
	// SystemAttributes will return the system attributes.
	SystemAttributes() map[string]string
	// Metadata will return the metadata sent throughout the request.
	Metadata() map[string]string
	// Identifier will return an identifier associated with the message ReceiptHandle.
	Identifier() string
	// Dispatch used to dispatch a message if necessary
	Dispatch()
	// Backoff used to change the visibilityTimeout of the message
	// when a message is backoff it will not be removed from the queue
	// instead it will extend the visibility timeout of the message
	Backoff(delay time.Duration)
	// BackedOff used to check if the message was backedOff by the handler
	BackedOff() bool
	// Body used to get the message Body
	Body() []byte
	// Message returns the body message
	Message() string
	// TimeStamp returns the message timestamp
	TimeStamp() time.Time
	// DecodeMessage will unmarshal the message into a supplied output using JSON
	DecodeMessage(out any) error
}

Message represents the message interface methods

type Mode

type Mode int

Mode defines how messages should be dispatched to workers.

const (
	// Parallel processes messages independently without considering group identifiers.
	Parallel Mode = iota

	// PerGroupID ensures startWorker processes messages based on MessageGroupId and custom grouping fields.
	PerGroupID
)

type NoOpLogger

type NoOpLogger struct{}

NoOpLogger is a logger that does nothing.

func (NoOpLogger) Log

func (NoOpLogger) Log(args ...any)

Log implements Logger but does nothing.

type Router

type Router interface {
	Configure(ctx context.Context) error
	GetMessages(ctx context.Context, logger Logger) ([]Message, error)
	HandlerMessage(ctx context.Context, msg Message) error
	Commit(ctx context.Context, m Message) error
	WorkerPoolSize(ctx context.Context) int32
	VisibilityTimeout(ctx context.Context) int32
	RunMode(ctx context.Context) Mode
	CustomGroupFields(ctx context.Context) []string
}

Router holds the Route methods to configure and run

type SNSClient

type SNSClient interface {
	Publish(ctx context.Context, params *sns.PublishInput, optFns ...func(*sns.Options)) (*sns.PublishOutput, error)
	PublishBatch(ctx context.Context, params *sns.PublishBatchInput, optFns ...func(*sns.Options)) (*sns.PublishBatchOutput, error)
}

SNSClient represents the aws sns client methods

type SQSClient

type SQSClient interface {
	ChangeMessageVisibility(
		ctx context.Context,
		params *sqs.ChangeMessageVisibilityInput,
		optFns ...func(*sqs.Options)) (*sqs.ChangeMessageVisibilityOutput, error)
	GetQueueUrl(ctx context.Context, params *sqs.GetQueueUrlInput, optFns ...func(*sqs.Options)) (*sqs.GetQueueUrlOutput, error)
	ReceiveMessage(ctx context.Context, params *sqs.ReceiveMessageInput, optFns ...func(*sqs.Options)) (*sqs.ReceiveMessageOutput, error)
	DeleteMessage(ctx context.Context, params *sqs.DeleteMessageInput, optFns ...func(*sqs.Options)) (*sqs.DeleteMessageOutput, error)
}

SQSClient represents the aws sqs client methods