azureservicebus

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: 26 Imported by: 0

README

Azure Service Bus

Parity grade: B · SDK azure-sdk-for-go/sdk/messaging/azservicebus@v1.10.0 · last audited 2026-09-07 (a5f048bc5)

Coverage

Metric Value
PARITY entries audited 17 (16 ok, 1 partial)
Feature families 9 (5 ok, 1 partial, 1 gap, 2 deferred)
Known gaps 4
Deferred items 2
Resource leaks clean
Known gaps
  • azservicebus (the only azure-sdk-for-go Service Bus client) is AMQP-only and cannot be pointed at this REST emulator -- see families.sdk_compat. This is a genuine SDK-compatibility limitation, not an MVP scope cut; a real AMQP 1.0 listener would be required to support it, which AZURE.md section 9's M5 rationale explicitly defers.
  • No SQL-filter rule evaluation for subscriptions -- every subscription is effectively TrueFilter (match-all). See families.filter_evaluation.
  • No sessions (ordered/exclusive per-SessionId delivery) -- SessionId round-trips but has no locking/FIFO semantics. See families.sessions_and_amqp.
  • No explicit client-initiated DeadLetter operation. Research into real Service Bus's REST wire contract did not turn up authoritative confirmation of the BrokerProperties-DeadLetterReason-on-the-Abandon-PUT shape (the current official REST API reference for the Unlock Message operation, which shares that same PUT .../messages// URI, documents it as taking no request body at all) -- rather than inventing an unconfirmed endpoint, this gap is left as-is (still only reached via the $DeadLetterQueue path or the Janitor's automatic sweep) pending a human decision on how to proceed. See the DeadLetter research note in the Notes section below. All gaps above are intentional MVP scope per AZURE.md section 9's M5 entry (or, for DeadLetter, a deliberate pause pending human input), not oversights.
Deferred
  • Initial implementation pass (2026-09-06): seeded this service from scratch per AZURE.md M5 (see AZURE.md section 9). Structurally mirrors services/azurequeue's pop-receipt/visibility-timeout pattern for message locks and services/sns's topic/subscription fan-out bookkeeping; no prior audit history to reconcile.
  • Follow-up pass (2026-09-07): closed five bounded parity gaps -- per-entity LockDuration/MaxDeliveryCount/DefaultMessageTimeToLive, PeekLock long-poll, full Atom+XML entity-kind parsing, and Get/List operations. Left the explicit client-initiated DeadLetter gap as-is after research found the wire shape unconfirmed (see gaps above); SQL-filter evaluation, sessions, and the AMQP/REST SDK-compatibility gap remain deferred per AZURE.md section 9.

More

Documentation

Overview

Package azureservicebus provides a local, in-memory emulation of Azure Service Bus's Brokered Messaging REST API: queue and topic/subscription CRUD, and the full send/peek-lock/complete/abandon/dead-letter message lifecycle over HTTP+SharedAccessSignature auth. It deliberately does NOT implement AMQP 1.0 -- sessions and full AMQP compatibility are out of scope for this MVP. See AZURE.md section 9 (M5) and PARITY.md for scope and known gaps.

Index

Constants

View Source 
const (
	// DefaultLockDuration is applied to a PeekLock when the caller specifies
	// no timeout query parameter.
	DefaultLockDuration = 60 * time.Second
	// DefaultMessageTTL is applied to Send when the caller supplies no
	// TimeToLive in BrokerProperties.
	DefaultMessageTTL = 14 * 24 * time.Hour
	// MaxDeliveryCount is how many times a message may be
	// peek-locked-and-abandoned before it is automatically moved to its
	// entity's dead-letter sub-queue, matching real Service Bus's own
	// default MaxDeliveryCount.
	MaxDeliveryCount = 10
	// MaxLockDuration is real Service Bus's documented maximum LockDuration
	// for a queue or subscription (its default is 1 minute; 5 minutes is the
	// upper bound a CreateQueue/CreateSubscription request may configure --
	// see https://learn.microsoft.com/en-us/azure/service-bus-messaging/message-transfers-locks-settlement#peeklock,
	// which documents "the maximum value is 5 minutes"). handler.go's
	// validateEntityConfig rejects a create request specifying more than
	// this with 400 Bad Request, matching real Service Bus's own behavior.
	MaxLockDuration = 5 * time.Minute
)

Default and bound values, matching real Service Bus's documented defaults.

