sqs

package
v1.148.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 11, 2026 License: MIT Imports: 12 Imported by: 0

Documentation

Overview

Package sqs solves the operational friction of using AWS SQS directly from the aws-sdk-go-v2 in application code. It wraps github.com/aws/aws-sdk-go-v2/service/sqs with a small, focused API that covers the common queue workflow: send, receive, decode, acknowledge (delete), and health-check.

Problem

Raw SQS integration requires repetitive boilerplate across services: queue and endpoint setup, FIFO message-group handling, long-poll configuration, visibility-timeout tuning, payload serialization, and error-safe delete logic. Teams often re-implement this stack in each service, which leads to subtle behavior drift and harder testing.

This package standardizes that flow into one client with explicit options and safe defaults.

How It Works

New creates a Client bound to a queue URL and optional message-group ID.

Receive flow behavior:

  • Client.Receive returns nil,nil when no message arrives within the long-poll window.
  • Client.ReceiveData returns an empty receipt handle when no message is available.
  • After successful processing, callers should acknowledge via Client.Delete using the receipt handle.
  • If decode fails, Client.ReceiveData still returns the receipt handle so callers can choose whether to delete or re-queue according to their policy.

Key Features

  • Simple SQS workflow API: send, receive, delete, and health check.
  • FIFO-aware safety: validates message-group usage for FIFO queues and supports explicit per-message deduplication IDs.
  • Typed payload support with customizable serialization/encryption via WithMessageEncodeFunc and WithMessageDecodeFunc.
  • Endpoint and AWS config customization via WithAWSOptions, WithSrvOptionFuncs, WithEndpointMutable, and WithEndpointImmutable for local testing and advanced deployments.
  • Full client injection via WithSQSClient for tests and advanced integrations; when set, the AWS configuration is not loaded and the AWS and service options above are ignored.
  • Health probe support through Client.HealthCheck, which verifies queue accessibility in the configured region.

Usage 

c, err := sqs.New(ctx,
    "https://sqs.us-east-1.amazonaws.com/123456789012/my-queue",
    "", // non-FIFO queue
    sqs.WithWaitTimeSeconds(20),
    sqs.WithVisibilityTimeout(300),
)
if err != nil {
    return err
}

// Send typed payload
if err := c.SendData(ctx, event); err != nil {
    return err
}

// Receive typed payload
var msg Event
receiptHandle, err := c.ReceiveData(ctx, &msg)
if err != nil {
    return err
}
if receiptHandle != "" {
    _ = c.Delete(ctx, receiptHandle)
}

This package is ideal for Go services that need a minimal, consistent, production-friendly abstraction over SQS without losing access to AWS SDK customization.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/aws/aws-sdk-go-v2/aws"
	awssqs "github.com/aws/aws-sdk-go-v2/service/sqs"
	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
	"github.com/tecnickcom/nurago/pkg/sqs"
)

// exampleSQSClient is a minimal in-memory stand-in for the AWS SQS client,
// implementing only the sqs.SQS interface. Real code would let New build the
// client from aws.Config instead.
type exampleSQSClient struct {
	body string
}

func (c *exampleSQSClient) SendMessage(
	_ context.Context,
	params *awssqs.SendMessageInput,
	_ ...func(*awssqs.Options),
) (*awssqs.SendMessageOutput, error) {
	c.body = aws.ToString(params.MessageBody)

	return &awssqs.SendMessageOutput{}, nil
}

func (c *exampleSQSClient) ReceiveMessage(
	_ context.Context,
	_ *awssqs.ReceiveMessageInput,
	_ ...func(*awssqs.Options),
) (*awssqs.ReceiveMessageOutput, error) {
	return &awssqs.ReceiveMessageOutput{
		Messages: []types.Message{{
			Body:          aws.String(c.body),
			ReceiptHandle: aws.String("example-receipt-handle"),
		}},
	}, nil
}

func (c *exampleSQSClient) DeleteMessage(
	_ context.Context,
	_ *awssqs.DeleteMessageInput,
	_ ...func(*awssqs.Options),
) (*awssqs.DeleteMessageOutput, error) {
	return &awssqs.DeleteMessageOutput{}, nil
}

