Documentation
¶
Overview ¶
Package sqs is doze-aws's ground-up, pure-Go SQS-compatible service: no LocalStack, no JVM. It speaks both wire protocols (AWS JSON 1.0 used by modern SDKs and the legacy Query/XML protocol used by aws-sdk-go v1-era clients), persists to a bbolt store under the data directory, and supports visibility timeout, delay, retention, message attributes, long polling, FIFO queues (group ordering + deduplication), dead-letter redrive, queue tags, and message move tasks.
See docs/api-support/sqs.md for the operation-by-operation support table.
Index ¶
- type Attr
- type Message
- type MoveTask
- type Options
- type Queue
- type Server
- type Store
- func (s *Store) Attributes(name string) (map[string]string, error)
- func (s *Store) CancelMessageMoveTask(handle string) error
- func (s *Store) ChangeVisibility(queue, handle string, timeout int) error
- func (s *Store) CreateQueue(name string, attrs map[string]string, tags map[string]string) (*Queue, error)
- func (s *Store) DeadLetterSourceQueues(dlq string) ([]string, error)
- func (s *Store) Delete(queue, handle string) error
- func (s *Store) DeleteQueue(name string) error
- func (s *Store) ListMessageMoveTasks(source string, max int) ([]MoveTask, error)
- func (s *Store) ListQueues(prefix string) ([]string, error)
- func (s *Store) Peek(queue string, max int) ([]Message, error)
- func (s *Store) Purge(queue string) error
- func (s *Store) Receive(queue string, max, waitSec int, visibilityOverride int) ([]Message, error)
- func (s *Store) Send(queue, body string, attrs map[string]Attr, delay int, groupID, dedupID string) (*Message, error)
- func (s *Store) SetAttributes(name string, attrs map[string]string) error
- func (s *Store) StartMessageMoveTask(source, dest string) (*MoveTask, error)
- func (s *Store) Sweep()
- func (s *Store) TagQueue(name string, tags map[string]string) error
- func (s *Store) Tags(name string) (map[string]string, error)
- func (s *Store) UntagQueue(name string, keys []string) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Attr ¶
type Attr struct {
DataType string `json:"data_type"`
StringValue string `json:"string_value,omitempty"`
BinaryValue []byte `json:"binary_value,omitempty"`
}
Attr is a message attribute (String/Number use StringValue; Binary uses BinaryValue).
type Message ¶
type Message struct {
ID string `json:"id"`
Body string `json:"body"`
Attrs map[string]Attr `json:"attrs,omitempty"`
MD5Body string `json:"md5_body"`
MD5Attrs string `json:"md5_attrs,omitempty"`
Sent int64 `json:"sent"` // unixnano
VisibleAt int64 `json:"visible_at"` // unixnano; <= now => visible
ReceiveCount int `json:"receive_count"`
FirstReceived int64 `json:"first_received"` // unixnano, 0 if never
GroupID string `json:"group_id,omitempty"`
DedupID string `json:"dedup_id,omitempty"`
Seq uint64 `json:"seq"`
}
Message is one stored message.
type MoveTask ¶
type MoveTask struct {
Handle string `json:"handle"`
Status string `json:"status"` // COMPLETED | FAILED
Source string `json:"source"` // queue name
Destination string `json:"destination"`
Moved int `json:"moved"`
StartedAt int64 `json:"started_at"` // unix seconds
FailureWhy string `json:"failure_why,omitempty"`
}
MoveTask records one message move task. Local moves are synchronous, so a stored task is always in a terminal state.
type Options ¶
type Options struct {
// DataDir holds the bbolt store (sqs.bolt). Required.
DataDir string
// Peers is accepted for constructor uniformity; SQS initiates no
// cross-service calls today (Lambda event source mappings poll SQS from
// the Lambda side).
Peers peers.Directory
// Logf receives log lines; nil discards.
Logf func(format string, args ...any)
// Clock overrides time.Now in tests.
Clock func() time.Time
}
Options configures the service.
type Queue ¶
type Queue struct {
Name string `json:"name"`
FIFO bool `json:"fifo"`
ContentBasedDedup bool `json:"content_based_dedup"`
VisibilityTimeout int `json:"visibility_timeout"` // seconds
DelaySeconds int `json:"delay_seconds"`
RetentionPeriod int `json:"retention_period"` // seconds
MaxMessageSize int `json:"max_message_size"`
WaitTimeSeconds int `json:"wait_time_seconds"` // default receive long-poll
DeadLetterTarget string `json:"dead_letter_target"` // target queue name, "" if none
MaxReceiveCount int `json:"max_receive_count"`
Created int64 `json:"created"` // unix seconds
Tags map[string]string `json:"tags,omitempty"`
}
Queue is a queue's durable definition.
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
Server is the SQS service: an http.Handler speaking both SQS wire protocols, and an io.Closer that stops the janitor and closes the store.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is the bbolt-backed SQS state.
func (*Store) Attributes ¶
Attributes returns the GetQueueAttributes view of a queue.
func (*Store) CancelMessageMoveTask ¶
CancelMessageMoveTask always fails locally: moves complete synchronously, so by the time a cancel arrives the task is already terminal — which is exactly what AWS reports for a finished task.
func (*Store) ChangeVisibility ¶
func (*Store) CreateQueue ¶
func (s *Store) CreateQueue(name string, attrs map[string]string, tags map[string]string) (*Queue, error)
CreateQueue creates (or, idempotently, updates the attributes of) a queue.
func (*Store) DeadLetterSourceQueues ¶
DeadLetterSourceQueues lists the queues whose redrive policy targets dlq.
func (*Store) DeleteQueue ¶
func (*Store) ListMessageMoveTasks ¶
ListMessageMoveTasks returns the recorded tasks for a source queue, newest first, up to max.
func (*Store) Peek ¶
Peek returns up to max currently-visible messages in queue (FIFO) order WITHOUT consuming them: it never changes visibility, never increments the receive count, and ignores FIFO group locking — so it shows the FULL queue contents (every message, not just the head of each message group, the way a plain Receive does). Purely read-only; the returned handles are still valid for Delete.
func (*Store) Receive ¶
Receive returns up to max visible messages, applying visibility timeout, FIFO group locking, DLQ redrive, and retention. waitSec long-polls when empty.
func (*Store) Send ¶
func (s *Store) Send(queue, body string, attrs map[string]Attr, delay int, groupID, dedupID string) (*Message, error)
Send enqueues a message. delay<0 means "use the queue default".
func (*Store) SetAttributes ¶
func (*Store) StartMessageMoveTask ¶
StartMessageMoveTask moves every currently-stored message from source to dest, synchronously — the local equivalent of a DLQ redrive. AWS moves asynchronously with rate control; locally the volumes are small enough that completing inline is simpler and deterministic.
func (*Store) Sweep ¶
func (s *Store) Sweep()
Sweep drops retention-expired messages and prunes stale dedup entries across every queue. Receive does this lazily for read queues; a periodic Sweep also reclaims write-only queues so nothing grows unbounded.
Source Files