kafkasubscriber

Documentation

Index

• Constants
• func DefaultSubscriberConfig() *sarama.Config
• func New(consumerOpt ConsumerOption, opts ...Option) easypubsub.Subscriber
• type ConsumerOption
  • func Consumer(brokers []string, config *sarama.Config) ConsumerOption
  • func ConsumerGroup(brokers []string, groupID string, config *sarama.Config) ConsumerOption
• type Option
  • func WithLogger(logger easypubsub.Logger) Option
  • func WithNackResend(maxAttempt uint, backoff backoffutils.BackoffFunc) Option
  • func WithReconnectBackoff(reconnectBackoff backoffutils.BackoffFunc) Option
  • func WithUnmarshalMsgFunc(unmarshalMsgFunc UnmarshalMsgFunc) Option
• type Subscriber
  • func (sub *Subscriber) Close() error
  • func (sub *Subscriber) String() string
  • func (sub *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *easypubsub.Message, <-chan error)
• type UnmarshalMsgFunc

Constants

const (
	NORMAL = 0
	CLOSED = 1
)

Functions

func DefaultSubscriberConfig

func DefaultSubscriberConfig() *sarama.Config

func New

func New(consumerOpt ConsumerOption, opts ...Option) easypubsub.Subscriber

Types

type ConsumerOption

type ConsumerOption func(o *consumerOptions)

func Consumer

func Consumer(brokers []string, config *sarama.Config) ConsumerOption

func ConsumerGroup

func ConsumerGroup(brokers []string, groupID string, config *sarama.Config) ConsumerOption

type Option

type Option func(o *options)

func WithLogger

func WithLogger(logger easypubsub.Logger) Option

func WithNackResend

func WithNackResend(maxAttempt uint, backoff backoffutils.BackoffFunc) Option

func WithReconnectBackoff

func WithReconnectBackoff(reconnectBackoff backoffutils.BackoffFunc) Option

func WithUnmarshalMsgFunc

func WithUnmarshalMsgFunc(unmarshalMsgFunc UnmarshalMsgFunc) Option

type Subscriber

type Subscriber struct {
	// contains filtered or unexported fields
}

func (*Subscriber) Close

func (sub *Subscriber) Close() error

func (*Subscriber) String

func (sub *Subscriber) String() string

func (*Subscriber) Subscribe

func (sub *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *easypubsub.Message, <-chan error)

type UnmarshalMsgFunc

type UnmarshalMsgFunc func(ctx context.Context, topic string, kafkaMsg *sarama.ConsumerMessage) (msg *easypubsub.Message, err error)