kafkatest

Documentation

Index

• func WithMockedBinder() test.Options
• type MessageMocker
• type MessageRecord
• type MessageRecorder
• type MockedBinder
  • func (b *MockedBinder) AllRecords() (ret []*MessageRecord)
  • func (b *MockedBinder) Consume(topic string, group string, _ ...kafka.ConsumerOptions) (kafka.GroupConsumer, error)
  • func (b *MockedBinder) ListTopics() []string
  • func (b *MockedBinder) Mock(ctx context.Context, topic string, msg *kafka.Message) error
  • func (b *MockedBinder) MockWithGroup(ctx context.Context, topic, group string, msg *kafka.Message) error
  • func (b *MockedBinder) Produce(topic string, _ ...kafka.ProducerOptions) (kafka.Producer, error)
  • func (b *MockedBinder) Record(record *MessageRecord)
  • func (b *MockedBinder) Records(topic string) (ret []*MessageRecord)
  • func (b *MockedBinder) Reset()
  • func (b *MockedBinder) Subscribe(topic string, _ ...kafka.ConsumerOptions) (kafka.Subscriber, error)
• type MockedConsumer
  • func NewMockedConsumer(topic, group string) *MockedConsumer
  • func (c *MockedConsumer) Group() string
  • func (c *MockedConsumer) Topic() string
• type MockedProducer
  • func NewMockedProducer(topic string, recorder messageRecorder) *MockedProducer
  • func (p *MockedProducer) ReadyCh() <-chan struct{}
  • func (p *MockedProducer) Send(_ context.Context, message interface{}, _ ...kafka.MessageOptions) error
  • func (p *MockedProducer) Topic() string
• type MockedSubscriber
  • func NewMockedSubscriber(topic string) *MockedSubscriber
  • func (s *MockedSubscriber) Partitions() []int32
  • func (s *MockedSubscriber) Topic() string

Functions

func WithMockedBinder

func WithMockedBinder() test.Options

WithMockedBinder returns a test.Options that provides mocked kafka.Binder and a MessageRecorder. Tests can wire the MessageRecorder and verify invocation of kafka.Producer Note: The main purpose of this test configuration is to fulfill dependency injection and validate kafka.Producer is

invoked as expected. It doesn't validate/invoke any message options such as ValueEncoder or Key, nor does it
respect any binding configuration

Types

type MessageMocker

type MessageMocker interface {
	Mock(ctx context.Context, topic string, msg *kafka.Message) error
	MockWithGroup(ctx context.Context, topic, group string, msg *kafka.Message) error
}

MessageMocker interface for mocking incoming messages.

type MessageRecord

type MessageRecord struct {
	Topic   string
	Payload interface{}
}

type MessageRecorder

type MessageRecorder interface {
	Reset()
	Records(topic string) []*MessageRecord
	AllRecords() []*MessageRecord
}

MessageRecorder interface for retrieve messages produced by MockedProducer

type MockedBinder

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

MockedBinder implements kafka.Binder and messageRecorder

func (*MockedBinder) AllRecords

func (b *MockedBinder) AllRecords() (ret []*MessageRecord)

func (*MockedBinder) Consume

func (b *MockedBinder) Consume(topic string, group string, _ ...kafka.ConsumerOptions) (kafka.GroupConsumer, error)

func (*MockedBinder) ListTopics

func (b *MockedBinder) ListTopics() []string

func (*MockedBinder) Mock

func (b *MockedBinder) Mock(ctx context.Context, topic string, msg *kafka.Message) error

func (*MockedBinder) MockWithGroup

func (b *MockedBinder) MockWithGroup(ctx context.Context, topic, group string, msg *kafka.Message) error

func (*MockedBinder) Produce

func (b *MockedBinder) Produce(topic string, _ ...kafka.ProducerOptions) (kafka.Producer, error)

func (*MockedBinder) Record

func (b *MockedBinder) Record(record *MessageRecord)

func (*MockedBinder) Records

func (b *MockedBinder) Records(topic string) (ret []*MessageRecord)

func (*MockedBinder) Reset

func (b *MockedBinder) Reset()

func (*MockedBinder) Subscribe

func (b *MockedBinder) Subscribe(topic string, _ ...kafka.ConsumerOptions) (kafka.Subscriber, error)

type MockedConsumer

type MockedConsumer struct {
	kafka.Dispatcher
	T string
	G string
}

func NewMockedConsumer

func NewMockedConsumer(topic, group string) *MockedConsumer

func (*MockedConsumer) Group

func (c *MockedConsumer) Group() string

func (*MockedConsumer) Topic

func (c *MockedConsumer) Topic() string

type MockedProducer

type MockedProducer struct {
	T        string
	Recorder messageRecorder
}

func NewMockedProducer

func NewMockedProducer(topic string, recorder messageRecorder) *MockedProducer

func (*MockedProducer) ReadyCh

func (p *MockedProducer) ReadyCh() <-chan struct{}

func (*MockedProducer) Send

func (p *MockedProducer) Send(_ context.Context, message interface{}, _ ...kafka.MessageOptions) error

func (*MockedProducer) Topic

func (p *MockedProducer) Topic() string

type MockedSubscriber

type MockedSubscriber struct {
	kafka.Dispatcher
	T     string
	Parts []int32
}

func NewMockedSubscriber

func NewMockedSubscriber(topic string) *MockedSubscriber

func (*MockedSubscriber) Partitions

func (s *MockedSubscriber) Partitions() []int32

func (*MockedSubscriber) Topic

func (s *MockedSubscriber) Topic() string