kafka

Documentation

Index

• func GroupReceive(consumerGroup sarama.ConsumerGroup, topics []string, ...) error
• func NewSyncConsumer(addresses []string, user string, password string) (sarama.Consumer, error)
• func NewSyncConsumerGroup(addresses []string, groupID, user string, password string) (sarama.ConsumerGroup, error)
• func NewSyncProducer(addresses []string, user string, password string) (sarama.SyncProducer, error)
• func Receive(consumer sarama.Consumer, topic string, ...) error
• func Send(msg *sarama.ProducerMessage, producer sarama.SyncProducer) error
• func SendB(b []byte, topic string, producer sarama.SyncProducer) error
• func SendT(t any, topic string, producer sarama.SyncProducer) error
• type ConsumerGroupHandler
  • func (*ConsumerGroupHandler) Cleanup(sarama.ConsumerGroupSession) error
  • func (handler *ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (*ConsumerGroupHandler) Setup(sarama.ConsumerGroupSession) error

Functions

func GroupReceive

func GroupReceive(consumerGroup sarama.ConsumerGroup, topics []string, handle func(message *sarama.ConsumerMessage) error) error

func NewSyncConsumer

func NewSyncConsumer(addresses []string, user string, password string) (sarama.Consumer, error)

func NewSyncConsumerGroup

func NewSyncConsumerGroup(addresses []string, groupID, user string, password string) (sarama.ConsumerGroup, error)

func NewSyncProducer

func NewSyncProducer(addresses []string, user string, password string) (sarama.SyncProducer, error)

func Receive

func Receive(consumer sarama.Consumer, topic string, handle func(msg *sarama.ConsumerMessage)) error

func Send

func Send(msg *sarama.ProducerMessage, producer sarama.SyncProducer) error

func SendB

func SendB(b []byte, topic string, producer sarama.SyncProducer) error

func SendT

func SendT(t any, topic string, producer sarama.SyncProducer) error

Types

type ConsumerGroupHandler

type ConsumerGroupHandler struct {
	//sarama.ConsumerGroupHandler
	Handle func(msg *sarama.ConsumerMessage) error
}

func (*ConsumerGroupHandler) Cleanup

func (*ConsumerGroupHandler) Cleanup(sarama.ConsumerGroupSession) error

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

func (*ConsumerGroupHandler) ConsumeClaim

func (handler *ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().

func (*ConsumerGroupHandler) Setup

func (*ConsumerGroupHandler) Setup(sarama.ConsumerGroupSession) error