azurequeue

package
v1.6.0 Latest Latest
Warning

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

 Go to latest Published: Sep 14, 2026 License: MIT Imports: 21 Imported by: 0

README

Azure Queue Storage

Parity grade: C · SDK azure-sdk-for-go/sdk/storage/azqueue@v1.0.1 · last audited 2026-09-04 (c7cf3aec)

Coverage

Metric Value
PARITY entries audited 11 (10 ok, 1 partial)
Feature families 5 (4 ok, 1 partial)
Known gaps 7
Deferred items 1
Resource leaks clean
Known gaps
  • No queue metadata (x-ms-meta-*) support -- neither stored on Create nor returned on List; Create is therefore always idempotent (204) for a pre-existing queue, and QueueAlreadyExists (409) is unreachable in this MVP.
  • messagettl=-1 (Azure's 'never expire' sentinel) is modeled as a 100-year TTL, not true infinite retention.
  • List Queues returns every result in one page; no prefix/marker/maxresults pagination.
  • No SetQueueMetadata/GetQueueMetadata, no queue ACL (Set/Get Queue ACL / SAS-scoped access policies).
  • No poison-message / dead-letter handling -- DequeueCount is tracked and returned but nothing acts on it (contrast services/sqs's DLQ).
  • Visibility-timeout expiry is checked lazily at read time (Get/Peek Messages), not proactively swept -- functionally correct (a message becomes visible again exactly when TimeNextVisible passes, regardless of sweep timing) but differs from message-TTL expiry, which the Janitor does proactively sweep.
  • Auth verification is not enforced -- see families.auth. pkgs/azureauth.VerifySharedKey exists and is unit-tested but checkAuth does not call it yet. All gaps above are intentional MVP scope per AZURE.md's M1 entry (see AZURE.md section 8), not oversights.
Deferred
  • Initial implementation pass (2026-09-04): seeded this service from scratch per AZURE.md M1 (see AZURE.md section 8). Structurally mirrors services/azureblob's M0 implementation and PARITY.md format; no prior audit history to reconcile.

More

Documentation

Overview

Package azurequeue provides a local, in-memory emulation of Azure Queue Storage's REST+XML wire protocol (queue CRUD plus the full message lifecycle: put/get/peek/delete/update-visibility/clear), Azurite-compatible enough for unmodified azure-sdk-for-go clients to operate against. See AZURE.md and PARITY.md for scope and known gaps.

Index

Constants

View Source 
const (
	// DefaultVisibilityTimeout is applied when a caller omits
	// visibilitytimeout on Put Message / Get Messages.
	DefaultVisibilityTimeout = 30 * time.Second
	// DefaultMessageTTL is applied when a caller omits messagettl on Put
	// Message (or passes -1, Azure's "infinite" sentinel is not modeled --
	// see PARITY.md known gaps).
	DefaultMessageTTL = 7 * 24 * time.Hour
	// MaxNumOfMessages is the largest numofmessages Get/Peek Messages
	// accepts per call.
	MaxNumOfMessages = 32
	// MinNumOfMessages is the smallest numofmessages Get/Peek Messages
	// accepts per call.
	MinNumOfMessages = 1
)

Default and bound values for Put/Get Messages query parameters, matching real Azure Queue Storage's documented limits.

View Source 
const DefaultPort = 10001

DefaultPort is Azure Queue's fixed, protocol-conventional TCP port. This follows the same pattern as services/azureblob's DefaultPort (10000) and, before that, services/iot's MQTT broker (1883): pick one default and try to bind exactly that, rather than drawing from cli.go's shared --port-range-start/--port-range-end PortAlloc pool (used for on-demand ephemeral resources like Lambda function URLs and ElastiCache, not fixed service ports) or inventing an alternative numbering scheme. The default value itself (10001) is Azurite's own Queue service port, so unmodified UseDevelopmentStorage=true-style SDK configuration works out of the box; see AZURE.md section 4 for the full rationale, including why this deliberately does NOT fall back into the shared PortAlloc pool if 10001 is taken (StartWorker fails fast instead -- see handler.go).

Variables

View Source 
var (
	ErrQueueNotFound        = errors.New("azurequeue: queue not found")
	ErrQueueAlreadyExists   = errors.New("azurequeue: queue already exists with different metadata")
	ErrMessageNotFound      = errors.New("azurequeue: message not found")
	ErrPopReceiptMismatch   = errors.New("azurequeue: pop receipt mismatch")
	ErrInvalidQueryParam    = errors.New("azurequeue: invalid query parameter value")
	ErrOutOfRangeQueryParam = errors.New("azurequeue: query parameter value out of range")

	// ErrSnapshotQueueNull and ErrSnapshotMessageNull are returned by Restore
	// when a snapshot's "queues" map (or a queue's "Messages" slice) holds a
	// JSON null entry, which decodes to a nil pointer that would panic on
	// first dereference if stored as-is. See persistence.go.
	ErrSnapshotQueueNull   = errors.New("azurequeue: restore snapshot: queue is null")
	ErrSnapshotMessageNull = errors.New("azurequeue: restore snapshot: message is null")

	// ErrSnapshotQueueNameMismatch is returned by Restore when a snapshot's
	// "queues" map key differs from that entry's storedQueue.Name. Queue
	// operations (CreateQueue, DeleteQueue, PutMessage, ...) all key off the
	// map, while ListQueues reads Name -- a mismatch would let those two
	// views disagree about a queue's identity. See persistence.go.
	ErrSnapshotQueueNameMismatch = errors.New("azurequeue: restore snapshot: queue map key does not match Name")
)

Sentinel errors for Azure Queue Storage operations.

View Source 
var ErrNilAppContext = errors.New("azurequeue: nil app context")

ErrNilAppContext is returned when Init is called with a nil AppContext.

Functions

This section is empty.

Types

type ConfigProvider 

type ConfigProvider interface {
	GetAzureQueueSettings() Settings
}

ConfigProvider is a private interface to extract AzureQueue configuration from the abstract AppContext Config, mirroring services/azureblob.ConfigProvider.

type Handler 

type Handler struct {
	Backend StorageBackend

	// Endpoint is e.g. "http://127.0.0.1:10001" -- used to build
	// ServiceEndpoint in List Queues responses.
	Endpoint string
	// Port is the TCP port StartWorker binds. Set from Settings at Init time
	// (see provider.go); defaults to DefaultPort. Like services/azureblob,
	// this is a single fixed, protocol-conventional port -- there is no
	// fallback pool, so StartWorker fails fast if it's unavailable rather
	// than silently binding a different port.
	Port int
	// contains filtered or unexported fields
}

Handler is the Echo HTTP handler for Azure Queue Storage operations.

func NewHandler 

func NewHandler(backend StorageBackend) *Handler

NewHandler creates a new Azure Queue Handler. Port defaults to DefaultPort; callers (typically provider.go) override it from Settings.

func (*Handler) ExtractOperation 

func (h *Handler) ExtractOperation(c *echo.Context) string

ExtractOperation extracts the Azure Queue operation name from the request, for metrics labeling.

func (*Handler) ExtractResource 

func (h *Handler) ExtractResource(c *echo.Context) string

ExtractResource extracts the queue/message resource identifier from the request path, for metrics labeling.

func (*Handler) GetSupportedOperations 

func (h *Handler) GetSupportedOperations() []string

GetSupportedOperations returns the list of supported Azure Queue operations.

func (*Handler) Handler 

func (h *Handler) Handler() echo.HandlerFunc

Handler returns the Echo handler function for Azure Queue operations.

func (*Handler) MatchPriority 

func (h *Handler) MatchPriority() int

MatchPriority returns the routing priority for the AzureQueue handler. Irrelevant in practice since RouteMatcher never matches; 0 (lowest) is the safe default.

func (*Handler) Name 

func (h *Handler) Name() string

Name returns the service name.

func (*Handler) Reset 

func (h *Handler) Reset()

Reset clears all in-memory state from the backend. It is used by the POST /_gopherstack/reset endpoint for CI pipelines and rapid local development.

func (*Handler) Restore 

func (h *Handler) Restore(ctx context.Context, data []byte) error

Restore implements persistence.Persistable by delegating to the backend.

func (*Handler) RouteMatcher 

func (h *Handler) RouteMatcher() service.Matcher

RouteMatcher exists only to satisfy service.Registerable's interface contract: like services/azureblob, AzureQueue deliberately never matches on the shared AWS single-port Router. It runs on its own dedicated listener started by StartWorker (see provider.go for the full rationale). AzureQueue's Provider IS registered in cli.go's getMostRecentServiceProviders like every other service -- startBackgroundWorkers calls StartWorker via the service.BackgroundWorker interface regardless of routing, which is how the dedicated listener comes up. Only RouteMatcher itself is inert, kept so *Handler satisfies service.Registerable.

func (*Handler) Shutdown 

func (h *Handler) Shutdown(ctx context.Context)

Shutdown stops the dedicated Queue listener. A graceful Shutdown error (e.g. its context expiring before active connections finish) is logged and followed by Close, which forcibly closes the listener and any remaining idle/active connections; any Close error is logged too rather than leaving the listener to leak silently.

func (*Handler) Snapshot 

func (h *Handler) Snapshot(ctx context.Context) []byte

Snapshot implements persistence.Persistable by delegating to the backend.

func (*Handler) StartWorker 

func (h *Handler) StartWorker(ctx context.Context) error

StartWorker binds the dedicated Queue listener, starts serving on it, and -- if WithJanitor was called -- starts the background TTL-expiry sweep. See provider.go's Provider doc comment for why AzureQueue needs its own listener instead of registering into the shared AWS Router, and services/azureblob's StartWorker for the synchronous-bind rationale this mirrors exactly.

func (*Handler) WithJanitor 

func (h *Handler) WithJanitor(interval time.Duration) *Handler

WithJanitor attaches a background TTL-expiry janitor to the handler, mirroring services/xray's WithJanitor. If Backend is not an *InMemoryBackend the call is a no-op (a fake StorageBackend in tests has no sweepable state).

type InMemoryBackend 

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

InMemoryBackend implements StorageBackend using in-memory maps guarded by a single RWMutex. Shaped after services/azureblob's InMemoryBackend.

func NewInMemoryBackend 

func NewInMemoryBackend() *InMemoryBackend

NewInMemoryBackend creates a new empty InMemoryBackend.

func (*InMemoryBackend) ClearMessages 

func (b *InMemoryBackend) ClearMessages(queue string) error

ClearMessages removes every message from queue. Returns ErrQueueNotFound if the queue does not exist.

func (*InMemoryBackend) CreateQueue 

func (b *InMemoryBackend) CreateQueue(name string) (bool, error)

CreateQueue creates a new, empty queue. If a queue with the same name already exists, created is false and err is nil -- this backend has no queue metadata (see PARITY.md known gaps), so any pre-existing queue is by definition "the same metadata" and Create is idempotent (204), matching real Azure Queue Storage's semantics for a metadata-identical retry. ErrQueueAlreadyExists is reserved for a future metadata-bearing Create.

func (*InMemoryBackend) DeleteMessage 

func (b *InMemoryBackend) DeleteMessage(queue, messageID, popReceipt string) error

DeleteMessage removes a message identified by messageID from queue, after verifying popReceipt matches its current value. Returns ErrQueueNotFound, ErrMessageNotFound, or ErrPopReceiptMismatch as appropriate.

func (*InMemoryBackend) DeleteQueue 

func (b *InMemoryBackend) DeleteQueue(name string) error

DeleteQueue removes a queue and all of its messages. Returns ErrQueueNotFound if the queue does not exist.

func (*InMemoryBackend) GetMessages 

func (b *InMemoryBackend) GetMessages(
	queue string, numOfMessages int, visibilityTimeout time.Duration,
) ([]MessageInfo, error)

GetMessages dequeues up to numOfMessages visible, non-expired messages from queue: each returned message is hidden for visibilityTimeout (its NextVisibleTime advances, so subsequent Get/Peek calls skip it until that timeout elapses), assigned a fresh PopReceipt, and has its DequeueCount incremented. Returns ErrQueueNotFound if the queue does not exist.

func (*InMemoryBackend) ListQueues 

func (b *InMemoryBackend) ListQueues() []QueueInfo

ListQueues returns a snapshot of all queues, sorted by name (the order Azure's List Queues returns them in).

func (*InMemoryBackend) PeekMessages 

func (b *InMemoryBackend) PeekMessages(queue string, numOfMessages int) ([]MessageInfo, error)

PeekMessages returns up to numOfMessages visible, non-expired messages from queue without changing their visibility, PopReceipt, or DequeueCount. Returns ErrQueueNotFound if the queue does not exist.

func (*InMemoryBackend) PutMessage 

func (b *InMemoryBackend) PutMessage(queue, text string, visibilityTimeout, ttl time.Duration) (MessageInfo, error)

PutMessage enqueues a new message on queue. visibilityTimeout delays the message's initial visibility (0 means immediately visible); ttl bounds how long the message survives before the janitor sweeps it (0 uses DefaultMessageTTL). Returns ErrQueueNotFound if the queue does not exist.

func (*InMemoryBackend) Reset 

func (b *InMemoryBackend) Reset()

Reset clears all in-memory state. It is used by the POST /_gopherstack/reset endpoint for CI pipelines and rapid local development.

func (*InMemoryBackend) Restore 

func (b *InMemoryBackend) Restore(ctx context.Context, data []byte) error

Restore loads backend state from a JSON snapshot. It implements persistence.Persistable.

func (*InMemoryBackend) Snapshot 

func (b *InMemoryBackend) Snapshot(ctx context.Context) []byte

Snapshot serialises the backend state to JSON. It implements persistence.Persistable.

func (*InMemoryBackend) UpdateMessage 

func (b *InMemoryBackend) UpdateMessage(
	queue, messageID, popReceipt string, visibilityTimeout time.Duration, text *string,
) (MessageInfo, error)

UpdateMessage sets a new visibility timeout (and, if text is non-nil, replaces the message body) for messageID after verifying popReceipt matches, then rotates the PopReceipt. Returns ErrQueueNotFound, ErrMessageNotFound, or ErrPopReceiptMismatch as appropriate.

type Janitor 

type Janitor struct {
	Backend  *InMemoryBackend
	Interval time.Duration
}

Janitor is the Azure Queue background worker that deletes messages whose message TTL (see PutMessage's ttl parameter / DefaultMessageTTL) has elapsed, mirroring services/sqs's Janitor (SQS's MessageRetentionPeriod sweep).

func NewJanitor 

func NewJanitor(backend *InMemoryBackend, interval time.Duration) *Janitor

NewJanitor creates a new Azure Queue Janitor for the given backend.

func (*Janitor) Run 

func (j *Janitor) Run(ctx context.Context)

Run runs the janitor loop until ctx is cancelled.

type MessageInfo 

type MessageInfo struct {
	InsertionTime   time.Time
	ExpirationTime  time.Time
	TimeNextVisible time.Time
	ID              string
	PopReceipt      string
	Text            string
	DequeueCount    int64
}

MessageInfo is a read-only snapshot of a message, returned by the StorageBackend message accessors. PopReceipt is empty for Peek Messages (peeking never assigns one -- see AZURE.md section 2), and Text is only populated for Get/Peek (Put/Update return no body, only metadata, mirroring real Azure Queue Storage).

type Provider 

type Provider struct{}

Provider implements service.Provider for the Azure Queue Storage service.

Like services/azureblob, AzureQueue does not register a RouteMatcher into the shared AWS single-port Router: Azure Queue's path shape (/<account>/<queue>[/messages[/<id>]]) has no service-identifying header the way AWS's X-Amz-Target does, and shares the same /<account>/<resource> shape as Azure Blob and Table, so multiplexing it onto the shared port (or even onto Azure Blob's own dedicated port) risks exactly the collision the router avoids by construction for AWS services (see AZURE.md section 4). Instead the returned Handler implements service.BackgroundWorker and stands up its own dedicated *echo.Echo/*http.Server, listening on a fixed, protocol-conventional port (Azurite's own Queue port, 10001). It is registered in cli.go's getMostRecentServiceProviders like every other provider; only its RouteMatcher (which always returns false) is inert.

func (*Provider) Init 

func (p *Provider) Init(ctx *service.AppContext) (service.Registerable, error)

Init initializes the AzureQueue service backend and handler. The configured port (Settings.Port, default DefaultPort) is only recorded here; the actual TCP bind happens synchronously in Handler.StartWorker, so a port-in-use failure is returned to the caller directly instead of being discovered later from a background goroutine.

func (*Provider) Name 

func (p *Provider) Name() string

Name returns the service provider name.

type QueueInfo 

type QueueInfo struct {
	CreatedAt time.Time
	Name      string
}

QueueInfo is a read-only snapshot of a queue's metadata, returned by StorageBackend.ListQueues. It intentionally excludes the queue's message slice so callers cannot mutate backend state through it.

type Settings 

type Settings struct {
	// Port is the fixed TCP port for the dedicated Queue listener. See
	// handler.go's StartWorker for what happens when it's unavailable
	// (fails fast; no fallback pool, matching services/azureblob).
	Port int `` //nolint:lll // config struct tags are intentionally verbose
	/* 178-byte string literal not displayed */
}

Settings holds service-level configuration for the Azure Queue backend. Fields are picked up by the Kong CLI parser when this struct is embedded in the root CLI command (see cli.go's CLI.AzureQueue field), mirroring services/azureblob's Settings pattern.

func DefaultSettings 

func DefaultSettings() Settings

DefaultSettings returns the default Settings. Used when no ConfigProvider is available at init time (e.g. tests constructing a Provider directly).

type StorageBackend 

type StorageBackend interface {
	CreateQueue(name string) (created bool, err error)
	DeleteQueue(name string) error
	ListQueues() []QueueInfo

	PutMessage(queue, text string, visibilityTimeout, ttl time.Duration) (MessageInfo, error)
	// GetMessages and PeekMessages return ErrOutOfRangeQueryParam if
	// numOfMessages is outside [MinNumOfMessages, MaxNumOfMessages] -- a
	// defense-in-depth check against direct callers that bypass the
	// handler's own range validation (see messages.go's parseNumOfMessages),
	// since an unvalidated negative value would otherwise panic on
	// allocation.
	GetMessages(queue string, numOfMessages int, visibilityTimeout time.Duration) ([]MessageInfo, error)
	PeekMessages(queue string, numOfMessages int) ([]MessageInfo, error)
	DeleteMessage(queue, messageID, popReceipt string) error
	UpdateMessage(
		queue, messageID, popReceipt string, visibilityTimeout time.Duration, text *string,
	) (MessageInfo, error)
	ClearMessages(queue string) error

	// Reset clears all in-memory state. Used by the
	// POST /_gopherstack/reset endpoint for CI pipelines and rapid local development.
	Reset()
}

StorageBackend defines the interface for an Azure Queue Storage backend. Shaped after services/azureblob's StorageBackend: a narrow, testable seam between the wire handler and storage, so handler tests can substitute a fake.

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.