kafka

Documentation

Index

• type Kafka
  • func (k *Kafka) Check() error
  • func (k *Kafka) Close()
  • func (k *Kafka) CreateTopic(topic string) error
  • func (k *Kafka) GetAdminClient() (sarama.ClusterAdmin, error)
  • func (k *Kafka) GetConsumer() (sarama.Consumer, error)
  • func (k *Kafka) GetConsumerGroup(id string) (sarama.ConsumerGroup, error)
  • func (k *Kafka) GetProducer() (sarama.AsyncProducer, error)
  • func (k *Kafka) Init(kafkaConfigUrl string)
  • func (k *Kafka) MessageListener(groupId, topic string, listener func(msg string) error) error
  • func (k *Kafka) Send(topic, data string) error
  • func (k *Kafka) SendMsgs(topic string, data []string) error
• type MsgHandler
  • func (MsgHandler) Cleanup(_ sarama.ConsumerGroupSession) error
  • func (h MsgHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (MsgHandler) Setup(_ sarama.ConsumerGroupSession) error

Types

type Kafka

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

func (*Kafka) Check

func (k *Kafka) Check() error

func (*Kafka) Close

func (k *Kafka) Close()

func (*Kafka) CreateTopic

func (k *Kafka) CreateTopic(topic string) error

func (*Kafka) GetAdminClient

func (k *Kafka) GetAdminClient() (sarama.ClusterAdmin, error)

func (*Kafka) GetConsumer

func (k *Kafka) GetConsumer() (sarama.Consumer, error)

func (*Kafka) GetConsumerGroup

func (k *Kafka) GetConsumerGroup(id string) (sarama.ConsumerGroup, error)

func (*Kafka) GetProducer

func (k *Kafka) GetProducer() (sarama.AsyncProducer, error)

func (*Kafka) Init

func (k *Kafka) Init(kafkaConfigUrl string)

func (*Kafka) MessageListener

func (k *Kafka) MessageListener(groupId, topic string, listener func(msg string) error) error

func (*Kafka) Send

func (k *Kafka) Send(topic, data string) error

func (*Kafka) SendMsgs

func (k *Kafka) SendMsgs(topic string, data []string) error

type MsgHandler

type MsgHandler struct {
	Handle func(msg string) error
}

func (MsgHandler) Cleanup

func (MsgHandler) Cleanup(_ sarama.ConsumerGroupSession) error

func (MsgHandler) ConsumeClaim

func (h MsgHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

func (MsgHandler) Setup

func (MsgHandler) Setup(_ sarama.ConsumerGroupSession) error