View Source 
const DefaultKeyName = "RootManageSharedAccessKey"

DefaultKeyName is gopherstack's fixed root shared-access-policy key name, mirroring the real Service Bus default policy name created on every new namespace.

View Source 
const DefaultKeyValue = "2R1W2VORtFi9HrmRQ1Gxp7xbySq7W0FAs2BvTZDdXeo="

DefaultKeyValue is gopherstack's fixed, publicly published development SAS key (base64-encoded), analogous to pkgs/azureauth.DefaultAccountKey / Azurite's devstoreaccount1 key: a fixed dev secret so azure-sdk-for-go's default connection-string shape works unmodified out of the box, and so WithSASValidation has something deterministic to check against by default.

View Source 
const DefaultNamespace = "sbemulatorns"

DefaultNamespace is gopherstack's fixed, Azurite-style default Service Bus namespace name. Real Service Bus namespaces are addressed as "<namespace>.servicebus.windows.net"; gopherstack has no DNS story for that, so this constant exists purely so a connection string built from it (Endpoint=sb://<DefaultNamespace>.servicebus.windows.net/;SharedAccessKeyName=...) has a plausible, stable shape for documentation/examples. The actual listener is addressed by host:port (see settings.go's DefaultPort), not by this name.

View Source 
const DefaultPort = 10003

DefaultPort is Azure Service Bus's fixed, gopherstack-chosen TCP port. This follows the same pattern as services/azureblob's DefaultPort (10000), services/azurequeue's DefaultPort (10001), and services/azuretable's DefaultPort (10002): 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. Unlike those three, there is no Azurite Service Bus emulator to mirror -- Azurite doesn't implement Service Bus -- so 10003 is simply the next available slot after Table's 10002 in gopherstack's own numbering convention. See AZURE.md section 9 (M5) for the full rationale.

View Source 
const MaxPeekLockWaitTimeout = 30 * time.Second

MaxPeekLockWaitTimeout is the ceiling PeekLock's long-poll "?timeout=" query parameter is clamped to. Real Service Bus documents 30 seconds as its own maximum long-poll/operation timeout, and gopherstack matches that value here as a deliberate cap on server-side resource use per in-flight long-poll request (one goroutine and one open connection held for up to this long). The wait is bounded by this cap plus request-context cancellation on client disconnect -- NOT by any http.Server timeout: azureServiceBusReadTimeout only bounds reading the request itself (headers/body), not handler execution time, and this server sets no WriteTimeout or handler deadline, so a long-poll handler blocking well past 60s would not be torn down by the server. See PARITY.md.

Variables

View Source 
var (
	ErrQueueNotFound        = errors.New("azureservicebus: queue not found")
	ErrQueueAlreadyExists   = errors.New("azureservicebus: queue already exists")
	ErrTopicNotFound        = errors.New("azureservicebus: topic not found")
	ErrTopicAlreadyExists   = errors.New("azureservicebus: topic already exists")
	ErrSubscriptionNotFound = errors.New("azureservicebus: subscription not found")
	ErrSubscriptionExists   = errors.New("azureservicebus: subscription already exists")
	ErrMessageNotFound      = errors.New("azureservicebus: message not found")
	ErrLockTokenMismatch    = errors.New("azureservicebus: lock token mismatch")
	ErrMessageNotLocked     = errors.New("azureservicebus: message is not locked")
	ErrInvalidEntityRef     = errors.New("azureservicebus: invalid entity reference")
	ErrInvalidEntityConfig  = errors.New("azureservicebus: invalid entity configuration")

	// ErrSnapshotNull* are returned by Restore when a snapshot map/slice holds
	// a JSON null entry, which decodes to a nil pointer that would panic on
	// first dereference if stored as-is. Mirrors services/azurequeue's
	// identical family of snapshot-validation errors.
	ErrSnapshotQueueNull        = errors.New("azureservicebus: restore snapshot: queue is null")
	ErrSnapshotTopicNull        = errors.New("azureservicebus: restore snapshot: topic is null")
	ErrSnapshotSubscriptionNull = errors.New("azureservicebus: restore snapshot: subscription is null")
	ErrSnapshotMessageNull      = errors.New("azureservicebus: restore snapshot: message is null")
)

Sentinel errors for Azure Service Bus operations.

View Source 
var ErrBadPath = errors.New("azureservicebus: unrecognized path")

