messaging

README

messaging

Kafka messaging plus transactional outbox and inbox (idempotent-consumer) patterns, as a single Go module.

go get github.com/zuksmaq/messaging

Packages

Package	Import path	Holds
root	github.com/zuksmaq/messaging	broker-agnostic contracts and sentinel errors; no third-party imports
kafka	github.com/zuksmaq/messaging/kafka	wire formats and connection settings both sides share
kafka/producer	…/kafka/producer	Config + New for publishing
kafka/consumer	…/kafka/consumer	Config + New, plus the hosted Runner loop
outbox	github.com/zuksmaq/messaging/outbox	transactional outbox and its relay
inbox	github.com/zuksmaq/messaging/inbox	idempotent-consumer inbox
integration	github.com/zuksmaq/messaging/integration	end-to-end tests only, no library code

outbox/inbox stay database-agnostic; the dialect-specific SQL lives in their postgres and sqlserver sub-packages (ADR 0005). The kafka split into producer/consumer is ADR 0008.

The whole repo versions as one module, so there is a single tag per release and no cross-package version skew (ADR 0009).

See messaging-handoff.md for the full design record: decisions, invariants, open items, and rejected alternatives.

Development

go build ./...
go test ./...

# Tests that stand up real brokers/databases via testcontainers:
go test -tags integration -timeout 20m ./...

Documentation

Index

• Constants
• Variables
• type Consumer
• type DeliveryStatus
  • func (s DeliveryStatus) String() string
• type Header
• type Message
• type ProducedMessage
• type Producer
• type ReceivedMessage
  • func (m ReceivedMessage[K, V]) EventID() string
  • func (m ReceivedMessage[K, V]) Tombstone() bool

Constants

const EventIDHeader = "event-id"

EventIDHeader is the header key carrying the producer-assigned idempotency id. The outbox relay stamps it from the outbox row id; consumers use it as the inbox de-duplication key.

Variables

var (
	// ErrSerialization indicates a value could not be encoded for
	// the wire.
	ErrSerialization = errors.New("serialization failed")

	// ErrDeserialization indicates a received value could not be
	// decoded from the wire.
	ErrDeserialization = errors.New("deserialization failed")

	// ErrSchemaRegistryRequired indicates the configured wire format
	// requires a schema registry client that was not provided.
	ErrSchemaRegistryRequired = errors.New("schema registry required")

	// ErrInvalidConfig indicates a config's Validate method rejected
	// it.
	ErrInvalidConfig = errors.New("invalid configuration")

	// ErrBroker indicates the underlying broker client returned an
	// error producing or consuming a message.
	ErrBroker = errors.New("broker error")
)

Sentinel errors for the categories callers need to branch on via errors.Is/errors.As. Wrap these with context as they travel up (fmt.Errorf("...: %w", err)); never compare with ==.

Types

type Consumer

type Consumer[K, V any] interface {
	// Consume blocks until the next message is available.
	Consume(ctx context.Context) (ReceivedMessage[K, V], error)

	// Commit advances the consumer's offset past msg.
	Commit(ctx context.Context, msg ReceivedMessage[K, V]) error
}

Consumer reads keyed messages from a broker. Auto-commit is intentionally absent from the contract: offsets advance only when the caller calls Commit after handling a message.

type DeliveryStatus

type DeliveryStatus int

DeliveryStatus reports how durably a produced message was persisted by the broker.

const (
	// NotPersisted means the broker gave no acknowledgement that the
	// message was written.
	NotPersisted DeliveryStatus = iota
	// PossiblyPersisted means the broker acknowledged the write but
	// durability is not guaranteed (e.g. a single in-sync replica).
	PossiblyPersisted
	// Persisted means the broker acknowledged the write with the
	// durability guarantees required by the producer's configuration.
	Persisted
)

func (DeliveryStatus) String

func (s DeliveryStatus) String() string

String returns the human-readable name of the delivery status.

type Header

type Header struct {
	Key   string
	Value []byte
}

Header is a single broker message header.

type Message

type Message[K, V any] struct {
	Key     K
	Value   V
	Headers map[string][]byte
}

Message is the logical content of a produced or received record: a key, a value, and broker headers.

type ProducedMessage

type ProducedMessage struct {
	Topic     string
	Partition int32
	Offset    int64
	Status    DeliveryStatus
}

ProducedMessage is the broker's acknowledgement of a produced message.

type Producer

type Producer[K, V any] interface {
	// Produce publishes a message and blocks until the broker
	// acknowledges it, returning the assigned coordinates and
	// DeliveryStatus.
	Produce(ctx context.Context, topic string, key K, value V, headers map[string][]byte) (ProducedMessage, error)

	// Close flushes any buffered messages and releases the
	// underlying broker connection.
	Close(ctx context.Context) error
}

Producer publishes keyed messages to a broker. Produce awaits the broker's acknowledgement before returning; implementations must never fire-and-forget.

type ReceivedMessage

type ReceivedMessage[K, V any] struct {
	Message[K, V]

	// RawKey and RawValue are the key and value exactly as the broker
	// delivered them, before deserialization. They survive a
	// deserialization failure, so a poison-message policy can forward
	// the original bytes to a dead-letter topic even when Key and Value
	// are still zero.
	RawKey   []byte
	RawValue []byte

	// HeaderList is every header as delivered, in order, preserving
	// duplicate keys that the Headers convenience map (last-wins)
	// collapses.
	HeaderList []Header

	Topic     string
	Partition int32
	Offset    int64
	Timestamp time.Time
}

ReceivedMessage is a message read from the broker, carrying its content plus the coordinates the broker assigned it.

func (ReceivedMessage[K, V]) EventID

func (m ReceivedMessage[K, V]) EventID() string

EventID returns the EventIDHeader value, or "" if absent.

func (ReceivedMessage[K, V]) Tombstone

func (m ReceivedMessage[K, V]) Tombstone() bool

Tombstone reports whether this message is a tombstone (a nil value), used by log-compacted topics to mark a key for deletion.