consumer

Documentation

Index

• Variables
• type Consumer
  • func New(reader *kafka.Reader, handler Handler, concurrency int, opts ...Option) *Consumer
  • func (c *Consumer) Close() error
  • func (c *Consumer) Healthcheck(ctx context.Context) error
  • func (c *Consumer) Run(ctx context.Context)
• type Delivery
  • func NewDelivery(donner Donner, reader *kafka.Reader, source *kafka.Message, ...) *Delivery
  • func (d *Delivery) Commit(ctx context.Context) error
  • func (d *Delivery) ConsumerGroupId() string
  • func (d *Delivery) Done()
  • func (d *Delivery) Source() *kafka.Message
• type Donner
• type Handler
• type HandlerFunc
  • func (f HandlerFunc) Handle(ctx context.Context, delivery *Delivery)
• type LogObserver
  • func NewLogObserver(ctx context.Context, logger log.Logger) LogObserver
  • func (l LogObserver) BeginConsuming()
  • func (l LogObserver) CloseDone()
  • func (l LogObserver) CloseStart()
  • func (l LogObserver) ConsumerError(err error)
• type Middleware
• type NoopObserver
  • func (n NoopObserver) BeginConsuming()
  • func (n NoopObserver) CloseDone()
  • func (n NoopObserver) CloseStart()
  • func (n NoopObserver) ConsumerError(err error)
• type Observer
• type Option
  • func WithMiddlewares(mws ...Middleware) Option
  • func WithObserver(observer Observer) Option

Variables

var (
	ErrDeliveryAlreadyHandled = errors.New("delivery already handled")
)

Types

type Consumer

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

func New

func New(reader *kafka.Reader, handler Handler, concurrency int, opts ...Option) *Consumer

func (*Consumer) Close

func (c *Consumer) Close() error

func (*Consumer) Healthcheck

func (c *Consumer) Healthcheck(ctx context.Context) error

func (*Consumer) Run

func (c *Consumer) Run(ctx context.Context)

type Delivery

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

func NewDelivery

func NewDelivery(donner Donner, reader *kafka.Reader, source *kafka.Message, consumerGroupId string) *Delivery

func (*Delivery) Commit

func (d *Delivery) Commit(ctx context.Context) error

func (*Delivery) ConsumerGroupId

func (d *Delivery) ConsumerGroupId() string

func (*Delivery) Done

func (d *Delivery) Done()

func (*Delivery) Source

func (d *Delivery) Source() *kafka.Message

type Donner

type Donner interface {
	Done()
}

type Handler

type Handler interface {
	Handle(ctx context.Context, delivery *Delivery)
}

type HandlerFunc

type HandlerFunc func(ctx context.Context, delivery *Delivery)

func (HandlerFunc) Handle

func (f HandlerFunc) Handle(ctx context.Context, delivery *Delivery)

type LogObserver

type LogObserver struct {
	NoopObserver
	// contains filtered or unexported fields
}

func NewLogObserver

func NewLogObserver(ctx context.Context, logger log.Logger) LogObserver

func (LogObserver) BeginConsuming

func (l LogObserver) BeginConsuming()

func (LogObserver) CloseDone

func (l LogObserver) CloseDone()

func (LogObserver) CloseStart

func (l LogObserver) CloseStart()

func (LogObserver) ConsumerError

func (l LogObserver) ConsumerError(err error)

type Middleware

type Middleware func(next Handler) Handler

type NoopObserver

type NoopObserver struct{}

func (NoopObserver) BeginConsuming

func (n NoopObserver) BeginConsuming()

func (NoopObserver) CloseDone

func (n NoopObserver) CloseDone()

func (NoopObserver) CloseStart

func (n NoopObserver) CloseStart()

func (NoopObserver) ConsumerError

func (n NoopObserver) ConsumerError(err error)

type Observer

type Observer interface {
	ConsumerError(err error)
	BeginConsuming()
	CloseStart()
	CloseDone()
}

type Option

type Option func(p *Consumer)

func WithMiddlewares

func WithMiddlewares(mws ...Middleware) Option

func WithObserver

func WithObserver(observer Observer) Option