func (c *exampleSQSClient) GetQueueAttributes(
	_ context.Context,
	_ *awssqs.GetQueueAttributesInput,
	_ ...func(*awssqs.Options),
) (*awssqs.GetQueueAttributesOutput, error) {
	return &awssqs.GetQueueAttributesOutput{}, nil
}

func main() {
	// A caller would normally configure a real client via options such as
	// WithAWSOptions or WithEndpointMutable (and the region would be derived from
	// the queue URL); here an injected client keeps the example self-contained, so
	// no AWS configuration is loaded.
	c, err := sqs.New(
		context.TODO(),
		"https://sqs.us-east-1.amazonaws.com/123456789012/my-queue",
		"", // standard (non-FIFO) queue: no message group ID
		sqs.WithSQSClient(&exampleSQSClient{}),
	)
	if err != nil {
		fmt.Println("error:", err)

		return
	}

	err = c.Send(context.TODO(), "hello world")
	if err != nil {
		fmt.Println("error:", err)

		return
	}

	msg, err := c.Receive(context.TODO())
	if err != nil {
		fmt.Println("error:", err)

		return
	}

	fmt.Println(msg.Body)

	// After processing, acknowledge the message by deleting it.
	err = c.Delete(context.TODO(), msg.ReceiptHandle)
	if err != nil {
		fmt.Println("error:", err)

		return
	}

}

Output:
hello world

Index

Examples

Constants

View Source 
const (
	// DefaultWaitTimeSeconds is the default duration (in seconds) for which the call waits for a message to arrive in the queue before returning.
	// This must be between 0 and 20 seconds.
	DefaultWaitTimeSeconds = 20

	// DefaultVisibilityTimeout is the default duration (in seconds) that the received messages are hidden from subsequent retrieve requests after being retrieved by a ReceiveMessage request.
	DefaultVisibilityTimeout = 600
)

Variables

View Source 
var (
	// ErrInvalidQueueURL is returned by New when queueURL is empty, unparseable,
	// or lacks a scheme or host.
	ErrInvalidQueueURL = errors.New("sqs: invalid queue URL")

	// ErrMissingMessageGroupID is returned by New when a FIFO queue URL is given
	// without a valid message group ID.
	ErrMissingMessageGroupID = errors.New("sqs: a valid message group ID is required for a FIFO queue")

	// ErrUnexpectedMessageGroupID is returned by New when a message group ID is
	// supplied for a standard (non-FIFO) queue, which does not use message groups.
	ErrUnexpectedMessageGroupID = errors.New("sqs: a message group ID must not be set for a standard queue")

	// ErrDedupIDNotAllowed is returned by SendWithDeduplicationID and
	// SendDataWithDeduplicationID when the target queue is not a FIFO queue.
	ErrDedupIDNotAllowed = errors.New("sqs: a message deduplication ID can only be used with FIFO queues")

	// ErrInvalidDedupID is returned by SendWithDeduplicationID and
	// SendDataWithDeduplicationID when the deduplication ID is empty or invalid.
	ErrInvalidDedupID = errors.New("sqs: invalid message deduplication ID")

	// ErrNilEncodeFunc is returned by New when the message encode function is nil.
	ErrNilEncodeFunc = errors.New("sqs: nil message encode function")

	// ErrNilDecodeFunc is returned by New when the message decode function is nil.
	ErrNilDecodeFunc = errors.New("sqs: nil message decode function")

	// ErrInvalidWaitTime is returned by New when waitTimeSeconds is outside the
	// valid 0..20 seconds range.
	ErrInvalidWaitTime = errors.New("sqs: waitTimeSeconds must be between 0 and 20 seconds")

	// ErrInvalidVisibilityTimeout is returned by New when visibilityTimeout is
	// outside the valid 0..43200 seconds range.
	ErrInvalidVisibilityTimeout = errors.New("sqs: visibilityTimeout must be between 0 and 43200 seconds")

	// ErrQueueNotResponding is returned by HealthCheck when the queue does not
	// return the expected attribute.
	ErrQueueNotResponding = errors.New("sqs: the queue is not responding")
)