ErrBadPath is returned by parseRequestPath for a path that doesn't match any recognized Service Bus REST shape.

View Source 
var ErrInvalidISO8601Duration = errors.New("azureservicebus: invalid ISO 8601 duration")

ErrInvalidISO8601Duration is returned by parseISO8601Duration for a string that doesn't match the supported PnDTnHnMnS/PTnHnMnS subset.

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

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

Functions

func SignSAS 

func SignSAS(resource string, expiry int64, keyValue string) string

SignSAS computes the Service Bus SAS signature for resource+expiry using keyValue (base64-encoded), matching the real algorithm: HMAC-SHA256 over "<url-encoded resource>\n<expiry>", base64-encoded.

func VerifySAS 

func VerifySAS(auth Authorization, keyValue string, now time.Time) bool

VerifySAS cryptographically verifies auth against keyValue, additionally rejecting an expired token (se in the past relative to now). Callers only invoke this when opting into validation (see Handler.checkAuth / WithSASValidation) -- structural parsing always accepts the header regardless of what this returns.

Types

type Authorization 

type Authorization struct {
	Resource  string // sr: URL-encoded resource URI the token authorizes
	Signature string // sig: base64 HMAC-SHA256
	KeyName   string // skn: shared-access-policy key name
	Expiry    int64  // se: unix seconds the token expires at
}

Authorization holds the structurally-parsed fields of a Service Bus "Authorization: SharedAccessSignature sr=<resource>&sig=<hmac>&se=<expiry>&skn=<keyname>" header. Parsing never fails on a bad/missing signature -- only on a header that isn't shaped like a SAS token at all -- mirroring pkgs/azureauth's "structural parse always succeeds, verification is opt-in" philosophy (see services/s3's PresignSecret and Blob/Queue/Table's WithSharedKeyValidation).

func ParseSASAuthorization 

func ParseSASAuthorization(header string) (Authorization, bool)

ParseSASAuthorization structurally parses a Service Bus SAS Authorization header value. It always extracts whatever key=value pairs are present; ok is false only when the header is missing the scheme prefix entirely.

type ConfigProvider 

type ConfigProvider interface {
	GetAzureServiceBusSettings() Settings
}

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

type EntityConfig 

type EntityConfig struct {
	LockDuration      time.Duration
	DefaultMessageTTL time.Duration
	MaxDeliveryCount  int
}

EntityConfig holds the per-entity configuration properties real Service Bus accepts on CreateQueue/CreateTopic/CreateSubscription (LockDuration, MaxDeliveryCount, DefaultMessageTimeToLive), parsed from the Atom+XML create-request body (see atom.go). Which fields are meaningful depends on the entity kind: a queue honors all three; a topic honors only DefaultMessageTTL (applied as the message-TTL cap at Send time, since a topic never itself holds messages); a subscription honors LockDuration and MaxDeliveryCount. A zero-valued field falls back to the corresponding package-level default (DefaultLockDuration/MaxDeliveryCount/ DefaultMessageTTL) -- see the lockDuration/maxDeliveryCount/ defaultMessageTTL accessor methods.

type EntityRef 

type EntityRef struct {
	Queue        string
	Topic        string
	Subscription string
}

