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)
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.
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
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.
Source Files