Exported sentinel errors returned by this package. Match them with errors.Is.

Functions

func DefaultMessageDecodeFunc 

func DefaultMessageDecodeFunc(_ context.Context, msg string, data any) error

DefaultMessageDecodeFunc is the default deserializer used by ReceiveData.

func DefaultMessageEncodeFunc 

func DefaultMessageEncodeFunc(_ context.Context, data any) (string, error)

DefaultMessageEncodeFunc is the default serializer used by SendData.

func MessageDecode 

func MessageDecode(msg string, data any) error

MessageDecode decodes a MessageEncode payload into data, which must be a pointer.

func MessageEncode 

func MessageEncode(data any) (string, error)

MessageEncode encodes and serializes data into an SQS-compatible string payload.

Types

type Client 

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

Client wraps AWS SQS operations and typed message encoding/decoding for a single queue URL.

func New 

func New(ctx context.Context, queueURL, msgGroupID string, opts ...Option) (*Client, error)

New builds a client for queueURL and validates FIFO message-group constraints.

queueURL must be a valid absolute URL. msgGroupID is required (and must be a valid FIFO group ID) when queueURL targets a FIFO queue, and must be empty for a standard queue. Cheap argument validation runs before any AWS configuration is loaded, so misconfiguration fails fast.

It returns ErrInvalidQueueURL when queueURL is empty or malformed, ErrMissingMessageGroupID when a FIFO queue is given without a valid group ID, and ErrUnexpectedMessageGroupID when a group ID is set for a standard queue. Invalid options surface ErrNilEncodeFunc, ErrNilDecodeFunc, ErrInvalidWaitTime, or ErrInvalidVisibilityTimeout.

The returned Client is safe for concurrent use.

func (*Client) Delete 

func (c *Client) Delete(ctx context.Context, receiptHandle string) error

Delete removes a message from the queue by receipt handle.

func (*Client) HealthCheck 

func (c *Client) HealthCheck(ctx context.Context) error

HealthCheck verifies queue reachability by fetching a known queue attribute.

It returns ErrQueueNotResponding when the queue call succeeds but does not return the expected attribute; the underlying AWS error is wrapped otherwise.

func (*Client) Receive 

func (c *Client) Receive(ctx context.Context) (*Message, error)

Receive retrieves one raw message from the queue with configured wait/visibility settings. Returns nil message when no message is available within waitTimeSeconds.

func (*Client) ReceiveData 

func (c *Client) ReceiveData(ctx context.Context, data any) (string, error)

ReceiveData receives one message, decodes its payload into data, and returns the receipt handle. If decoding fails, receipt handle is still returned so callers can decide whether to delete or requeue.

func (*Client) Send 

func (c *Client) Send(ctx context.Context, message string) error

Send publishes a raw string message to the queue.

NOTE: no MessageDeduplicationId is set, so for FIFO queues this requires ContentBasedDeduplication to be enabled on the queue; otherwise AWS rejects the request. Use SendWithDeduplicationID to supply an explicit deduplication ID instead.

func (*Client) SendData 

func (c *Client) SendData(ctx context.Context, data any) error

SendData encodes data via configured codec and publishes it to the queue.

NOTE: no MessageDeduplicationId is set, so for FIFO queues this requires ContentBasedDeduplication to be enabled on the queue; otherwise AWS rejects the request. Use SendDataWithDeduplicationID to supply an explicit deduplication ID instead.

func (*Client) SendDataWithDeduplicationID 

func (c *Client) SendDataWithDeduplicationID(ctx context.Context, data any, dedupID string) error

SendDataWithDeduplicationID encodes data via configured codec and publishes it to the queue with an explicit MessageDeduplicationId. See SendWithDeduplicationID for the FIFO-queue deduplication constraints.

func (*Client) SendWithDeduplicationID 

func (c *Client) SendWithDeduplicationID(ctx context.Context, message, dedupID string) error

SendWithDeduplicationID publishes a raw string message to the queue with an explicit MessageDeduplicationId.

