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(sarama.ConsumerGroupSession) error
• type Kafka
  • func (mq *Kafka) Close() error
  • func (mq *Kafka) Init(ctx context.Context) error
  • func (k *Kafka) Publish(ctx context.Context, target string, message query.Message) error
  • func (mq *Kafka) Subscribe(target string, message query.Response) error
  • func (mq *Kafka) UnSubscribe(target string) error

Types

type Config

type Config struct {
	URI           []string // addresses of available kafka brokers
	ConsumerGroup 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().

func (*Consumer) Setup

func (consumer *Consumer) Setup(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) Close

func (mq *Kafka) Close() error

func (*Kafka) Init

func (mq *Kafka) Init(ctx context.Context) error

func (*Kafka) Publish

func (k *Kafka) Publish(ctx context.Context, target string, message query.Message) error

func (*Kafka) Subscribe

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

func (*Kafka) UnSubscribe

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