kafka

Dec 13, 2019

Changes in this version

+ type CommitFilterOptFunc func(*commitCheckOption)

+ func WithCommitFilterCheckChanSize(size int) CommitFilterOptFunc

+ func WithCommitFilterCheckInterval(interval time.Duration) CommitFilterOptFunc

+ func WithCommitFilterCheckNum(num int) CommitFilterOptFunc

type KafkaCli

+ func NewKafkaCliWithGroupID(ctx context.Context, cfg *KafkaCliCfg, opts ...CommitFilterOptFunc) (*KafkaCli, error)

Apr 17, 2019

Changes in this version

+ type CommitFilter struct

+ func NewCommitFilter(cfg *CommitFilterCfg) *CommitFilter

+ func (f *CommitFilter) GetAfterChan() chan *KafkaMsg

+ func (f *CommitFilter) GetBeforeChan() chan *KafkaMsg

+ type CommitFilterCfg struct

+ IntervalDuration time.Duration

+ IntervalNum int

+ KMsgPool *sync.Pool

+ type KafkaCli struct

+ func NewKafkaCliWithGroupId(cfg *KafkaCliCfg) (*KafkaCli, error)

+ func (k *KafkaCli) Close()

+ func (k *KafkaCli) CommitWithMsg(kmsg *KafkaMsg)

+ func (k *KafkaCli) ListenNotifications()

+ func (k *KafkaCli) Messages() <-chan *KafkaMsg

+ type KafkaCliCfg struct

+ Brokers []string

+ Groupid string

+ IntervalDuration time.Duration

+ IntervalNum int

+ KMsgPool *sync.Pool

+ Topics []string

+ type KafkaMsg struct

+ Message []byte

+ Offset int64

+ Partition int32

+ Timestamp time.Time

+ Topic string