It is only valid for FIFO queues and is required when the queue does not have ContentBasedDeduplication enabled. Messages sent with the same deduplication ID within the 5-minute deduplication interval are accepted but not delivered again. The dedupID can contain up to 128 alphanumeric and punctuation characters.

It returns ErrDedupIDNotAllowed when the queue is not a FIFO queue and ErrInvalidDedupID when dedupID is empty or malformed.

type Message 

type Message struct {
	// Body is the message content and can contain: JSON, XML or plain text.
	Body string

	// ReceiptHandle is the identifier used to delete the message.
	ReceiptHandle string
}

Message holds a received payload and the receipt handle used for deletion.

type Option 

type Option func(*cfg)

Option applies a configuration change to the internal SQS client settings.

func WithAWSOptions 

func WithAWSOptions(opt awsopt.Options) Option

WithAWSOptions appends awsopt options used to build aws.Config.

func WithEndpointImmutable 

func WithEndpointImmutable(url string) Option

WithEndpointImmutable installs a fixed EndpointResolverV2 for deterministic endpoint routing.

func WithEndpointMutable 

func WithEndpointMutable(url string) Option

WithEndpointMutable sets BaseEndpoint while preserving SDK endpoint mutability.

func WithMessageDecodeFunc 

func WithMessageDecodeFunc(f TDecodeFunc) Option

WithMessageDecodeFunc overrides the deserializer used by ReceiveData. The data argument passed to ReceiveData must be a pointer to the expected type.

func WithMessageEncodeFunc 

func WithMessageEncodeFunc(f TEncodeFunc) Option

WithMessageEncodeFunc overrides the serializer used by SendData.

func WithSQSClient 

func WithSQSClient(client SQS) Option

WithSQSClient injects a custom SQS implementation.

This is primarily useful for tests and advanced integrations where the caller needs full control over request behavior without creating a real sqs.Client from aws.Config. When set, the injected client is used as-is: the AWS configuration is not loaded and the AWS/service options (WithAWSOptions, WithSrvOptionFuncs, WithEndpointMutable, WithEndpointImmutable) are ignored.

func WithSrvOptionFuncs 

func WithSrvOptionFuncs(opt ...SrvOptionFunc) Option

WithSrvOptionFuncs appends service-level sqs.Options mutators.

func WithVisibilityTimeout 

func WithVisibilityTimeout(t int32) Option

WithVisibilityTimeout sets message invisibility duration in seconds after receive. Values range: 0 to 43200. Maximum: 12 hours.

func WithWaitTimeSeconds 

func WithWaitTimeSeconds(t int32) Option

WithWaitTimeSeconds sets long-poll wait duration in seconds for ReceiveMessage calls. Values range: 0 to 20 seconds.

type SQS 

type SQS interface {
	DeleteMessage(ctx context.Context, params *sqs.DeleteMessageInput, optFns ...func(*sqs.Options)) (*sqs.DeleteMessageOutput, error)
	GetQueueAttributes(ctx context.Context, params *sqs.GetQueueAttributesInput, optFns ...func(*sqs.Options)) (*sqs.GetQueueAttributesOutput, error)
	ReceiveMessage(ctx context.Context, params *sqs.ReceiveMessageInput, optFns ...func(*sqs.Options)) (*sqs.ReceiveMessageOutput, error)
	SendMessage(ctx context.Context, params *sqs.SendMessageInput, optFns ...func(*sqs.Options)) (*sqs.SendMessageOutput, error)
}

SQS defines the AWS SDK SQS calls used by Client.

type SrvOptionFunc 

type SrvOptionFunc = func(*sqs.Options)

SrvOptionFunc aliases an AWS SDK SQS service option mutator.

type TDecodeFunc 

type TDecodeFunc func(ctx context.Context, msg string, data any) error

TDecodeFunc is the type of function used to replace the default message decoding function used by ReceiveData().

type TEncodeFunc 

type TEncodeFunc func(ctx context.Context, data any) (string, error)

TEncodeFunc is the type of function used to replace the default message encoding function used by SendData().

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.