kafka

Documentation

Overview

Package kafka provides a client for interacting with Apache Kafka message queues.This package facilitates interaction with Apache Kafka, allowing publishing and subscribing to topics, managing consumer groups, and handling messages.

Package kafka is a generated GoMock package.

Package kafka is a generated GoMock package.

Index

• Constants
• Variables
• func New(conf Config, logger pubsub.Logger, metrics Metrics) *kafkaClient
• type Config
• type Connection
• type Metrics
• type MockConnection
  • func NewMockConnection(ctrl *gomock.Controller) *MockConnection
  • func (m *MockConnection) Close() error
  • func (m *MockConnection) Controller() (kafka.Broker, error)
  • func (m *MockConnection) CreateTopics(topics ...kafka.TopicConfig) error
  • func (m *MockConnection) DeleteTopics(topics ...string) error
  • func (m *MockConnection) EXPECT() *MockConnectionMockRecorder
• type MockConnectionMockRecorder
  • func (mr *MockConnectionMockRecorder) Close() *gomock.Call
  • func (mr *MockConnectionMockRecorder) Controller() *gomock.Call
  • func (mr *MockConnectionMockRecorder) CreateTopics(topics ...any) *gomock.Call
  • func (mr *MockConnectionMockRecorder) DeleteTopics(topics ...any) *gomock.Call
• type MockMetrics
  • func NewMockMetrics(ctrl *gomock.Controller) *MockMetrics
  • func (m *MockMetrics) EXPECT() *MockMetricsMockRecorder
  • func (m *MockMetrics) IncrementCounter(ctx context.Context, name string, labels ...string)
• type MockMetricsMockRecorder
  • func (mr *MockMetricsMockRecorder) IncrementCounter(ctx, name any, labels ...any) *gomock.Call
• type MockReader
  • func NewMockReader(ctrl *gomock.Controller) *MockReader
  • func (m *MockReader) Close() error
  • func (m *MockReader) CommitMessages(ctx context.Context, msgs ...kafka.Message) error
  • func (m *MockReader) EXPECT() *MockReaderMockRecorder
  • func (m *MockReader) ReadMessage(ctx context.Context) (kafka.Message, error)
  • func (m *MockReader) Stats() kafka.ReaderStats
• type MockReaderMockRecorder
  • func (mr *MockReaderMockRecorder) Close() *gomock.Call
  • func (mr *MockReaderMockRecorder) CommitMessages(ctx any, msgs ...any) *gomock.Call
  • func (mr *MockReaderMockRecorder) ReadMessage(ctx any) *gomock.Call
  • func (mr *MockReaderMockRecorder) Stats() *gomock.Call
• type MockWriter
  • func NewMockWriter(ctrl *gomock.Controller) *MockWriter
  • func (m *MockWriter) Close() error
  • func (m *MockWriter) EXPECT() *MockWriterMockRecorder
  • func (m *MockWriter) Stats() kafka.WriterStats
  • func (m *MockWriter) WriteMessages(ctx context.Context, msg ...kafka.Message) error
• type MockWriterMockRecorder
  • func (mr *MockWriterMockRecorder) Close() *gomock.Call
  • func (mr *MockWriterMockRecorder) Stats() *gomock.Call
  • func (mr *MockWriterMockRecorder) WriteMessages(ctx any, msg ...any) *gomock.Call
• type Reader
• type Writer

Constants

const (
	DefaultBatchSize    = 100
	DefaultBatchBytes   = 1048576
	DefaultBatchTimeout = 1000
)

Variables

var (
	ErrConsumerGroupNotProvided = errors.New("consumer group id not provided")
)

Functions

func New

func New(conf Config, logger pubsub.Logger, metrics Metrics) *kafkaClient

Types

type Config

type Config struct {
	Broker          string
	Partition       int
	ConsumerGroupID string
	OffSet          int
	BatchSize       int
	BatchBytes      int
	BatchTimeout    int
}

type Connection

type Connection interface {
	Controller() (broker kafka.Broker, err error)
	CreateTopics(topics ...kafka.TopicConfig) error
	DeleteTopics(topics ...string) error
	Close() error
}

type Metrics

type Metrics interface {
	IncrementCounter(ctx context.Context, name string, labels ...string)
}

type MockConnection

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

MockConnection is a mock of Connection interface.

func NewMockConnection

func NewMockConnection(ctrl *gomock.Controller) *MockConnection

NewMockConnection creates a new mock instance.

func (*MockConnection) Close

func (m *MockConnection) Close() error

Close mocks base method.

func (*MockConnection) Controller

func (m *MockConnection) Controller() (kafka.Broker, error)

Controller mocks base method.

func (*MockConnection) CreateTopics

func (m *MockConnection) CreateTopics(topics ...kafka.TopicConfig) error

CreateTopics mocks base method.

func (*MockConnection) DeleteTopics

func (m *MockConnection) DeleteTopics(topics ...string) error

DeleteTopics mocks base method.

func (*MockConnection) EXPECT

func (m *MockConnection) EXPECT() *MockConnectionMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

