pubsub

Documentation

Overview

Package pubsub provides publish/subscribe abstractions for Helix Cluster OS. It ships two implementations behind a common interface:

• InMemoryBroker – lock-free channel map, used in unit tests.
• KafkaBroker – real segmentio/kafka-go producer+consumer, acks=all, at-least-once delivery with explicit offset commit.

Package pubsub – real Kafka broker implementation. Uses github.com/segmentio/kafka-go:

• Producer: WriterConfig{RequiredAcks: RequireAll, Async: false} — idempotent acks=all.
• Consumer: ReaderConfig{GroupID: …} — at-least-once + explicit CommitMessages.

This is the production implementation of Broker; InMemoryBroker is used in unit tests.

Package pubsub – in-memory broker implementation. This file contains InMemoryBroker, the lightweight, race-safe implementation used in unit tests and as a reference. For real Kafka connectivity see kafka_broker.go.

Package pubsub – Kafka topic configuration and idempotent topic provisioning.

TopicCatalog is the authoritative table of Helix control-plane topics. EnsureTopics creates topics that do not yet exist via the kafka-go admin API, setting NumPartitions, ReplicationFactor and retention.ms for each entry. It is idempotent: calling it on an already-existing topic is a safe no-op.

Index

• func EnsureTopics(ctx context.Context, brokers []string) error
• func EnsureTopicsWithSeam(_ context.Context, admin AdminSeam) error
• func NewBroker() *legacyBroker
• func TopicCatalog() map[string]TopicSpec
• func TopicNames() []string
• type AdminSeam
• type Broker
• type InMemoryBroker
  • func NewInMemoryBroker() *InMemoryBroker
  • func (b *InMemoryBroker) Close() error
  • func (b *InMemoryBroker) Publish(ctx context.Context, topic, key, value string) error
  • func (b *InMemoryBroker) Subscribe(ctx context.Context, topic, _ string) (<-chan Message, error)
• type KafkaBroker
  • func NewKafkaBroker(brokers []string) *KafkaBroker
  • func (kb *KafkaBroker) CapturedReaderConfig(topic, group string) ReaderConfig
  • func (kb *KafkaBroker) CapturedWriterConfig(topic string) WriterConfig
  • func (kb *KafkaBroker) Close() error
  • func (kb *KafkaBroker) Publish(ctx context.Context, topic, key, value string) error
  • func (kb *KafkaBroker) Subscribe(ctx context.Context, topic, group string) (<-chan Message, error)
• type Message
• type ReaderConfig
• type TopicSpec
• type WriterConfig

Functions

func EnsureTopics

func EnsureTopics(ctx context.Context, brokers []string) error

EnsureTopics creates all control-plane topics that do not yet exist. It is idempotent: "topic already exists" errors are silently ignored. The admin connection is acquired from brokers[0]; subsequent elements are used as fall-backs if the first is unreachable.

EnsureTopicsWithSeam is the seam-injectable variant used by unit tests.

func EnsureTopicsWithSeam

func EnsureTopicsWithSeam(_ context.Context, admin AdminSeam) error

EnsureTopicsWithSeam is the seam-injectable entry point that accepts any AdminSeam. Unit tests call this directly with a fakeAdmin; production code calls EnsureTopics which wires a kafkaConnAdmin.

func NewBroker

func NewBroker() *legacyBroker

NewBroker creates a new backward-compatible Broker.

func TopicCatalog

func TopicCatalog() map[string]TopicSpec

TopicCatalog returns a snapshot copy of the authoritative topic table. Callers may iterate or assert individual entries without risk of mutation.

func TopicNames

func TopicNames() []string

TopicNames returns the sorted list of control-plane topic names.

Types

type AdminSeam

type AdminSeam interface {
	// CreateTopics sends one CreateTopics call per topic.  Implementations
	// MUST tolerate "topic already exists" as a non-error (idempotency).
	CreateTopics(topics ...kafka.TopicConfig) error
	// Close releases the underlying connection.
	Close() error
}

AdminSeam is the injection point used by unit tests. The real implementation is kafkaConnAdmin (backed by kafka-go Conn); tests inject fakeAdmin to assert exact TopicConfig arguments without a live broker.

type Broker

type Broker interface {
	// Publish sends a single message to the given topic.
	// Key is used for Kafka partition routing (hash-based).
	// Implementations MUST use idempotent, acks=all semantics where the
	// underlying transport supports it.
	Publish(ctx context.Context, topic, key, value string) error

	// Subscribe returns a channel that receives all messages for topic/group.
	// group is the consumer-group identifier; each group member competes for
	// messages (at-least-once, exactly-one-consumer-per-key within a group).
	// The returned channel is closed when ctx is cancelled.
	Subscribe(ctx context.Context, topic, group string) (<-chan Message, error)

	// Close releases all resources held by the broker.
	Close() error
}

