sikafka

Sep 11, 2024

Changes in this version

type AsyncProducer

+ func RetryableAsyncProducer(brokers []string, version, topic string) (*AsyncProducer, error)

Sep 10, 2024

Changes in this version

type SyncProducer

+ func RetryableSyncProducer(brokers []string, version, topic string) (*SyncProducer, error)

Sep 10, 2024

Changes in this version

+ type SyncProducerOption interface

+ type SyncProducerOptionFunc func(o *SyncProducer) error

+ func WithSyncProducerOptionRetyMax(retryMax uint16) SyncProducerOptionFunc

May 19, 2024

Changes in this version

+ func AsyncProducerWithConfig(config *sarama.Config, brokers []string) (sarama.AsyncProducer, error)

+ func ConsumerGroupWithConfig(config *sarama.Config, brokers []string, group string) (sarama.ConsumerGroup, error)

+ func DefaultAsyncProducer(brokers []string) (sarama.AsyncProducer, error)

+ func DefaultConsumerGroup(brokers []string, group string, version string, assignor string, oldest bool) (sarama.ConsumerGroup, error)

+ func DefaultSyncProducer(brokers []string) (sarama.SyncProducer, error)

+ func SyncProducerWithConfig(config *sarama.Config, brokers []string) (sarama.SyncProducer, error)

+ type AsyncProducer struct

+ func NewAsyncProducer(producer sarama.AsyncProducer, topic string) *AsyncProducer

+ func (ap *AsyncProducer) Produce(key []byte, value []byte) (partition int32, offset int64, err error)

+ func (ap *AsyncProducer) ProduceWithMessage(msg *sarama.ProducerMessage) (partition int32, offset int64, err error)

+ func (ap *AsyncProducer) ProduceWithTopic(topic string, key []byte, value []byte) (partition int32, offset int64, err error)

+ type CgConsumer struct

+ func NewCgConsumer(msgHandler MessageHandler) *CgConsumer

+ func (c *CgConsumer) CloseReady()

+ func (c *CgConsumer) MakeReady()

+ func (c *CgConsumer) WaitReady()

+ func (consumer *CgConsumer) Cleanup(sarama.ConsumerGroupSession) error

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

+ func (consumer *CgConsumer) Setup(sarama.ConsumerGroupSession) error

+ type Consumer interface

+ Cleanup func(sarama.ConsumerGroupSession) error

+ CloseReady func()

+ ConsumeClaim func(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

+ MakeReady func()

+ Setup func(sarama.ConsumerGroupSession) error

+ WaitReady func()

+ type ConsumerGroup struct

+ func NewConsumerGroup(cg sarama.ConsumerGroup, consumer Consumer, topics []string) *ConsumerGroup

+ func (cg *ConsumerGroup) Finish() error

+ func (cg *ConsumerGroup) Start() error

+ func (cg *ConsumerGroup) StartWith(loaded chan bool) error

+ func (cg *ConsumerGroup) Stop() error

+ func (cg *ConsumerGroup) Toggle()

+ type MessageHandler interface

+ Handle func(msg *sarama.ConsumerMessage) error

+ type SyncProducer struct

+ func NewSyncProducer(producer sarama.SyncProducer, topic string) *SyncProducer

+ func (sp *SyncProducer) Produce(key []byte, value []byte) (partition int32, offset int64, err error)

+ func (sp *SyncProducer) ProduceWithMessage(msg *sarama.ProducerMessage) (partition int32, offset int64, err error)

+ func (sp *SyncProducer) ProduceWithTopic(topic string, key []byte, value []byte) (partition int32, offset int64, err error)