README
¶
BabelQueue RabbitMQ transport (Go)
A RabbitMQ (AMQP 0-9-1) babelqueue.Transport for the BabelQueue Go runtime.
Producing publishes to a durable queue with persistent delivery and the
contract AMQP properties (type = URN, correlation_id = trace_id, message_id
= meta.id, and x-schema-version / x-source-lang / x-attempts headers) — so a
PHP/Python/… consumer can route on properties.type without parsing the body.
Consuming uses basic.get + manual ack (at-least-once). Built on
amqp091-go.
go get github.com/babelqueue/babelqueue-go/amqp
import (
babelqueue "github.com/babelqueue/babelqueue-go"
"github.com/babelqueue/babelqueue-go/amqp"
)
tr := amqp.New("amqp://guest:guest@localhost:5672/") // lazy connect
defer tr.Close()
app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
app.Handle("urn:babel:orders:created", func(ctx context.Context, env babelqueue.Envelope) error {
// handle env.Data ...
return nil
})
app.Publish(ctx, "urn:babel:orders:created", map[string]any{"order_id": 1042})
app.Consume(ctx)
Out-of-band transport headers
This transport implements the optional babelqueue.HeaderPublisher capability:
out-of-band headers ride in the AMQP message headers (amqp091.Table) beside the
frozen envelope, never in it (GR-1), and are surfaced back to the consumer on
ReceivedMessage.Headers. So the optional otel module's W3C
traceparent (ADR-0028) — and the bq-replay-bypass marker (ADR-0027) —
propagate over RabbitMQ for true cross-hop span parent-child linkage. The header is
merged on top of the contract x-* headers without clobbering them (a contract
header always wins a key collision).
The BabelQueue core stays zero-dependency; this module is the only place
amqp091-go is pulled in.
Full spec: babelqueue.com · MIT © Muhammet Şafak
Documentation
¶
Overview ¶
Package amqp is a RabbitMQ-backed babelqueue.Transport (AMQP 0-9-1) for the BabelQueue Go runtime. Producing publishes to a durable queue with persistent delivery and the AMQP properties that are part of the cross-language contract (type = URN, correlation_id = trace_id, message_id = meta.id, and the x-schema-version / x-source-lang / x-attempts headers) — so a PHP/Python/... consumer can route on properties.type without parsing the body. Consuming uses basic.get + manual ack (at-least-once), matching the PHP RabbitMQ driver.
It also implements the optional babelqueue.HeaderPublisher capability: out-of-band transport headers (e.g. a W3C traceparent for cross-hop span linkage, ADR-0028, or the bq-replay-bypass marker, ADR-0027) ride in the AMQP message headers (amqp091.Table) beside the frozen envelope — never in it (GR-1) — and are surfaced back to the consumer on babelqueue.ReceivedMessage.Headers.
tr := amqp.New("amqp://guest:guest@localhost:5672/")
app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
The connection is lazy: it (re)connects on first use and after a drop.
Index ¶
- type Transport
- func (t *Transport) Ack(_ 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
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Transport ¶
type Transport struct {
// contains filtered or unexported fields
}
Transport implements babelqueue.Transport over RabbitMQ. It is safe for concurrent use; broker operations are serialized on a single channel.
func New ¶
New returns a transport for the given AMQP URL. It does not connect until the first Publish/Pop.
func (*Transport) Ack ¶
func (t *Transport) Ack(_ context.Context, msg *babelqueue.ReceivedMessage) error
Ack acknowledges the reserved delivery.
func (*Transport) Pop ¶
func (t *Transport) Pop(ctx context.Context, queue string, timeout time.Duration) (*babelqueue.ReceivedMessage, error)
Pop reserves the next message with basic.get (manual ack). When the queue is empty it waits up to timeout (heartbeat-safe) and returns (nil, nil).
func (*Transport) Publish ¶
Publish declares the durable queue and publishes body with persistent delivery and the contract AMQP properties.
func (*Transport) PublishWithHeaders ¶ added in v1.1.0
func (t *Transport) PublishWithHeaders(ctx context.Context, queue, body string, headers map[string]string) error
PublishWithHeaders publishes body together with out-of-band transport headers (babelqueue.HeaderPublisher). The headers ride in the AMQP message headers (amqp091.Table) beside the frozen envelope (GR-1) — e.g. a W3C traceparent for cross-hop span linkage (ADR-0028). They are merged on top of the contract x-* headers without clobbering them (a contract header always wins a key collision), and empty/blank keys are skipped. A nil/empty map behaves exactly like Transport.Publish.
Source Files