consumer

Documentation

Index

• type Client
  • func NewConsumer(cfg Config, opts ...Option) Client
• type Config
• type HandlerFunc
• type Option
  • func Handler(handler HandlerFunc) Option
  • func IsDebug(isDebug bool) Option
  • func RebalanceCb(rebalanceCb kafka.RebalanceCb) Option
• type Options

Types

type Client

type Client interface {
	Start(ctx context.Context) error
	Stop() error
}

func NewConsumer

func NewConsumer(cfg Config, opts ...Option) Client

type Config

type Config struct {
	Topics           []string                     `json:"topics"`
	GroupId          string                       `json:"group.id"`
	BootstrapServers string                       `json:"bootstrap.servers"`
	SecurityProtocol string                       `json:"security.protocol"`
	SaslMechanism    string                       `json:"sasl.mechanism"`
	SaslUsername     string                       `json:"sasl.username"`
	SaslPassword     string                       `json:"sasl.password"`
	SslCaLocation    string                       `json:"ssl.ca.location"`
	ConfigMap        map[string]kafka.ConfigValue `json:"config.map"`
}

type HandlerFunc

type HandlerFunc func(message *kafka.Message)

type Option

type Option func(opts *Options)

func Handler

func Handler(handler HandlerFunc) Option

func IsDebug

func IsDebug(isDebug bool) Option

func RebalanceCb

func RebalanceCb(rebalanceCb kafka.RebalanceCb) Option

type Options

type Options struct {
	Topics      []string
	Handler     HandlerFunc
	RebalanceCb kafka.RebalanceCb
	IsDebug     bool
}