kafka

Documentation

Index

• 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(topic string, consumer sarama.Consumer, do func(msg *sarama.ConsumerMessage)) error
• func ReceiveGroup(consumerGroup sarama.ConsumerGroup, topics []string, ...) error
• func Send(b []byte, topic string, producer sarama.SyncProducer) error
• func SendT(t any, topic string, producer sarama.SyncProducer) error
• func SendTT(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 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(topic string, consumer sarama.Consumer, do func(msg *sarama.ConsumerMessage)) error

func ReceiveGroup

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

func Send

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

func SendT

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

func SendTT

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

Types

type ConsumerGroupHandler

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

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