Documentation
¶
Overview ¶
Package azureservicebus is an Azure Service Bus-backed babelqueue.Transport for the BabelQueue Go runtime. Producing sends the canonical envelope as the message body and projects the contract envelope fields onto native Service Bus fields (Subject = URN, CorrelationID = trace_id, MessageID = meta.id, plus the bq- application properties) — so a .NET/Java/... peer can route on Subject and correlate on CorrelationID without parsing the body. Consuming uses the PeekLock reservation model (ReceiveMessages -> process -> CompleteMessage); the broker's native DeliveryCount is reconciled onto the envelope as attempts = max(body, DeliveryCount−1) so a crash-redelivered message reflects its true delivery count without lowering the runtime's own counter.
tr, _ := azureservicebus.New(azureservicebus.WithConnectionString(cs))
app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
This binding implements §4 of the BabelQueue broker-bindings contract. The envelope is unchanged (schema_version stays 1); Azure Service Bus is purely additive.
Before the attempts reconcile, a bq-schema-version application property that is present but not the core's supported version is version-gated (§4.7): the body is not decoded and the message is routed to a reserved unsupported-version URN no handler owns, so the App dead-letters it directly, with no retry (whatever the unknown-URN strategy, StrategyRelease included). With dead-lettering off it is not completed: the transport's Transport.SettleNative dead-letters it natively into $DeadLetterQueue (reason "unsupported schema version"); a poison body takes the same native settle on the first delivery. The optional OnSchemaVersionRejected hook names the rejected version.
Full spec: https://babelqueue.com
Index ¶
- type Client
- type OnSchemaVersionRejected
- type Option
- type Receiver
- type Sender
- type Transport
- func (t *Transport) Ack(ctx context.Context, msg *babelqueue.ReceivedMessage) error
- func (t *Transport) Pop(ctx context.Context, queue string, timeout time.Duration) (*babelqueue.ReceivedMessage, error)
- func (t *Transport) Publish(ctx context.Context, queue, body string) error
- func (t *Transport) Rewind(ctx context.Context, msg *babelqueue.ReceivedMessage) error
- func (t *Transport) SettleNative(ctx context.Context, msg *babelqueue.ReceivedMessage, ...) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Client ¶
type Client interface {
NewSender(queueOrTopic string) (Sender, error)
NewReceiver(queue string) (Receiver, error)
}
Client creates per-entity senders and receivers. A *azservicebus.Client is wrapped to satisfy it (see WithAzureClient); a fake satisfies it in tests.
type OnSchemaVersionRejected ¶ added in v1.1.0
type OnSchemaVersionRejected func(version, queue string)
OnSchemaVersionRejected is called when the version gate (§4.7) quarantines a message whose bq-schema-version application property is present but not the core's supported version. version is the property value rendered as text and queue the queue it was received from. The gate runs before the body is decoded; the App's quarantine error names the reserved route, and this hook additionally hands an operator the rejected version and queue directly (including for a body that is not valid JSON). It is a diagnostic hook: a panic in it is recovered and ignored, and it never changes how the message is delivered, acked or retried.
type Option ¶
type Option func(*Transport)
Option customizes New.
func WithAzureClient ¶
func WithAzureClient(c *azservicebus.Client) Option
WithAzureClient wraps a preconfigured *azservicebus.Client (e.g. built with a namespace + azidentity TokenCredential), bypassing connection-string parsing.
func WithClient ¶
WithClient injects a Client (a fake in tests), bypassing all Azure wiring.
func WithConnectionString ¶
WithConnectionString builds the client from an ASB connection string (Endpoint=sb://...;SharedAccessKeyName=...;SharedAccessKey=...).
func WithMaxWaitTime ¶
WithMaxWaitTime caps how long a single Pop blocks waiting for a message. The runtime's per-iteration poll timeout still bounds it; this caps the upper limit.
func WithOnSchemaVersionRejected ¶ added in v1.1.0
func WithOnSchemaVersionRejected(fn OnSchemaVersionRejected) Option
WithOnSchemaVersionRejected registers an optional OnSchemaVersionRejected hook (default: none).
type Receiver ¶
type Receiver interface {
ReceiveMessages(ctx context.Context, maxMessages int, options *azservicebus.ReceiveMessagesOptions) ([]*azservicebus.ReceivedMessage, error)
CompleteMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.CompleteMessageOptions) error
}
Receiver is the subset of *azservicebus.Receiver the transport uses.
type Sender ¶
type Sender interface {
SendMessage(ctx context.Context, message *azservicebus.Message, options *azservicebus.SendMessageOptions) error
}
Sender is the subset of *azservicebus.Sender the transport uses; the concrete sender satisfies it, and a fake satisfies it in tests.
type Transport ¶
type Transport struct {
// contains filtered or unexported fields
}
Transport implements babelqueue.Transport over Azure Service Bus. It is safe for concurrent use; the per-queue sender/receiver caches are guarded by a mutex.
func New ¶
New builds a transport. Provide a connection string (WithConnectionString), a preconfigured Azure client (WithAzureClient), or an injected Client (WithClient).
func (*Transport) Ack ¶
func (t *Transport) Ack(ctx context.Context, msg *babelqueue.ReceivedMessage) error
Ack completes the reserved message (removing it from the entity).
func (*Transport) Pop ¶
func (t *Transport) Pop(ctx context.Context, queue string, timeout time.Duration) (*babelqueue.ReceivedMessage, error)
Pop reserves the next message (PeekLock), bounded by timeout (and WithMaxWaitTime). It reconciles attempts to max(body.attempts, DeliveryCount − 1). Returns (nil, nil) when no message arrives.
func (*Transport) Rewind ¶ added in v1.1.0
func (t *Transport) Rewind(ctx context.Context, msg *babelqueue.ReceivedMessage) error
Rewind implements babelqueue.Rewinder: it abandons the peek-lock so the message is redelivered right away instead of when the lock lapses. Nothing is completed or dead-lettered. A receiver without AbandonMessage (a test fake) leaves the lock to lapse on its own.
func (*Transport) SettleNative ¶ added in v1.3.0
func (t *Transport) SettleNative(ctx context.Context, msg *babelqueue.ReceivedMessage, reason, description string) error
SettleNative implements babelqueue.NativeSettler (ADR-0037 decisions 1 and 2): a version-gated delivery or a poison body that the App may neither process nor drop, with dead-lettering off, is moved by an explicit native DeadLetter(reason, description) into the entity's $DeadLetterQueue. Nothing is completed, nothing is re-published or retried, and the body stays exactly as it arrived; reason ("unsupported schema version" or "poison") becomes the DeadLetterReason and description the DeadLetterErrorDescription. A receiver without DeadLetterMessage (a test fake) falls back to Abandon, which leaves the message to MaxDeliveryCount - also a conformant native settle; a receiver with neither lets the lock lapse on its own.
Source Files