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.
Full spec: https://babelqueue.com
Index ¶
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 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.
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.
Source Files