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), or drops it when dead-lettering is off. 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
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.
Source Files