squeue

README

squeue is a Golang library designed to facilitate interactions with Amazon SQS queues, providing 100% type-safety powered by Golang generics.

---

Key Features

Driver-based Implementation

squeue abstracts the internal driver implementation through its interface, allowing for seamless integration of various drivers to suit different development environments and needs.

Currently implemented drivers:

• In-Memory
• Amazon SQS

You can easily deploy the Amazon SQS driver in any staging or production environment and use the In-Memory driver for lightweight testing.

Type Safety

The library uses an opinionated JSON format for serializing and deserializing messages. To achieve 100% type safety, simply provide a message type that implements the Golang json.Marshaler and json.Unmarshaler interfaces. This way, you maintain direct control over the raw data serialization and deserialization.

Channel of Messages

Consuming messages is straightforward—just loop through a channel of messages:

messages, _ := sub.Consume(ctx, "queue")
for m := range messages {
    go func(message squeue.Message[*my.Message]) {
        // Do your magic...
    }(m)
}

Examples

For more detailed documentation, please refer to the internal/examples/ directory.

In-Memory Driver

go run internal/examples/memory/main.go

Amazon SQS Driver

go run internal/examples/sqs/consumer/consumer.go
# Open another shell
go run internal/examples/sqs/producer/producer.go

Documentation

Index

• Constants
• type Consumer
  • func NewConsumer[T json.Unmarshaler](d driver.Driver, queue string) Consumer[T]
  • func (p *Consumer[T]) Ack(m Message[T]) error
  • func (p *Consumer[T]) Consume(opts ...func(message any)) (chan Message[T], error)
  • func (p Consumer[T]) Ping() error
  • func (p *Consumer[T]) Stop()
• type Err
  • func (e *Err) Code() int
  • func (e *Err) Data() []byte
  • func (e *Err) Error() string
  • func (e *Err) Unwrap() error
• type Message
• type Producer
  • func NewProducer(d driver.Driver, queue string) Producer
  • func (q *Producer) Enqueue(message json.Marshaler, opts ...func(message any)) error

Constants

const (
	ErrUnmarshal int = -1
	ErrMarshal   int = -2
	ErrDriver    int = -3
)

Types

type Consumer

type Consumer[T json.Unmarshaler] struct {
	// contains filtered or unexported fields
}

func NewConsumer

func NewConsumer[T json.Unmarshaler](d driver.Driver, queue string) Consumer[T]

NewConsumer creates a new consumer for the given T type of messages

func (*Consumer[T]) Ack

func (p *Consumer[T]) Ack(m Message[T]) error

Ack explicitly acknowldge the message handling. It can be implemented as No Operation for some drivers.

func (*Consumer[T]) Consume

func (p *Consumer[T]) Consume(opts ...func(message any)) (chan Message[T], error)

Consume retrieves messages from the given queue. Any provided options will be sent to the underlying driver. The messages are indefinetely consumed from the queue and sent to the chan Message[T].

func (Consumer[T]) Ping

func (p Consumer[T]) Ping() error

Ping runs a ping on the underlying driver

func (*Consumer[T]) Stop

func (p *Consumer[T]) Stop()

type Err

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

func (*Err) Code

func (e *Err) Code() int

func (*Err) Data

func (e *Err) Data() []byte

func (*Err) Error

func (e *Err) Error() string

func (*Err) Unwrap

func (e *Err) Unwrap() error

type Message

type Message[T json.Unmarshaler] struct {
	Content T
	ID      string
	Error   error
}

sub := squeue.NewConsumer[*myMessage](d)

for m := range sub.Consume() {
    do something with m
}

type Producer

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

func NewProducer

func NewProducer(d driver.Driver, queue string) Producer

func (*Producer) Enqueue

func (q *Producer) Enqueue(message json.Marshaler, opts ...func(message any)) error

Enqueue sends a message to the underlying queue, any provided options will be sent to the driver.