Broker is the top-level publish/subscribe capability. Both the in-memory stub and the Kafka client implement this interface so that production code and unit tests share the same surface.

type InMemoryBroker

type InMemoryBroker struct {
	// contains filtered or unexported fields
}

InMemoryBroker is a simple in-memory pub/sub broker using buffered channels. It retains the original fan-out semantics: every subscriber on a topic receives every published message. The channel buffer is fixed at 10 slots; overflow is silently dropped (non-blocking contract).

group is accepted for API compatibility but ignored – in-memory delivery does not model consumer-group exclusivity.

func NewInMemoryBroker

func NewInMemoryBroker() *InMemoryBroker

NewInMemoryBroker creates a ready-to-use InMemoryBroker.

func (*InMemoryBroker) Close

func (b *InMemoryBroker) Close() error

Close is a no-op for the in-memory broker.

func (*InMemoryBroker) Publish

func (b *InMemoryBroker) Publish(ctx context.Context, topic, key, value string) error

Publish sends msg to all subscribers of topic. ctx is honoured: if it is already cancelled, Publish returns ctx.Err() without delivering.

func (*InMemoryBroker) Subscribe

func (b *InMemoryBroker) Subscribe(ctx context.Context, topic, _ string) (<-chan Message, error)

Subscribe returns a buffered channel that receives all messages published to topic while ctx is alive. The channel is closed when ctx is cancelled.

type KafkaBroker

type KafkaBroker struct {
	// contains filtered or unexported fields
}

KafkaBroker wraps segmentio/kafka-go writers and readers behind the Broker interface. One Writer per topic is created lazily and reused. Readers are created per (topic, group) pair; each call to Subscribe starts a dedicated fetch-and-forward goroutine.

func NewKafkaBroker

func NewKafkaBroker(brokers []string) *KafkaBroker

NewKafkaBroker creates a KafkaBroker connected to the given broker addresses.

func (*KafkaBroker) CapturedReaderConfig

func (kb *KafkaBroker) CapturedReaderConfig(topic, group string) ReaderConfig

CapturedReaderConfig returns the ReaderConfig that would be applied for the given topic/group pair. Used by unit tests to verify the seam.

func (*KafkaBroker) CapturedWriterConfig

func (kb *KafkaBroker) CapturedWriterConfig(topic string) WriterConfig

CapturedWriterConfig returns the WriterConfig that would be (or was) applied for the given topic. Used by unit tests to verify the seam without a live broker. When writerFactory is the default, this constructs a *Writer solely to inspect its exported fields, then discards it.

func (*KafkaBroker) Close

func (kb *KafkaBroker) Close() error

Close flushes and closes all writers.

func (*KafkaBroker) Publish

func (kb *KafkaBroker) Publish(ctx context.Context, topic, key, value string) error

Publish writes a single message to Kafka with acks=all. Key is used for partition routing (Hash balancer: same key → same partition).

func (*KafkaBroker) Subscribe

func (kb *KafkaBroker) Subscribe(ctx context.Context, topic, group string) (<-chan Message, error)

Subscribe starts a consumer-group reader for topic/group and forwards messages to the returned channel. The channel is closed when ctx is cancelled. Offset is committed after every successfully forwarded message (at-least-once).

type Message

type Message struct {
	Topic string
	Key   []byte
	Value []byte
}

Message is a message delivered by a Broker subscription.

type ReaderConfig

type ReaderConfig struct {
	Brokers []string
	GroupID string
	Topic   string
}

ReaderConfig captures configuration applied to each Kafka reader.

type TopicSpec

type TopicSpec struct {
	// NumPartitions controls parallelism and key-routing fan-out.
	NumPartitions int
	// ReplicationFactor is 1 for single-node, ≥2 for HA clusters.
	ReplicationFactor int
	// RetentionMs is the retention.ms topic config value (milliseconds).
	// 0 means "use broker default" (not written as a ConfigEntry).
	RetentionMs int64
}

TopicSpec is the authoritative configuration for a single Kafka topic.

type WriterConfig

type WriterConfig struct {
	Brokers      []string
	RequiredAcks int  // kafka-go: RequireAll == -1
	Async        bool // must be false for acks=all
}

WriterConfig captures the configuration that was applied to the Kafka writer. Exposed so unit tests can assert the seam without touching a live broker.