Documentation
¶
Overview ¶
Package sqs implements a Gas job queue on AWS SQS, or on an SQS-compatible service through a custom endpoint (for example LocalStack or ElasticMQ).
Role in the framework ¶
Service implements the core interfaces gas.Service, gas.JobQueueProvider and gas.ReadyReporter. Queues are named by their queue URL, and the package does not create them. Client returns the underlying *sqs.Client for calls beyond gas.JobQueueProvider, such as creating or purging a queue; it returns nil when WithClient injected a client of another type.
Wiring ¶
New takes Options (WithConfig, WithClient) and returns a constructor that the DI container calls with gas.ConfigProvider and gas.Logger. The app registers that constructor; the package registers nothing itself.
Configuration ¶
Unless WithConfig is given, Init binds Config from the gas.ConfigProvider over DefaultConfig, so a key that is not set keeps its default. The keys sit under queue: region ("us-east-1"; must not be empty), endpoint (optional custom endpoint; empty means the AWS default), accesskeyid and secretaccesskey (static credentials, used only when both are set; otherwise the default AWS credential chain applies), visibilitytimeout (30s; must not be negative) and waittimeseconds (20; 0 to 20). WithConfig replaces the whole Config and skips the provider, so start from DefaultConfig. Config.Validate enforces these rules, and an invalid Config fails Init, even when WithClient supplies the client.
Lifecycle ¶
Init runs at startup. It binds and validates the configuration and, unless WithClient was given, builds the SQS client from the AWS default configuration (environment and shared config files), loaded under Init's context, and then fetches the credentials under the same context, so a missing or broken credential setup fails startup. It does not contact SQS, so an unreachable endpoint, rejected credentials or a missing queue surface on the first operation.
Close runs at shutdown. It only marks the Service closed: Enqueue, Dequeue, Ack, Nack and CheckReady then return queue.ErrClosed, so a readiness probe takes the instance out of rotation while it drains. CheckReady also fails before Init has built a client. The Service starts no goroutines; a consumer loop that calls Dequeue belongs to the app.
Operations ¶
Enqueue sends the payload as the message body. Its options map to SQS: a delay becomes DelaySeconds (whole seconds), a group ID and a deduplication ID are passed for FIFO queues, and attributes become String message attributes. Dequeue long-polls ReceiveMessage. It clamps maxMessages to SQS's range of 1 to 10, uses waittimeseconds when the wait it is given truncates to zero whole seconds, and applies visibilitytimeout to every message it receives. Each gas.Job carries the message ID, receipt handle and body, and its Attributes merge the system attributes with the String message attributes, the latter winning on a shared name. Ack deletes the message; Nack sets its visibility timeout to zero, so it can be received again at once. Delivery, ordering and retries are SQS's: a message that is neither acked nor nacked becomes visible again when its visibility timeout ends.
Errors ¶
queue.ErrClosed is the only sentinel. Errors from SQS calls start with the service name, "gas/queue/sqs", and wrap the SDK's error with %w. Errors from Validate name the offending field instead, for example Queue.Region, and a Dequeue argument that overflows int32 returns an error that names the argument.
SPDX-License-Identifier: MIT
Index ¶
- func New(opts ...Option) func(gas.ConfigProvider, gas.Logger) (*Service, error)
- type Config
- type Option
- type Service
- func (s *Service) Ack(ctx context.Context, queueURL string, job gas.Job) error
- func (s *Service) CheckReady(_ context.Context) error
- func (s *Service) Client() *awssqs.Client
- func (s *Service) Close(_ context.Context) error
- func (s *Service) Dequeue(ctx context.Context, queueURL string, maxMessages int, wait time.Duration) ([]gas.Job, error)
- func (s *Service) Enqueue(ctx context.Context, queueURL string, payload []byte, ...) error
- func (s *Service) Init(ctx context.Context) error
- func (s *Service) Nack(ctx context.Context, queueURL string, job gas.Job) error
- func (s *Service) Name() string
- type Settings
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Config ¶
type Config struct {
gas.WithGasEnv
Queue Settings
}
Config holds SQS queue settings.
func DefaultConfig ¶
func DefaultConfig() *Config
DefaultConfig returns a Config with sensible defaults.
type Option ¶
type Option func(*Service)
Option configures a Service.
func WithClient ¶
func WithClient(client sqsClient) Option
WithClient injects a pre-configured SQS client. Useful for testing or when the caller manages AWS credentials externally.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service is an SQS-backed queue implementing gas.Service and gas.JobQueueProvider.
func (*Service) CheckReady ¶
CheckReady reports whether the service is ready to accept traffic. Returns an error before Init has run or after Close has been called.
func (*Service) Client ¶
Client returns the underlying *sqs.Client for advanced operations beyond the JobQueueProvider interface (e.g. CreateQueue, PurgeQueue, GetQueueAttributes). Returns nil if a custom sqsClient was injected via WithClient that is not an *sqs.Client.
func (*Service) Dequeue ¶
func (s *Service) Dequeue( ctx context.Context, queueURL string, maxMessages int, wait time.Duration, ) ([]gas.Job, error)
Dequeue receives messages from the specified SQS queue.
func (*Service) Enqueue ¶
func (s *Service) Enqueue( ctx context.Context, queueURL string, payload []byte, opts ...gas.EnqueueOption, ) error
Enqueue sends a message to the specified SQS queue.
func (*Service) Init ¶
Init binds the configuration and creates the SQS client. A config bound from the ConfigProvider is validated by Bind against its validate tags; one passed with WithConfig is used as given. ctx bounds loading the AWS configuration and fetching the credentials, so Init fails when no credentials can be found.
type Settings ¶
type Settings struct {
// Region is the AWS region for the SQS service. Required.
Region string `validate:"required"`
// Endpoint is an optional custom endpoint URL (e.g. for LocalStack).
// Empty means use the default AWS endpoint.
Endpoint string
// AccessKeyID is the AWS access key ID for static credentials.
// If empty, the default AWS credential chain is used.
AccessKeyID string
// SecretAccessKey is the AWS secret access key for static credentials.
// If empty, the default AWS credential chain is used.
SecretAccessKey string
// VisibilityTimeout is how long a dequeued message stays invisible
// to other consumers. SQS max is 12 hours. Must not be negative.
VisibilityTimeout time.Duration `validate:"gte=0s"`
// WaitTimeSeconds is the long-poll duration for ReceiveMessage calls.
// Must be 0-20 (SQS hard limit).
WaitTimeSeconds int `validate:"min=0,max=20"`
}
Settings represents the configuration for the SQS queue service.
Source Files