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"))
A message whose bq_schema_version application property is present, not blank and not the core's supported version is version-gated (§7.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.
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 OnSchemaVersionRejected
- 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
- func (t *Transport) PublishWithHeaders(ctx context.Context, queue, body string, headers map[string]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 {
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 OnSchemaVersionRejected ¶ added in v1.1.0
type OnSchemaVersionRejected func(version, queue string)
OnSchemaVersionRejected is called when the version gate (§7.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, accepted or retried.
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.
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 {
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.
func (*Transport) PublishWithHeaders ¶ added in v1.3.0
func (t *Transport) PublishWithHeaders(ctx context.Context, queue, body string, headers map[string]string) error
PublishWithHeaders implements babelqueue.HeaderPublisher: it sends body like Transport.Publish and adds the out-of-band transport headers as extra string-valued AMQP 1.0 application properties beside the §7 bq_ projection - e.g. a W3C traceparent / tracestate for cross-hop span linkage (ADR-0028; the names are hyphen-free, so JMS-legal). The body is the unchanged envelope (GR-1). A bq_ contract property always wins a key collision, and blank keys and values are skipped. A nil or empty map behaves exactly like Publish.
func (*Transport) Rewind ¶ added in v1.1.0
func (t *Transport) Rewind(ctx context.Context, msg *babelqueue.ReceivedMessage) error
Rewind implements babelqueue.Rewinder: it releases the reserved message back to the broker (AMQP 1.0 "released" disposition) instead of leaving it unsettled on the link, so it is redelivered. Nothing is accepted or deleted. A receiver without ReleaseMessage (a test fake) leaves the message reserved, and the broker redelivers it when the link closes.
Source Files