tkafka

Documentation

Index

• Variables
• func BrokerAddress() []string
• func Config() wkafka.Config
• func Consume[T any](ctx context.Context, cfg ReadConfig, ...) error
• func TestClient(opts ...OptionTestClient) (*wkafka.Client, error)
• type Generate
  • func NewGenerate(c *wkafka.Client) *Generate
  • func (g *Generate) Cleanup() (kadm.DeleteTopicResponses, error)
  • func (g *Generate) CreateTopics(ctx context.Context, topics ...Topic) ([]kadm.CreateTopicResponse, error)
  • func (g *Generate) DeleteGroups(ctx context.Context, groups ...string) (kadm.DeleteGroupResponses, error)
  • func (g *Generate) DeleteTopics(ctx context.Context, topics ...string) (kadm.DeleteTopicResponses, error)
• type OptionTestClient
  • func WithConfig(cfg wkafka.Config) OptionTestClient
  • func WithContext(ctx context.Context) OptionTestClient
  • func WithWafkaOptions(opts ...wkafka.Option) OptionTestClient
• type ReadConfig
• type Topic

Variables

var (
	DefaultBrokerAddress = []string{"127.0.0.1:9092"}
)

Functions

func BrokerAddress

func BrokerAddress() []string

func Config

func Config() wkafka.Config

Config returns the default configuration for Kafka for testing.

func Consume

func Consume[T any](ctx context.Context, cfg ReadConfig, fn func(ctx context.Context, data T) error) error

Consume for testing creating client and start consuming messages to your function.

• Cancel the context to stop this function.

func TestClient

func TestClient(opts ...OptionTestClient) (*wkafka.Client, error)

Types

type Generate

type Generate struct {
	Topics map[string]struct{}
	// contains filtered or unexported fields
}

func NewGenerate

func NewGenerate(c *wkafka.Client) *Generate

func (*Generate) Cleanup

func (g *Generate) Cleanup() (kadm.DeleteTopicResponses, error)

func (*Generate) CreateTopics

func (g *Generate) CreateTopics(ctx context.Context, topics ...Topic) ([]kadm.CreateTopicResponse, error)

func (*Generate) DeleteGroups

func (g *Generate) DeleteGroups(ctx context.Context, groups ...string) (kadm.DeleteGroupResponses, error)

func (*Generate) DeleteTopics

func (g *Generate) DeleteTopics(ctx context.Context, topics ...string) (kadm.DeleteTopicResponses, error)

type OptionTestClient

type OptionTestClient func(*optionTestClient)

func WithConfig

func WithConfig(cfg wkafka.Config) OptionTestClient

func WithContext

func WithContext(ctx context.Context) OptionTestClient

func WithWafkaOptions

func WithWafkaOptions(opts ...wkafka.Option) OptionTestClient

type ReadConfig

type ReadConfig struct {
	Name string

	KafkaConfig   wkafka.Config
	KafkaConsumer wkafka.ConsumerConfig
}

type Topic

type Topic struct {
	Name string

	// Partitions defaults to -1, which means the broker default.
	Partitions int32
	// ReplicationFactor defaults to -1, which means the broker default.
	ReplicationFactor int16
}