Documentation
¶
Overview ¶
Package artemis is an Apache ActiveMQ Artemis-backed babelqueue.Transport for the BabelQueue Go runtime, speaking AMQP 1.0 on the pure-Go github.com/Azure/go-amqp (no CGo). Artemis speaks AMQP 1.0 (not RabbitMQ's 0-9-1), so this is a distinct transport from the amqp (0-9-1) one.
Producing sends the canonical envelope as the message body and projects the contract envelope fields onto the AMQP a JMS peer reads: correlation-id = trace_id (JMSCorrelationID), creation-time = meta.created_at (JMSTimestamp), the x-opt-jms-type message annotation = URN (JMSType, the AMQP-JMS mapping), plus the bq_ application properties (string-valued; underscored names, since JMS property names must be valid Java identifiers) — so a Java/.NET/Node/Python peer can route and correlate without parsing the body. Consuming receives one message at a time (Receive -> process -> Ack); the broker's native AMQP delivery-count is reconciled onto the envelope as attempts = max(body, delivery-count) — no −1, because the AMQP counter is 0-based (0 on first delivery). (The Java JMS binding reads the 1-based JMSXDeliveryCount and subtracts 1, arriving at the same 0-based attempts.)
tr, _ := artemis.New(ctx, artemis.WithURL("amqp://localhost:5672"))
app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
This binding implements §7 of the BabelQueue broker-bindings contract. The envelope is unchanged (schema_version stays 1); Apache ActiveMQ Artemis is purely additive.
Full spec: https://babelqueue.com
Index ¶
- type Client
- type Option
- type Receiver
- type Sender
- type Transport
- func (t *Transport) Ack(ctx context.Context, msg *babelqueue.ReceivedMessage) error
- func (t *Transport) Close() 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
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Client ¶
type Client interface {
Sender(ctx context.Context, queue string) (Sender, error)
Receiver(ctx context.Context, queue string) (Receiver, error)
}
Client creates per-queue senders and receivers. An amqp.Session is wrapped to satisfy it (see [WithSession]); a fake satisfies it in tests.
type Option ¶
type Option func(*Transport)
Option customizes New.
func WithClient ¶
WithClient injects a Client (a fake in tests), bypassing all AMQP wiring.
func WithMaxWaitTime ¶
WithMaxWaitTime caps how long a single Pop blocks waiting for a message.
type Receiver ¶
type Receiver interface {
Receive(ctx context.Context, opts *amqp.ReceiveOptions) (*amqp.Message, error)
AcceptMessage(ctx context.Context, msg *amqp.Message) error
}
Receiver is the subset of amqp.Receiver the transport uses.
type Sender ¶
type Sender interface {
Send(ctx context.Context, msg *amqp.Message, opts *amqp.SendOptions) error
}
Sender is the subset of amqp.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 Apache ActiveMQ Artemis (AMQP 1.0). It is safe for concurrent use; the per-queue sender/receiver caches are guarded by a mutex.
func New ¶
New builds a transport. Provide an AMQP URL (WithURL) or an injected Client (WithClient). With a URL it dials the broker and opens a session, both closed by Transport.Close.
func (*Transport) Ack ¶
func (t *Transport) Ack(ctx context.Context, msg *babelqueue.ReceivedMessage) error
Ack acknowledges (accepts) the reserved message.
func (*Transport) Close ¶
Close releases the dialed connection (a no-op when a Client was injected).
func (*Transport) Pop ¶
func (t *Transport) Pop(ctx context.Context, queue string, timeout time.Duration) (*babelqueue.ReceivedMessage, error)
Pop receives the next message, bounded by timeout (and WithMaxWaitTime). It reconciles attempts to max(body.attempts, delivery-count). Returns (nil, nil) when no message arrives.
Source Files