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); the broker's native RedeliveryCount is reconciled onto the envelope as attempts = max(body, RedeliveryCount) (no −1, because Pulsar's redelivery count is 0-based) 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 Option
- func WithClient(c Client) Option
- func WithMaxWaitTime(d time.Duration) 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
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 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 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 attempts to max(body.attempts, RedeliveryCount). Returns (nil, nil) when no message arrives.
Source Files