saramax

Documentation

Index

• type BatchHandler
  • func NewBatchHandler[T interface{}](l logger.Logger, fn func(msgs []*sarama.ConsumerMessage, t []T) error) *BatchHandler[T]
  • func (h *BatchHandler[T]) Cleanup(session sarama.ConsumerGroupSession) error
  • func (h *BatchHandler[T]) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (h *BatchHandler[T]) Setup(session sarama.ConsumerGroupSession) error
• type Consumer
• type Handler
  • func NewHandler[T interface{}](l logger.Logger, fn func(msg *sarama.ConsumerMessage, t T) error) *Handler[T]
  • func (h *Handler[T]) Cleanup(session sarama.ConsumerGroupSession) error
  • func (h *Handler[T]) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (h *Handler[T]) Setup(session sarama.ConsumerGroupSession) error

Types

type BatchHandler

type BatchHandler[T interface{}] struct {
	// contains filtered or unexported fields
}

func NewBatchHandler

func NewBatchHandler[T interface{}](l logger.Logger,
	fn func(msgs []*sarama.ConsumerMessage, t []T) error) *BatchHandler[T]

func (*BatchHandler[T]) Cleanup

func (h *BatchHandler[T]) Cleanup(session sarama.ConsumerGroupSession) error

func (*BatchHandler[T]) ConsumeClaim

func (h *BatchHandler[T]) ConsumeClaim(session sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim) error

func (*BatchHandler[T]) Setup

func (h *BatchHandler[T]) Setup(session sarama.ConsumerGroupSession) error

type Consumer

type Consumer interface {
	Start() error
}

type Handler

type Handler[T interface{}] struct {
	// contains filtered or unexported fields
}

func NewHandler

func NewHandler[T interface{}](l logger.Logger,
	fn func(msg *sarama.ConsumerMessage, t T) error) *Handler[T]

func (*Handler[T]) Cleanup

func (h *Handler[T]) Cleanup(session sarama.ConsumerGroupSession) error

func (*Handler[T]) ConsumeClaim

func (h *Handler[T]) ConsumeClaim(session sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim) error

ConsumeClaim 可以考虑在这个封装里面提供统一的重试机制

func (*Handler[T]) Setup

func (h *Handler[T]) Setup(session sarama.ConsumerGroupSession) error