Documentation
¶
Overview ¶
Package kafka is an Apache Kafka-backed babelqueue.Transport for the BabelQueue Go runtime, on the pure-Go github.com/segmentio/kafka-go (no CGo, no librdkafka). Producing writes the canonical envelope as the record value and projects the contract fields onto native Kafka record headers (bq-job = URN, bq-trace-id, bq-message-id, bq-schema-version, bq-source-lang, bq-attempts — all UTF-8 byte strings) with the record timestamp mirroring meta.created_at, so a Java/.NET/... peer routes on bq-job without parsing the body. Consuming is process-then-commit: FetchMessage reserves a record (no auto-commit), and Ack commits the offset only after the handler returns (at-least-once). Kafka has no native delivery count, so the bq-attempts header is the authoritative retry counter (the body's attempts is the fallback for non-BabelQueue producers); the App owns retry by republishing with attempts+1 and dead-letters to <queue>.dlq.
tr, _ := kafka.New(kafka.WithBrokers("localhost:9092"), kafka.WithGroupID("orders-workers"))
app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
This binding implements §6 of the BabelQueue broker-bindings contract. The envelope is unchanged (schema_version stays 1); Apache Kafka is purely additive.
Full spec: https://babelqueue.com
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Option ¶
type Option func(*config)
Option customizes New.
func WithBrokers ¶
WithBrokers sets the bootstrap broker list (host:port).
func WithGroupID ¶
WithGroupID sets the consumer group used for offset commits (required to consume).
func WithMaxWaitTime ¶
WithMaxWaitTime caps how long a single Pop blocks waiting for a record.
func WithReaderFactory ¶
WithReaderFactory injects a per-queue Reader factory (a fake in tests).
func WithWriter ¶
WithWriter injects a Writer (a fake in tests), bypassing all Kafka wiring.
type Reader ¶
type Reader interface {
FetchMessage(ctx context.Context) (kafkago.Message, error)
CommitMessages(ctx context.Context, msgs ...kafkago.Message) error
}
Reader is the subset of *kafkago.Reader the transport uses (manual commit — no auto-commit).
type Transport ¶
type Transport struct {
// contains filtered or unexported fields
}
Transport implements babelqueue.Transport over Apache Kafka. It is safe for concurrent Publish; run Pop/Ack from a single goroutine per queue (a Kafka consumer is single-threaded).
func New ¶
New builds a transport. Provide brokers + a group id (WithBrokers / WithGroupID), or inject a Writer and a reader factory (WithWriter / WithReaderFactory) for tests.
func (*Transport) Ack ¶
func (t *Transport) Ack(ctx context.Context, msg *babelqueue.ReceivedMessage) error
Ack commits the reserved record's offset (process-then-commit; at-least-once).
func (*Transport) Pop ¶
func (t *Transport) Pop(ctx context.Context, queue string, timeout time.Duration) (*babelqueue.ReceivedMessage, error)
Pop reserves the next record (FetchMessage, no auto-commit), bounded by timeout (and WithMaxWaitTime). It reconciles attempts from the authoritative bq-attempts header. Returns (nil, nil) when no record arrives before the deadline.
Source Files