Help us shape the future of Go! Take the 2026 Go Developer Survey

azureservicebus

package
v1.9.0 Latest Latest
Warning

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

 Go to latest Published: Oct 27, 2026 License: MIT Imports: 36 Imported by: 0

README

Azure Service Bus

Parity grade: B · SDK azure-sdk-for-go/sdk/messaging/azservicebus@v1.11.0 · last audited 2026-10-20 (a5f048bc5)

Coverage

Metric Value
PARITY entries audited 25 (22 ok, 3 partial)
Feature families 10 (8 ok, 1 partial, 1 deferred)
Known gaps 3
Deferred items 4
Resource leaks clean
Known gaps
  • AMQP bodies are flattened: amqp-value and sequence bodies are stored as bytes and returned as a data section; application property types survive in memory but are strings after a persistence restore or when set through REST. See families.amqp_transport.
  • No SQL-filter rule evaluation for subscriptions -- every subscription is effectively TrueFilter (match-all). See families.filter_evaluation.
  • Client dead-letter and PeekMessages use emulator-extension REST shapes (BrokerProperties DeadLetterReason on the Unlock PUT; GET .../messages); the AMQP plane now does both natively. 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 remained deferred per AZURE.md section 9.
  • Sessions pass (2026-10-20): RequiresSession entities, session-filter receivers with locks, session state, renew-session-lock, FIFO per session, persisted in the snapshot.
  • AMQP 1.0 pass (2026-10-20): added pkgs/amqp1 and the AMQP listener (plain 5672, TLS 5671) over the shared backend, with Defer, ReceiveDeferred, RenewLock, scheduled enqueue/cancel and lock-token lookup added to the backend; verified with the real azservicebus client.

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, plus an AMQP 1.0 plane (pkgs/amqp1) over the same backend for the official SDKs, including sessions over AMQP. See PARITY.md for scope and known gaps.

Index

Constants

View Source 
const (
	DefaultAMQPPort    = 5672
	DefaultAMQPTLSPort = 5671
)

Default AMQP 1.0 ports (the IANA assignments Service Bus uses).

View Source 
const (
	ReasonMaxDeliveryCountExceeded = "MaxDeliveryCountExceeded"
	ReasonTTLExpired               = "TTLExpiredException"
)

Auto dead-letter reasons real Service Bus stamps on system dead-lettering.

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.

View Source 
const (

	// MaxSessionStateBytes is Service Bus's documented session state limit.
	MaxSessionStateBytes = 256 * 1024
)

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")
	ErrDeadLetterSource     = errors.New("azureservicebus: cannot dead-letter a dead-lettered message")

	ErrSessionIDRequired = errors.New("the SessionId was not set on a message, and it cannot be sent to the entity. " +
		"Entities that have session support enabled can only receive messages that have the SessionId set to a valid value")
	ErrSessionRequired = errors.New("it is not possible for an entity that requires sessions " +
		"to create a non-sessionful message receiver")
	ErrNotSessionEntity = errors.New("it is not possible to create a session receiver " +
		"on an entity that does not require sessions")
	ErrSessionCannotBeLocked = errors.New("the requested session cannot be accepted: it is locked by another receiver")
	ErrSessionLockLost       = errors.New("the session lock has expired or was lost")
	ErrNoSessionAvailable    = errors.New("no session is available")

	// 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 AMQPBackend 

type AMQPBackend interface {
	StorageBackend
	SessionBackend
	Defer(ref EntityRef, messageID, lockToken string) error
	ReceiveDeferred(ref EntityRef, seqs []int64, lock bool) ([]MessageInfo, error)
	RenewLock(ref EntityRef, lockToken string) (time.Time, error)
	CancelScheduled(ref EntityRef, seqs []int64) error
	FindByLockToken(ref EntityRef, lockToken string) (MessageInfo, bool, error)
}

AMQPBackend is the backend surface the AMQP plane needs on top of the REST StorageBackend.

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
	RequiresSession   bool
}

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

	AMQPCert *tls.Certificate

	Endpoint    string
	SASKeyValue string
	Port        int
	AMQPPort    int
	AMQPTLSPort int
	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) ServeAMQP 

