internal

Dec 10, 2023

Changes in this version

+ const MessageErrHeaderKey

Nov 16, 2023

Changes in this version

+ const RetryAttemptHeaderKey

type MessageWrapper

+ RetryAttemptCount int

+ func (m *MessageWrapper) IncreaseRetryAttemptCount()

+ func (m *MessageWrapper) ResetRetryAttempt()

Aug 19, 2023

Changes in this version

+ const Name

+ type Collector struct

+ func NewCollector(cronsumerMetric *CronsumerMetric) *Collector

+ func (s *Collector) Collect(ch chan<- prometheus.Metric)

+ func (s *Collector) Describe(ch chan<- *prometheus.Desc)

+ type CronsumerMetric struct

+ TotalDiscardedMessagesCounter int64

+ TotalRetriedMessagesCounter int64

Aug 3, 2023

Changes in this version

type MessageWrapper

+ func (m *MessageWrapper) IsGteMaxRetryCount(maxRetry int) bool

Jun 15, 2023

Changes in this version

type Producer

+ ProduceBatch func(messages []kafka.Message) error

Jun 5, 2023

Changes in this version

type Producer

+ Close func()

Dec 30, 2022

Changes in this version

+ func FromHeaders(sh []segmentio.Header) []kafka.Header

+ func ToHeaders(h []kafka.Header) []segmentio.Header

type Producer

+ Produce func(message kafka.Message) error

Dec 5, 2022

Changes in this version

+ const MessageProduceTimeHeaderKey

+ const RetryHeaderKey

+ func Mechanism(sasl kafka.SASLConfig) sasl.Mechanism

+ func NewCronsumer(cfg *kafka.Config, fn kafka.ConsumeFn) kafka.Cronsumer

+ func NewTLSConfig(sasl kafka.SASLConfig) *tls.Config

+ type Consumer interface

+ ReadMessage func(ctx context.Context) (*MessageWrapper, error)

+ Stop func()

+ type MessageWrapper struct

+ ProduceTime int64

+ RetryCount int

+ func NewMessageWrapper(msg segmentio.Message) *MessageWrapper

+ func (m *MessageWrapper) GetHeaders() map[string][]byte

+ func (m *MessageWrapper) IncreaseRetryCount()

+ func (m *MessageWrapper) IsExceedMaxRetryCount(maxRetry int) bool

+ func (m *MessageWrapper) NewProduceTime()

+ func (m *MessageWrapper) RouteMessageToTopic(topic string)

+ func (m *MessageWrapper) To(increaseRetry bool) segmentio.Message

+ type Producer interface

+ ProduceWithRetryOption func(message MessageWrapper, increaseRetry bool) error