Documentation
¶
Overview ¶
Package pulsar is an Apache Pulsar-backed babelqueue.Transport for the BabelQueue Go runtime, on the pure-Go github.com/apache/pulsar-client-go (no CGo, no libpulsar). Producing sends the canonical envelope as the message payload and projects the contract envelope fields onto native Pulsar message properties (string→string): bq-job = URN, bq-trace-id = trace_id, bq-message-id = meta.id, plus bq-schema-version / bq-source-lang / bq-attempts — so a Java/.NET/... peer can route on bq-job without parsing the body. Consuming receives one message at a time (Receive -> process -> Ack). Per §5.2 / §5.7 the bq-job property is the routing key (the body's job is used only when bq-job is absent or blank; a differing body URN is reported to the optional OnRoutingMismatch hook) and bq-attempts is the authoritative counter (the body's attempts is used only when the property is absent or not a plain non-negative decimal integer). Before any of that, a bq-schema-version property that is present but not the core's supported version is version-gated (§5.7 step 3): 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 (also under StrategyRelease, which would otherwise loop; the optional OnSchemaVersionRejected hook names the rejected version). That route is an in-process marker only: it carries the delivery's original bq-job and bq-schema-version, and a retry or dead-letter re-publish writes those back verbatim, so the sentinel never reaches the wire and a redrive meets the same gate. The result is then raised to the broker's native RedeliveryCount (no −1, because Pulsar's redelivery count is 0-based): attempts = max(bq-attempts, RedeliveryCount), so a redelivered message reflects its true count without lowering the runtime's own counter.
tr, _ := pulsar.New(pulsar.WithURL("pulsar://localhost:6650"))
app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
This binding implements §5 of the BabelQueue broker-bindings contract. The envelope is unchanged (schema_version stays 1); Apache Pulsar is purely additive.
Full spec: https://babelqueue.com
Index ¶
- type Client
- type Consumer
- type OnRoutingMismatch
- type OnSchemaVersionRejected
- type Option
- func WithClient(c Client) Option
- func WithMaxWaitTime(d time.Duration) Option
- func WithOnRoutingMismatch(fn OnRoutingMismatch) Option
- func WithOnSchemaVersionRejected(fn OnSchemaVersionRejected) Option
- func WithPulsarClient(c pulsarsdk.Client) Option
- func WithSubscription(name string) Option
- func WithSubscriptionType(st pulsarsdk.SubscriptionType) Option
- func WithTopicPrefix(prefix string) Option
- func WithURL(url string) Option
- type Producer
- type Transport
- func (t *Transport) Ack(_ context.Context, msg *babelqueue.ReceivedMessage) 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) PublishFromWithHeaders(ctx context.Context, queue, body string, from *babelqueue.ReceivedMessage, ...) error
- func (t *Transport) PublishRouted(ctx context.Context, queue, body, routeURN string) error
- func (t *Transport) PublishWithHeaders(ctx context.Context, queue, body string, headers map[string]string) error
- func (t *Transport) Rewind(_ 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 {
CreateProducer(queue string) (Producer, error)
Subscribe(queue string) (Consumer, error)
}
Client creates per-queue producers and subscriptions. A pulsarsdk.Client is wrapped to satisfy it (see WithPulsarClient); a fake satisfies it in tests.
type Consumer ¶
type Consumer interface {
Receive(ctx context.Context) (pulsarsdk.Message, error)
Ack(msg pulsarsdk.Message) error
Nack(msg pulsarsdk.Message)
}
Consumer is the subset of pulsarsdk.Consumer the transport uses.
type OnRoutingMismatch ¶ added in v1.1.0
type OnRoutingMismatch func(propertyURN, bodyURN string, env *babelqueue.Envelope)
OnRoutingMismatch is called when a message's bq-job property and its body's job name different URNs (the property wins and the message is routed on it). propertyURN is the trimmed bq-job value, bodyURN the body's job, and env the accepted envelope as the handler will see it (body unchanged, attempts reconciled) — treat it as read-only. It is not called for an envelope the consumer rejects. It is a diagnostic hook, separate from any error handling: a panic in it is recovered and ignored, and it never changes how the message is delivered, acked or retried.
type OnSchemaVersionRejected ¶ added in v1.1.0
type OnSchemaVersionRejected func(version, queue string)
OnSchemaVersionRejected is called when the version gate (§5.7 step 3) quarantines a message whose bq-schema-version property is present but not the core's supported version. version is the raw property value 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, acked or retried.
type Option ¶
type Option func(*Transport)
Option customizes New.
func WithClient ¶
WithClient injects a Client (a fake in tests), bypassing all Pulsar wiring.
func WithMaxWaitTime ¶
WithMaxWaitTime caps how long a single Pop blocks waiting for a message.
func WithOnRoutingMismatch ¶ added in v1.1.0
func WithOnRoutingMismatch(fn OnRoutingMismatch) Option
WithOnRoutingMismatch registers an optional OnRoutingMismatch hook (default: none).
func WithOnSchemaVersionRejected ¶ added in v1.1.0
func WithOnSchemaVersionRejected(fn OnSchemaVersionRejected) Option
WithOnSchemaVersionRejected registers an optional OnSchemaVersionRejected hook (default: none).
func WithPulsarClient ¶
WithPulsarClient wraps a preconfigured pulsarsdk.Client (e.g. built with TLS or auth), bypassing URL parsing.
func WithSubscription ¶
WithSubscription sets the subscription name (default "babelqueue").
func WithSubscriptionType ¶
func WithSubscriptionType(st pulsarsdk.SubscriptionType) Option
WithSubscriptionType sets the subscription type (default pulsarsdk.Shared).
func WithTopicPrefix ¶
WithTopicPrefix prefixes every queue name to form the topic (e.g. "persistent://public/default/"). Empty by default — the bare queue name is the topic.
type Producer ¶
type Producer interface {
Send(ctx context.Context, msg *pulsarsdk.ProducerMessage) (pulsarsdk.MessageID, error)
}
Producer is the subset of pulsarsdk.Producer the transport uses; the concrete producer 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 Pulsar. It is safe for concurrent use; the per-queue producer/consumer caches are guarded by a mutex.
func New ¶
New builds a transport. Provide a service URL (WithURL), a preconfigured Pulsar client (WithPulsarClient), or an injected Client (WithClient).
func (*Transport) Ack ¶
func (t *Transport) Ack(_ context.Context, msg *babelqueue.ReceivedMessage) error
Ack acknowledges the reserved message.
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 the routing URN (bq-job, falling back to the body; set on ReceivedMessage.RouteURN, the body is not rewritten) and attempts (bq-attempts, falling back to the body, floored at RedeliveryCount; written into the body). Returns (nil, nil) when no message arrives.
func (*Transport) PublishFromWithHeaders ¶ added in v1.5.0
func (t *Transport) PublishFromWithHeaders(ctx context.Context, queue, body string, from *babelqueue.ReceivedMessage, headers map[string]string) error
PublishFromWithHeaders is the header-carrying republish (babelqueue.HeaderRepublisher): it sends body like Transport.PublishRouted with from.RouteURN - so the routing property, or the original bq-job / bq-schema-version of a version-gated delivery, is kept exactly as a plain republish keeps it (§5.2/§5.7) - and adds the out-of-band transport headers as extra properties, with the same rules as Transport.PublishWithHeaders (a bq- contract property wins a collision, blanks are skipped). babelqueue.Redrive uses it to stamp bq-replay-bypass without flipping the route or bypassing the version gate. A nil from, or one with no RouteURN, derives bq-job from the body.
func (*Transport) PublishRouted ¶ added in v1.1.0
PublishRouted is [Publish] with the routing property pinned: a non-empty routeURN is written as bq-job instead of the body's job, so an App retry or release re-publish keeps the original delivery's routing URN (§5.2/§5.7). A blank routeURN derives bq-job from the body.
func (*Transport) PublishWithHeaders ¶ added in v1.5.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 message properties beside the §5 bq- projection - e.g. a W3C traceparent / tracestate for cross-hop span linkage (ADR-0028). The payload 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.2.0
func (t *Transport) Rewind(_ context.Context, msg *babelqueue.ReceivedMessage) error
Rewind hands a reserved message back to the broker with a negative acknowledgement, so it is redelivered after the client's negative-ack delay instead of being lost. The App calls it (the transport implements babelqueue.Rewinder) when it could not write the dead-letter record for a version-gated message and therefore must not ack. It never acks. A handle that is not a Pulsar message is ignored.
Source Files