func (h *Handler) ServeAMQP(ln net.Listener, useTLS bool) error

ServeAMQP serves AMQP 1.0 on ln (TLS-wrapped when useTLS) until Shutdown.

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

func (b *InMemoryBackend) AcceptSessionWait(
	ctx context.Context, ref EntityRef, sessionID, owner string, timeout time.Duration,
) (SessionGrant, error)

AcceptSessionWait locks sessionID for owner. An empty sessionID selects the next session holding deliverable messages, waiting up to timeout for one; ErrNoSessionAvailable is returned when none appears. A specific session held by another owner fails at once with ErrSessionCannotBeLocked.

func (*InMemoryBackend) CancelScheduled 

func (b *InMemoryBackend) CancelScheduled(ref EntityRef, seqs []int64) error

CancelScheduled removes still-pending scheduled messages. For a topic ref every subscription copy of the schedule is removed. Any unknown sequence number yields ErrMessageNotFound after the known ones were cancelled.

func (*InMemoryBackend) CheckSessionLock 

func (b *InMemoryBackend) CheckSessionLock(ref EntityRef, sessionID, owner string) error

CheckSessionLock returns ErrSessionLockLost unless owner holds sessionID.

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

func (b *InMemoryBackend) DeadLetter(ref EntityRef, messageID, lockToken, reason, description string) error

DeadLetter implements StorageBackend.DeadLetter.

func (*InMemoryBackend) Defer 

func (b *InMemoryBackend) Defer(ref EntityRef, messageID, lockToken string) error

Defer parks a locked message; it is then retrievable only by sequence number via ReceiveDeferred.

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

func (b *InMemoryBackend) FindByLockToken(ref EntityRef, token string) (MessageInfo, bool, error)

FindByLockToken returns the locked message holding token and whether it sits in the dead-letter sub-queue.

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

func (b *InMemoryBackend) GetSessionState(ref EntityRef, sessionID, owner string) ([]byte, error)

GetSessionState returns the session's state, nil when unset.

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

func (b *InMemoryBackend) Peek(ref EntityRef, deadLetter bool, fromSeq int64, maxCount int) ([]MessageInfo, error)

Peek implements StorageBackend.Peek.

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

func (b *InMemoryBackend) PeekLockSessionWait(
	ctx context.Context, ref EntityRef, sessionID, owner string, timeout time.Duration,
) (MessageInfo, error)

PeekLockSessionWait locks the oldest deliverable message of the session owner holds, waiting up to timeout. ErrSessionLockLost means the session lock expired or moved.

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

func (b *InMemoryBackend) PeekSession(
	ref EntityRef, sessionID string, fromSeq int64, maxCount int,
) ([]MessageInfo, error)

PeekSession is Peek restricted to one session.

func (*InMemoryBackend) QueueExists 

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

QueueExists reports whether a queue named name exists.

func (*InMemoryBackend) ReceiveAndDelete 

func (b *InMemoryBackend) ReceiveAndDelete(ref EntityRef, deadLetter bool) (MessageInfo, error)

ReceiveAndDelete implements StorageBackend.ReceiveAndDelete.

func (*InMemoryBackend) ReceiveDeferred 

func (b *InMemoryBackend) ReceiveDeferred(ref EntityRef, seqs []int64, lock bool) ([]MessageInfo, error)

ReceiveDeferred returns deferred messages by sequence number, locking them (lock=true) or removing them (lock=false). Unknown sequence numbers are skipped.

func (*InMemoryBackend) ReleaseSession 

func (b *InMemoryBackend) ReleaseSession(ref EntityRef, sessionID, owner string)

ReleaseSession drops owner's lock on sessionID, if it still holds it.

func (*InMemoryBackend) RenewLock 

func (b *InMemoryBackend) RenewLock(ref EntityRef, lockToken string) (time.Time, error)

RenewLock extends the lock identified by lockToken and returns the new expiry. ErrMessageNotFound means the lock was lost.

func (*InMemoryBackend) RenewSessionLock 

func (b *InMemoryBackend) RenewSessionLock(ref EntityRef, sessionID, owner string) (time.Time, error)

RenewSessionLock extends owner's lock on sessionID and returns its expiry.

