kafka

Documentation

Index

• func RegisterConsumer(ctx context.Context, handler MessageHandler, kafkaBrokers []string, ...) error
• type Consumer
  • func NewConsumer(_ context.Context, brokers []string, topicName string) (*Consumer, error)
  • func (c *Consumer) Close() error
  • func (c *Consumer) Consume() <-chan *sarama.ConsumerMessage
• type MessageHandler
• type Producer
  • func NewProducer(ctx context.Context, kafkaBrokers []string) (*Producer, error)
  • func (p *Producer) Close() error
  • func (p *Producer) ProduceMessage(topic string, messageBytes []byte)

Functions

func RegisterConsumer

func RegisterConsumer(
	ctx context.Context,
	handler MessageHandler,
	kafkaBrokers []string,
	topic string,
) error

Types

type Consumer

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

func NewConsumer

func NewConsumer(_ context.Context, brokers []string, topicName string) (*Consumer, error)

func (*Consumer) Close

func (c *Consumer) Close() error

func (*Consumer) Consume

func (c *Consumer) Consume() <-chan *sarama.ConsumerMessage

type MessageHandler

type MessageHandler interface {
	Consume(ctx context.Context, msg []byte) error
}

type Producer

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

func NewProducer

func NewProducer(ctx context.Context, kafkaBrokers []string) (*Producer, error)

func (*Producer) Close

func (p *Producer) Close() error

func (*Producer) ProduceMessage

func (p *Producer) ProduceMessage(topic string, messageBytes []byte)