kafka

Documentation

Index

• type Consumer
  • func NewConsumer(bootStrapServer string, topic string, consumerGroup string, logger log.Logger) *Consumer
  • func (c *Consumer) Close() error
  • func (c *Consumer) Consume(ctx context.Context) (*kafka.Message, error)
  • func (c *Consumer) SetLogger(log log.Logger)
  • func (c *Consumer) Validate() (*Consumer, error)
• type Producer
  • func NewProducer(bootStrapServer string, topic string, logger log.Logger) *Producer
  • func (p *Producer) Close() error
  • func (p *Producer) Produce(ctx context.Context, message string) error
  • func (p *Producer) SetLogger(log log.Logger)

Types

type Consumer

type Consumer struct {
	BootstrapServer string `mapstructure:"bootstrap_server"`
	Topic           string `mapstructure:"topic"`
	ConsumerGroup   string `mapstructure:"consumer_group"`
	// contains filtered or unexported fields
}

func NewConsumer

func NewConsumer(bootStrapServer string, topic string, consumerGroup string, logger log.Logger) *Consumer

func (*Consumer) Close

func (c *Consumer) Close() error

func (*Consumer) Consume

func (c *Consumer) Consume(ctx context.Context) (*kafka.Message, error)

func (*Consumer) SetLogger

func (c *Consumer) SetLogger(log log.Logger)

func (*Consumer) Validate

func (c *Consumer) Validate() (*Consumer, error)

type Producer

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

func NewProducer

func NewProducer(bootStrapServer string, topic string, logger log.Logger) *Producer

func (*Producer) Close

func (p *Producer) Close() error

func (*Producer) Produce

func (p *Producer) Produce(ctx context.Context, message string) error

func (*Producer) SetLogger

func (p *Producer) SetLogger(log log.Logger)