kafka

Documentation

Index

• type Config
• type Consumer
  • func (consumer *Consumer) Cleanup(sarama.ConsumerGroupSession) error
  • func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (consumer *Consumer) Setup(s sarama.ConsumerGroupSession) error
• type Kafka
  • func (mq *Kafka) Init(ctx context.Context, log logger.Logger) error
  • func (mq *Kafka) Publish(_ context.Context, target string, routingKey []byte, payload []byte) error
  • func (mq *Kafka) Subscribe(ctx context.Context, target string, message query.Response) error
  • func (mq *Kafka) UnSubscribe(target string) error

Types

type Config

type Config struct {
	ConsumerGroup string
	URI           []string
}

type Consumer

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

Consumer represents a Sarama consumer group consumer

func (*Consumer) Cleanup

func (consumer *Consumer) Cleanup(sarama.ConsumerGroupSession) error

Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited

func (*Consumer) ConsumeClaim

func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages(). Once the Messages() channel is closed, the Handler must finish its processing loop and exit.

func (*Consumer) Setup

func (consumer *Consumer) Setup(s sarama.ConsumerGroupSession) error

Setup is run at the beginning of a new session, before ConsumeClaim

type Kafka

type Kafka struct {
	*Config
	// contains filtered or unexported fields
}

func (*Kafka) Init

func (mq *Kafka) Init(ctx context.Context, log logger.Logger) error

func (*Kafka) Publish

func (mq *Kafka) Publish(_ context.Context, target string, routingKey []byte, payload []byte) error

func (*Kafka) Subscribe

func (mq *Kafka) Subscribe(ctx context.Context, target string, message query.Response) error

Subscribe - subscribe to message

func (*Kafka) UnSubscribe

func (mq *Kafka) UnSubscribe(target string) error