type MockConnectionMockRecorder

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

MockConnectionMockRecorder is the mock recorder for MockConnection.

func (*MockConnectionMockRecorder) Close

func (mr *MockConnectionMockRecorder) Close() *gomock.Call

Close indicates an expected call of Close.

func (*MockConnectionMockRecorder) Controller

func (mr *MockConnectionMockRecorder) Controller() *gomock.Call

Controller indicates an expected call of Controller.

func (*MockConnectionMockRecorder) CreateTopics

func (mr *MockConnectionMockRecorder) CreateTopics(topics ...any) *gomock.Call

CreateTopics indicates an expected call of CreateTopics.

func (*MockConnectionMockRecorder) DeleteTopics

func (mr *MockConnectionMockRecorder) DeleteTopics(topics ...any) *gomock.Call

DeleteTopics indicates an expected call of DeleteTopics.

type MockMetrics

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

MockMetrics is a mock of Metrics interface.

func NewMockMetrics

func NewMockMetrics(ctrl *gomock.Controller) *MockMetrics

NewMockMetrics creates a new mock instance.

func (*MockMetrics) EXPECT

func (m *MockMetrics) EXPECT() *MockMetricsMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

func (*MockMetrics) IncrementCounter

func (m *MockMetrics) IncrementCounter(ctx context.Context, name string, labels ...string)

IncrementCounter mocks base method.

type MockMetricsMockRecorder

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

MockMetricsMockRecorder is the mock recorder for MockMetrics.

func (*MockMetricsMockRecorder) IncrementCounter

func (mr *MockMetricsMockRecorder) IncrementCounter(ctx, name any, labels ...any) *gomock.Call

IncrementCounter indicates an expected call of IncrementCounter.

type MockReader

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

MockReader is a mock of Reader interface.

func NewMockReader

func NewMockReader(ctrl *gomock.Controller) *MockReader

NewMockReader creates a new mock instance.

func (*MockReader) Close

func (m *MockReader) Close() error

Close mocks base method.

func (*MockReader) CommitMessages

func (m *MockReader) CommitMessages(ctx context.Context, msgs ...kafka.Message) error

CommitMessages mocks base method.

func (*MockReader) EXPECT

func (m *MockReader) EXPECT() *MockReaderMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

func (*MockReader) ReadMessage

func (m *MockReader) ReadMessage(ctx context.Context) (kafka.Message, error)

ReadMessage mocks base method.

func (*MockReader) Stats

func (m *MockReader) Stats() kafka.ReaderStats

Stats mocks base method.

type MockReaderMockRecorder

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

MockReaderMockRecorder is the mock recorder for MockReader.

func (*MockReaderMockRecorder) Close

func (mr *MockReaderMockRecorder) Close() *gomock.Call

Close indicates an expected call of Close.

func (*MockReaderMockRecorder) CommitMessages

func (mr *MockReaderMockRecorder) CommitMessages(ctx any, msgs ...any) *gomock.Call

CommitMessages indicates an expected call of CommitMessages.

func (*MockReaderMockRecorder) ReadMessage

func (mr *MockReaderMockRecorder) ReadMessage(ctx any) *gomock.Call

ReadMessage indicates an expected call of ReadMessage.

func (*MockReaderMockRecorder) Stats

func (mr *MockReaderMockRecorder) Stats() *gomock.Call

Stats indicates an expected call of Stats.

type MockWriter

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

MockWriter is a mock of Writer interface.

func NewMockWriter

func NewMockWriter(ctrl *gomock.Controller) *MockWriter

NewMockWriter creates a new mock instance.

func (*MockWriter) Close

func (m *MockWriter) Close() error

Close mocks base method.

func (*MockWriter) EXPECT

func (m *MockWriter) EXPECT() *MockWriterMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

func (*MockWriter) Stats

func (m *MockWriter) Stats() kafka.WriterStats

Stats mocks base method.

func (*MockWriter) WriteMessages

func (m *MockWriter) WriteMessages(ctx context.Context, msg ...kafka.Message) error

WriteMessages mocks base method.

type MockWriterMockRecorder

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

MockWriterMockRecorder is the mock recorder for MockWriter.

func (*MockWriterMockRecorder) Close

func (mr *MockWriterMockRecorder) Close() *gomock.Call

Close indicates an expected call of Close.

func (*MockWriterMockRecorder) Stats

func (mr *MockWriterMockRecorder) Stats() *gomock.Call

Stats indicates an expected call of Stats.

func (*MockWriterMockRecorder) WriteMessages

func (mr *MockWriterMockRecorder) WriteMessages(ctx any, msg ...any) *gomock.Call

WriteMessages indicates an expected call of WriteMessages.

type Reader

type Reader interface {
	ReadMessage(ctx context.Context) (kafka.Message, error)
	CommitMessages(ctx context.Context, msgs ...kafka.Message) error
	Stats() kafka.ReaderStats
	Close() error
}

type Writer

type Writer interface {
	WriteMessages(ctx context.Context, msg ...kafka.Message) error
	Close() error
	Stats() kafka.WriterStats
}