EntityRef identifies a brokered-messaging entity that can hold messages: either a queue, or one specific subscription of a topic (a topic itself never holds messages directly -- sending to a topic fans the message out to every one of its subscriptions' own message lists). Exactly one of Queue or (Topic, Subscription) is set.

func (EntityRef) IsQueue 

func (r EntityRef) IsQueue() bool

IsQueue reports whether ref addresses a queue rather than a subscription.

type Handler 

type Handler struct {
	Backend StorageBackend

	// Endpoint is e.g. "http://127.0.0.1:10003".
	Endpoint string
	// SASKeyValue is the base64 key WithSASValidation checks signatures
	// against. Defaults to DefaultKeyValue when empty.
	SASKeyValue string
	// Port is the TCP port StartWorker binds. Set from Settings at Init time
	// (see provider.go); defaults to DefaultPort. Single fixed,
	// protocol-conventional port -- no fallback pool.
	Port int
	// ValidateSAS opts checkAuth into cryptographic SAS verification. See
	// WithSASValidation.
	ValidateSAS bool
	// contains filtered or unexported fields
}

Handler is the Echo HTTP handler for Azure Service Bus operations.

func NewHandler 

func NewHandler(backend StorageBackend) *Handler

NewHandler creates a new Azure Service Bus 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 Service Bus operation name from the request, for metrics labeling.

func (*Handler) ExtractResource 

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

ExtractResource extracts the entity resource identifier from the request path, for metrics labeling.

func (*Handler) GetSupportedOperations 

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

GetSupportedOperations returns the list of supported Azure Service Bus operations.

func (*Handler) Handler 

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

Handler returns the Echo handler function for Azure Service Bus operations.

func (*Handler) MatchPriority 

func (h *Handler) MatchPriority() int

MatchPriority returns the routing priority for the AzureServiceBus handler. Irrelevant in practice since RouteMatcher never matches.

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. 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: AzureServiceBus never matches on the shared AWS single-port Router (see Provider's doc comment).

func (*Handler) Shutdown 

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

Shutdown stops the dedicated Service Bus listener. Mirrors services/azurequeue's Shutdown.

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 Service Bus listener, starts serving on it, and -- if WithJanitor was called -- starts the background lock-expiry/dead-letter sweep. Mirrors services/azurequeue's StartWorker exactly.

func (*Handler) WithJanitor 

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

WithJanitor attaches a background lock-expiry/dead-letter janitor to the handler, mirroring services/azurequeue's WithJanitor. If Backend is not an *InMemoryBackend the call is a no-op.

func (*Handler) WithSASValidation 

func (h *Handler) WithSASValidation(key string) *Handler

WithSASValidation enables cryptographic verification of SAS (SharedAccessSignature) Authorization headers, checking each signature against the given key (base64-encoded). A blank key defaults to DefaultKeyValue. Mirrors services/s3's WithPresignValidation and Blob/Queue/Table's WithSharedKeyValidation: when never called, SAS headers are parsed structurally only, never cryptographically rejected.

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/azurequeue's InMemoryBackend.

func NewInMemoryBackend 

func NewInMemoryBackend() *InMemoryBackend

NewInMemoryBackend creates a new empty InMemoryBackend.

func (*InMemoryBackend) Abandon 

func (b *InMemoryBackend) Abandon(ref EntityRef, deadLetter bool, messageID, lockToken string) error

Abandon releases a locked message's lock (making it immediately available again), after verifying lockToken matches. If DeliveryCount has reached ref's configured MaxDeliveryCount, the message is moved to the entity's dead-letter sub-queue instead of being made available, matching real Service Bus's automatic dead-lettering on delivery-count exhaustion. Either outcome -- released back to the live list, or moved to the dead-letter sub-queue -- makes a message newly visible somewhere on ref, so both wake any PeekLockWait waiter on ref (the same notify channel is shared by the live and dead-letter lists; a waiter on the list that didn't change simply re-checks and goes back to sleep, a harmless spurious wakeup).

func (*InMemoryBackend) Complete 

func (b *InMemoryBackend) Complete(ref EntityRef, deadLetter bool, messageID, lockToken string) error

Complete permanently removes a locked message identified by messageID from ref (or its dead-letter sub-queue), after verifying lockToken matches.

func (*InMemoryBackend) CreateQueue 

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

CreateQueue creates a new, empty queue with the given configuration (see EntityConfig; cfg is variadic so pre-existing callers passing only a name keep compiling -- see the StorageBackend interface's doc comment). If a queue with the same name already exists, created is false, err is nil, and its existing configuration is left untouched (idempotent, mirroring services/azurequeue's CreateQueue). ErrQueueAlreadyExists is reserved for a future strict-create variant.

func (*InMemoryBackend) CreateSubscription 

func (b *InMemoryBackend) CreateSubscription(topic, name string, cfg ...EntityConfig) (bool, error)

CreateSubscription creates a new, empty subscription of topic with the given configuration (see EntityConfig -- LockDuration and MaxDeliveryCount are meaningful for a subscription; cfg is variadic so pre-existing callers passing only topic/name keep compiling). If a subscription with the same name already exists on that topic, created is false and err is nil (idempotent, mirroring CreateQueue/CreateTopic). Returns ErrTopicNotFound if the topic does not exist.

Real Service Bus subscription creation accepts an optional SQL filter rule in the request body; this MVP deliberately does not parse or store it -- every subscription behaves as if it had Service Bus's default "match all" rule (TrueFilter). See PARITY.md's filter-evaluation gap and handler.go's createSubscription, which discards the request body's filter rule after extracting any EntityConfig properties it also carries.

func (*InMemoryBackend) CreateTopic 

func (b *InMemoryBackend) CreateTopic(name string, cfg ...EntityConfig) (bool, error)

CreateTopic creates a new, empty topic (no subscriptions) with the given configuration (see EntityConfig -- only DefaultMessageTTL is meaningful for a topic; cfg is variadic so pre-existing callers passing only a name keep compiling). If a topic with the same name already exists, created is false and err is nil, mirroring CreateQueue's idempotent-retry semantics.

func (*InMemoryBackend) DeleteQueue 

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

DeleteQueue removes a queue and all of its messages (including any dead-lettered ones). Returns ErrQueueNotFound if the queue does not exist.

func (*InMemoryBackend) DeleteSubscription 

func (b *InMemoryBackend) DeleteSubscription(topic, name string) error

DeleteSubscription removes a subscription and all of its messages (including any dead-lettered ones). Returns ErrTopicNotFound or ErrSubscriptionNotFound as appropriate.

func (*InMemoryBackend) DeleteTopic 

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

DeleteTopic removes a topic, all of its subscriptions, and every message held by those subscriptions. Returns ErrTopicNotFound if the topic does not exist.

func (*InMemoryBackend) GetQueueInfo 

func (b *InMemoryBackend) GetQueueInfo(name string) (QueueInfo, error)

GetQueueInfo returns name's metadata. Returns ErrQueueNotFound if it does not exist.

func (*InMemoryBackend) GetSubscriptionInfo 

func (b *InMemoryBackend) GetSubscriptionInfo(topic, name string) (SubscriptionInfo, error)

GetSubscriptionInfo returns name's metadata within topic. Returns ErrTopicNotFound or ErrSubscriptionNotFound as appropriate.

func (*InMemoryBackend) GetTopicInfo 

func (b *InMemoryBackend) GetTopicInfo(name string) (TopicInfo, error)

GetTopicInfo returns name's metadata. Returns ErrTopicNotFound if it does not exist.

func (*InMemoryBackend) ListQueues 

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

ListQueues returns every queue's metadata, sorted by name.

func (*InMemoryBackend) ListSubscriptions 

func (b *InMemoryBackend) ListSubscriptions(topic string) ([]SubscriptionInfo, error)

ListSubscriptions returns every subscription of topic's metadata, sorted by name. Returns ErrTopicNotFound if topic does not exist.

func (*InMemoryBackend) ListTopics 

func (b *InMemoryBackend) ListTopics() []TopicInfo

ListTopics returns every topic's metadata, sorted by name.

func (*InMemoryBackend) PeekLock 

func (b *InMemoryBackend) PeekLock(ref EntityRef, deadLetter bool, lockDuration time.Duration) (MessageInfo, error)

PeekLock returns the oldest visible (unlocked, non-expired) message on ref -- or its dead-letter sub-queue, if deadLetter is true -- locking it for lockDuration and incrementing its delivery count. Returns ErrMessageNotFound if none is available. See the StorageBackend doc comment for lockDuration <= 0's fallback behavior.

func (*InMemoryBackend) PeekLockWait 

func (b *InMemoryBackend) PeekLockWait(
	ctx context.Context, ref EntityRef, deadLetter bool, lockDuration, timeout time.Duration,
) (MessageInfo, error)

PeekLockWait is PeekLock's long-poll variant. See the StorageBackend doc comment. Mirrors services/sqs's ReceiveMessage/pollReceive/receiveOnce long-poll shape (a broadcast notify channel plus a 1-second recheck-timer backstop), adapted to take a context so a disconnecting client releases the waiting goroutine immediately -- a deliberate improvement over the SQS precedent, which has no ctx parameter (see PARITY.md). Deadline/backstop timing uses the real wall clock (time.Now/time.Timer), not the backend's mockable nowFunc -- matching services/sqs's identical choice and making this method exercisable under testing/synctest's fake clock.

func (*InMemoryBackend) QueueExists 

func (b *InMemoryBackend) QueueExists(name string) bool

QueueExists reports whether a queue named name exists.

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) Send 

func (b *InMemoryBackend) Send(ref EntityRef, msg NewMessage) (MessageInfo, error)

Send enqueues msg on ref. See the StorageBackend.Send doc comment for the queue-vs-topic fan-out distinction and the per-entity TTL cap.

func (*InMemoryBackend) Snapshot 

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

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

func (*InMemoryBackend) SubscriptionExists 

func (b *InMemoryBackend) SubscriptionExists(topic, name string) bool

SubscriptionExists reports whether topic has a subscription named name. Returns false (not an error) if topic itself does not exist.

func (*InMemoryBackend) TopicExists 

func (b *InMemoryBackend) TopicExists(name string) bool

TopicExists reports whether a topic named name exists.

type Janitor 

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

Janitor is the Azure Service Bus background worker responsible for the two things real Service Bus does automatically over time: releasing a peek-locked message whose lock has expired without being Completed or Abandoned (making it available for redelivery, or dead-lettering it if that expiry pushed its delivery count past MaxDeliveryCount), and moving a TTL-expired message to its entity's dead-letter sub-queue. Mirrors services/azurequeue's Janitor/TTL-sweep shape.

func NewJanitor 

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

NewJanitor creates a new Azure Service Bus 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 {
	EnqueuedTime   time.Time
	LockedUntil    time.Time
	CustomHeaders  map[string]string
	ContentType    string
	Label          string
	CorrelationID  string
	MessageID      string
	ReplyTo        string
	SessionID      string
	LockToken      string
	Body           []byte
	SequenceNumber int64
	DeliveryCount  int64
}

MessageInfo is the public, read-only snapshot of a stored message returned by Send/PeekLock. LockToken/LockedUntil are populated only for a peek-locked read; Send returns them zero.

type NewMessage 

type NewMessage struct {
	CustomHeaders map[string]string
	ContentType   string
	Label         string
	CorrelationID string
	MessageID     string
	ReplyTo       string
	SessionID     string
	Body          []byte
	TimeToLive    time.Duration
}

NewMessage is the input shape for sending a message, built from the incoming request's body and BrokerProperties header (see message_ops.go).

type Provider 

type Provider struct{}

Provider implements service.Provider for the Azure Service Bus service.

Like services/azureblob/azurequeue/azuretable, AzureServiceBus does not register a RouteMatcher into the shared AWS single-port Router: its path shape (/<queue-or-topic>[/subscriptions/<name>][/messages[/...]]) has no service-identifying header the way AWS's X-Amz-Target does, and multiplexing it onto a shared 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 port (DefaultPort, 10003). 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 AzureServiceBus 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
	LockDuration      time.Duration
	DefaultMessageTTL time.Duration
	MaxDeliveryCount  int
}

QueueInfo is the read-only metadata snapshot returned by GetQueueInfo/ ListQueues.

type Settings 

type Settings struct {
	// Port is the fixed TCP port for the dedicated Service Bus listener. See
	// handler.go's StartWorker for what happens when it's unavailable (fails
	// fast; no fallback pool, matching services/azureblob/azurequeue/azuretable).
	Port int `` //nolint:lll // config struct tags are intentionally verbose
	/* 189-byte string literal not displayed */
	// ValidateSAS opts the handler into cryptographic verification of SAS
	// (SharedAccessSignature) tokens against DevKeyValue (or a caller-supplied
	// key via WithSASValidation), mirroring services/s3's
	// --validate-sigv4/PresignSecret opt-in pattern and Blob/Queue/Table's
	// WithSharedKeyValidation. Off by default: SAS tokens are always
	// structurally parsed (key name + resource scope extracted) but not
	// cryptographically checked unless this is set.
	ValidateSAS bool `` //nolint:lll // config struct tags are intentionally verbose
	/* 247-byte string literal not displayed */
}

Settings holds service-level configuration for the Azure Service Bus 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.AzureServiceBus field), mirroring services/azurequeue's and services/azuretable'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 creates a queue. cfg is variadic purely so every
	// pre-existing call site that only ever passed a name keeps compiling;
	// at most cfg[0] is used (see EntityConfig, firstConfig).
	CreateQueue(name string, cfg ...EntityConfig) (created bool, err error)
	DeleteQueue(name string) error
	QueueExists(name string) bool
	GetQueueInfo(name string) (QueueInfo, error)
	ListQueues() []QueueInfo

	// CreateTopic creates a topic. Only cfg's DefaultMessageTTL field is
	// meaningful (see EntityConfig's doc comment); see CreateQueue's doc
	// comment for why cfg is variadic.
	CreateTopic(name string, cfg ...EntityConfig) (created bool, err error)
	DeleteTopic(name string) error
	TopicExists(name string) bool
	GetTopicInfo(name string) (TopicInfo, error)
	ListTopics() []TopicInfo

	// CreateSubscription creates a subscription of topic. The filter rule
	// (if any) is accepted structurally but not stored/evaluated -- every
	// rule is treated as match-all (see PARITY.md's filter-evaluation gap).
	// Only cfg's LockDuration/MaxDeliveryCount fields are meaningful; see
	// CreateQueue's doc comment for why cfg is variadic.
	CreateSubscription(topic, name string, cfg ...EntityConfig) (created bool, err error)
	DeleteSubscription(topic, name string) error
	SubscriptionExists(topic, name string) bool
	GetSubscriptionInfo(topic, name string) (SubscriptionInfo, error)
	ListSubscriptions(topic string) ([]SubscriptionInfo, error)

	// Send enqueues msg on ref. For a queue ref it is appended to that
	// queue's own message list. For a topic ref (Topic set, Subscription
	// empty) it is fanned out: an independent copy is appended to every one
	// of that topic's subscriptions' own message lists, matching real
	// Service Bus's one-to-many topic delivery (see AZURE.md section 9's M5
	// entry and services/sns's topic/subscription fan-out as the structural
	// reference). Returns ErrTopicNotFound if the topic has no
	// subscriptions registered as an error -- fan-out to zero subscriptions
	// is not an error, matching real Service Bus (the message is simply
	// dropped, as no subscription exists to receive it). msg.TimeToLive, if
	// set, is capped at the target entity's configured
	// DefaultMessageTimeToLive (real Service Bus semantics); if unset, the
	// entity's DefaultMessageTimeToLive is used outright. Send also wakes
	// any goroutine blocked in PeekLockWait on the affected entity/entities.
	Send(ref EntityRef, msg NewMessage) (MessageInfo, error)

	// PeekLock performs a destructive read: it returns the oldest visible,
	// non-expired message on ref (or its dead-letter sub-queue, if
	// deadLetter is true), locking it for lockDuration and incrementing its
	// delivery count. lockDuration <= 0 resolves to ref's own configured
	// LockDuration (falling back to DefaultLockDuration if ref has none
	// configured) rather than a caller-supplied value -- see EntityConfig.
	// Returns ErrMessageNotFound if none is available.
	PeekLock(ref EntityRef, deadLetter bool, lockDuration time.Duration) (MessageInfo, error)

	// PeekLockWait is PeekLock's long-poll variant: if no message is
	// immediately visible, it waits up to timeout for one to arrive (a
	// message becoming visible via Send, Abandon's release path, or the
	// Janitor's lock-release path all wake a waiter), or until ctx is
	// cancelled, whichever comes first. timeout <= 0 behaves exactly like an
	// immediate PeekLock call. Callers are expected to have already clamped
	// timeout to a sane maximum (see handler.go's MaxPeekLockWaitTimeout);
	// PeekLockWait itself does not enforce a cap.
	PeekLockWait(
		ctx context.Context,
		ref EntityRef,
		deadLetter bool,
		lockDuration, timeout time.Duration,
	) (MessageInfo, error)

	// Complete permanently removes a locked message identified by messageID,
	// after verifying lockToken matches. Returns ErrLockTokenMismatch or
	// ErrMessageNotFound as appropriate.
	Complete(ref EntityRef, deadLetter bool, messageID, lockToken string) error

	// Abandon releases a locked message's lock (making it immediately
	// available again) after verifying lockToken matches, without altering
	// its position. If the message's delivery count has reached ref's
	// configured MaxDeliveryCount (see EntityConfig) it is moved to the
	// dead-letter sub-queue instead.
	Abandon(ref EntityRef, deadLetter bool, messageID, lockToken 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 Service Bus backend. Shaped after services/azurequeue's StorageBackend: a narrow, testable seam between the wire handler and storage, so handler tests can substitute a fake.

type SubscriptionInfo 

type SubscriptionInfo struct {
	CreatedAt        time.Time
	Name             string
	LockDuration     time.Duration
	MaxDeliveryCount int
}

SubscriptionInfo is the read-only metadata snapshot returned by GetSubscriptionInfo/ListSubscriptions.

type TopicInfo 

type TopicInfo struct {
	CreatedAt         time.Time
	Name              string
	DefaultMessageTTL time.Duration
}

TopicInfo is the read-only metadata snapshot returned by GetTopicInfo/ ListTopics.

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.