kafka

Documentation

Overview

Package kafka provides a Kafka backend for the ling-base/mq broker abstraction. It uses github.com/segmentio/kafka-go for transport and maps the mq.Broker/Producer/Consumer interfaces onto Kafka topics and consumer groups.

Kafka has no exchanges, queues, or bindings. Topics play the role of exchanges, partitions play the role of routing keys, and consumer groups play the role of queues. The topology methods (DeclareQueue, Bind, Unbind, DeleteQueue, DeleteExchange) are therefore no-ops that return nil. DeclareExchange optionally creates a Kafka topic.

Index

• type Broker
  • func New(cfg Config) (*Broker, error)
  • func (b *Broker) Bind(queue, exchange, routingKey string) error
  • func (b *Broker) Close() error
  • func (b *Broker) Connect() error
  • func (b *Broker) Consumer(topic string, opts mq.ConsumeOptions) (mq.Consumer, error)
  • func (b *Broker) DeclareExchange(name string, opts mq.ExchangeOptions) error
  • func (b *Broker) DeclareQueue(name string, opts mq.QueueOptions) error
  • func (b *Broker) DeleteExchange(name string) error
  • func (b *Broker) DeleteQueue(name string) error
  • func (b *Broker) IsConnected() bool
  • func (b *Broker) Metrics() mq.Metrics
  • func (b *Broker) Producer(topic string, opts mq.PublishOptions) (mq.Producer, error)
  • func (b *Broker) Unbind(queue, exchange, routingKey string) error
• type Config
  • func DefaultConfig() Config
• type Consumer
  • func (c *Consumer) IsRunning() bool
  • func (c *Consumer) Start(ctx context.Context) error
  • func (c *Consumer) Stop(timeout time.Duration) error
  • func (c *Consumer) String() string
• type Producer
  • func (p *Producer) Close() error
  • func (p *Producer) Publish(ctx context.Context, msg *mq.Message) error

Types

type Broker

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

Broker implements mq.Broker for Kafka.

func New

func New(cfg Config) (*Broker, error)

New creates a new Kafka broker. Call Connect() to verify broker reachability (optional — Kafka connections are per-operation).

func (*Broker) Bind

func (b *Broker) Bind(queue, exchange, routingKey string) error

Bind is a no-op for Kafka. Kafka routes by topic and partition, not by exchange/queue bindings.

func (*Broker) Close

func (b *Broker) Close() error

Close shuts down the broker, closing all producers and consumers.

func (*Broker) Connect

func (b *Broker) Connect() error

Connect verifies that at least one broker is reachable. For Kafka, connections are established per-operation by the reader/writer, so this is optional. It performs a quick metadata request.

func (*Broker) Consumer

func (b *Broker) Consumer(topic string, opts mq.ConsumeOptions) (mq.Consumer, error)

Consumer creates a consumer for the given topic with the given options. The consumer is not started; call Start to begin.

If topic is empty, the broker's default Topic is used. The consumer group ID is taken from ConsumeOptions.ConsumerTag if non-empty, otherwise the broker's default GroupID.

func (*Broker) DeclareExchange

func (b *Broker) DeclareExchange(name string, opts mq.ExchangeOptions) error

DeclareExchange creates a Kafka topic if it does not already exist. Kafka has no exchanges; topics serve as the publish target.

func (*Broker) DeclareQueue

func (b *Broker) DeclareQueue(name string, opts mq.QueueOptions) error

DeclareQueue is a no-op for Kafka. Kafka has no queues; consumer groups serve as the subscription unit.

func (*Broker) DeleteExchange

func (b *Broker) DeleteExchange(name string) error

DeleteExchange deletes a Kafka topic.

func (*Broker) DeleteQueue

func (b *Broker) DeleteQueue(name string) error

DeleteQueue is a no-op for Kafka.

func (*Broker) IsConnected

func (b *Broker) IsConnected() bool

IsConnected reports whether the broker has brokers configured. For Kafka, connections are per-operation, so this is a lightweight check.

func (*Broker) Metrics

func (b *Broker) Metrics() mq.Metrics

Metrics returns a snapshot of broker metrics.

func (*Broker) Producer

func (b *Broker) Producer(topic string, opts mq.PublishOptions) (mq.Producer, error)

Producer creates or returns a cached producer for the given topic. If topic is empty, the broker's default Topic is used.

func (*Broker) Unbind

func (b *Broker) Unbind(queue, exchange, routingKey string) error

Unbind is a no-op for Kafka.

type Config

type Config struct {
	// Brokers is the list of Kafka bootstrap servers.
	// Example: []string{"localhost:9092"}
	Brokers []string

	// Topic is the default topic used when a producer or consumer is
	// created without an explicit topic. May be empty if topics are
	// always passed explicitly to Producer/Consumer.
	Topic string

	// GroupID is the default consumer group ID. May be overridden per
	// consumer via ConsumeOptions.ConsumerTag.
	GroupID string

	// DialerTimeout is the timeout for establishing connections.
	// Default: 10s.
	DialerTimeout time.Duration

	// EnableTLS enables TLS for all connections.
	EnableTLS bool

	// TLSConfig is the TLS configuration. If EnableTLS is true and
	// TLSConfig is nil, a default (InsecureSkipVerify=false) config is
	// used.
	TLSConfig *tls.Config

	// CommitInterval is the auto-commit interval for consumers.
	// Default: 1s. Set to 0 to disable auto-commit (manual commit only).
	CommitInterval time.Duration
}

Config configures a Kafka broker.

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns a Config with sensible defaults for localhost.

type Consumer

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

Consumer implements mq.Consumer for Kafka using a kafka.Reader.

func (*Consumer) IsRunning

func (c *Consumer) IsRunning() bool

IsRunning reports whether the consumer is actively consuming.

func (*Consumer) Start

func (c *Consumer) Start(ctx context.Context) error

Start begins consuming from the Kafka topic. The consumer runs until ctx is cancelled or Stop is called.

func (*Consumer) Stop

func (c *Consumer) Stop(timeout time.Duration) error

Stop gracefully stops consuming, waiting for in-flight handlers to complete up to the given timeout.

func (*Consumer) String

func (c *Consumer) String() string

String returns a debug description of the consumer.

type Producer

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

Producer implements mq.Producer for Kafka using a kafka.Writer.

func (*Producer) Close

func (p *Producer) Close() error

Close releases producer resources.

func (*Producer) Publish

func (p *Producer) Publish(ctx context.Context, msg *mq.Message) error

Publish sends a message to the Kafka topic.