func (*InMemoryBackend) RequiresSession 

func (b *InMemoryBackend) RequiresSession(ref EntityRef) (bool, error)

RequiresSession reports whether the addressed queue or subscription is session-enabled.

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

func (b *InMemoryBackend) SessionOf(ref EntityRef, owner string) (string, bool)

SessionOf returns the session owner currently holds on ref.

func (*InMemoryBackend) SetSessionState 

func (b *InMemoryBackend) SetSessionState(ref EntityRef, sessionID, owner string, state []byte) error

SetSessionState stores the session's state; nil clears it.

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
	ExpiresAt                  time.Time
	ScheduledFor               time.Time
	CustomHeaders              map[string]string
	AppProps                   map[string]any
	ReplyToSessionID           string
	MessageID                  string
	SessionID                  string
	To                         string
	ContentType                string
	CorrelationID              string
	Label                      string
	ReplyTo                    string
	DeadLetterErrorDescription string
	LockToken                  string
	DeadLetterReason           string
	Body                       []byte
	SequenceNumber             int64
	DeliveryCount              int64
	Deferred                   bool
}

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
	To            string
	// ReplyToSessionID, AppProps and ScheduledFor are populated by the AMQP
	// plane; ScheduledFor in the future holds the message back until then.
	ReplyToSessionID string
	AppProps         map[string]any
	ScheduledFor     time.Time
	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
	RequiresSession   bool

	ActiveMessageCount     int
	DeadLetterMessageCount int
}

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

type SessionBackend 

type SessionBackend interface {
	RequiresSession(ref EntityRef) (bool, error)
	AcceptSessionWait(
		ctx context.Context, ref EntityRef, sessionID, owner string, timeout time.Duration,
	) (SessionGrant, error)
	PeekLockSessionWait(
		ctx context.Context, ref EntityRef, sessionID, owner string, timeout time.Duration,
	) (MessageInfo, error)
	CheckSessionLock(ref EntityRef, sessionID, owner string) error
	RenewSessionLock(ref EntityRef, sessionID, owner string) (time.Time, error)
	GetSessionState(ref EntityRef, sessionID, owner string) ([]byte, error)
	SetSessionState(ref EntityRef, sessionID, owner string, state []byte) error
	ReleaseSession(ref EntityRef, sessionID, owner string)
	SessionOf(ref EntityRef, owner string) (string, bool)
	PeekSession(ref EntityRef, sessionID string, fromSeq int64, maxCount int) ([]MessageInfo, error)
}

SessionBackend is the session surface the AMQP plane needs.

type SessionGrant 

type SessionGrant struct {
	LockedUntil time.Time
	SessionID   string
}

SessionGrant describes an acquired session lock.

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.
	// AMQPPort is the plain-text AMQP 1.0 listener port (0 disables). AMQP is
	// what every Service Bus SDK speaks; azservicebus reaches it with
	// UseDevelopmentEmulator=true in the connection string.
	AMQPPort int `` //nolint:lll // config struct tags are intentionally verbose
	/* 156-byte string literal not displayed */
	// AMQPTLSPort is the AMQP 1.0 over TLS listener port (0 disables). It serves
	// a generated self-signed dev certificate for localhost.
	AMQPTLSPort int `` //nolint:lll // config struct tags are intentionally verbose
	/* 174-byte string literal not displayed */
	ValidateSAS bool `` //nolint:lll // config struct tags are intentionally verbose
	/* 249-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

	// Peek returns up to maxCount non-expired messages with SequenceNumber >=
	// fromSeq, oldest first, without locking them or touching DeliveryCount.
	Peek(ref EntityRef, deadLetter bool, fromSeq int64, maxCount int) ([]MessageInfo, error)

	// ReceiveAndDelete removes and returns the oldest visible message
	// (ErrMessageNotFound if none).
	ReceiveAndDelete(ref EntityRef, deadLetter bool) (MessageInfo, error)

	// DeadLetter moves a locked live message to the dead-letter sub-queue,
	// recording reason and description.
	DeadLetter(ref EntityRef, messageID, lockToken, reason, description 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
	RequiresSession  bool

	ActiveMessageCount     int
	DeadLetterMessageCount 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.